perf(reverse): add retained GDB and objdump summaries (#5614)

Add opt-in retained GDB and objdump summaries using the shared analysis-view layer. Preserve default complete Evidence, ordered records, provenance and completion facts.

Validate real Bash disassembly, whole-client context costs, CLI publication and cleanup; retain the original measured cohorts and explicit coverage limitations.

Closes #5584.

experiments/analysis-efficiency/command-observation-costs.py

@@ -0,0 +1,32 @@
+"""Count all actual client JSON inputs and preserve bounded family resource receipts."""
+from pathlib import Path
+import argparse, json, hashlib
+import tiktoken
+parser=argparse.ArgumentParser(description=__doc__)
+parser.add_argument('--root',type=Path,required=True)
+parser.add_argument('--output',type=Path,required=True)
+args=parser.parse_args();lab=args.root;enc=tiktoken.get_encoding('cl100k_base')
+compact=lambda value:json.dumps(value,ensure_ascii=False,separators=(',',':'),allow_nan=False)
+count=lambda value:len(enc.encode(compact(value),disallowed_special=()))
+def resources(root):
+ scope=json.loads(Path(str(root)+'.log').read_text().splitlines()[-1]);o=scope['payload_resource_observation']
+ assert scope['abort'] is None and scope['payload_resource_observation_verified'] and o['payload_exit_code']==0 and o['payload_cleanup_verified']
+ for event in o['resource_events'].values():assert not any(event.values())
+ return {'wall_seconds':scope['elapsed_seconds'],'cpu_seconds':o['cpu_stat']['usage_usec']/1e6,'family_peak_mib':o['memory_peak_bytes']/1048576,'payload_cleanup_verified':True,'final_whole_scope_counters_verified':scope['resource_counters_verified']}
+wire=[];sdk=[]
+for root in sorted(lab.glob('final-wire-*')):
+ if not root.is_dir():continue
+ report=json.loads((root/'report.json').read_text());requests=json.loads((root/'model-requests.json').read_text());assert report['status'] in ['passed','expected-incomplete-task'] and not report['cleanup_failures'] and report['appserver_exit_code']==0 and report['owned_scope_drain']['remaining_processes']==[]
+ rows=[]
+ for index,captured in enumerate(requests):
+  body=captured['body'];namespace=[n for n in body.get('tools',[]) if n.get('type')=='namespace' and n.get('name')=='mcp__rea_audit'];compaction=not namespace
+  if compaction:assert 'CONTEXT CHECKPOINT COMPACTION' in compact(body['input'][-1])
+  else:assert len(namespace)==1 and len(namespace[0]['tools'])==210
+  rows.append({'index':index+1,'kind':'compaction' if compaction else 'normal','whole_json_tokens_proxy':count(body),'rea_namespace_tokens_proxy':0 if not namespace else count(namespace[0]),'rea_function_count':0 if not namespace else len(namespace[0]['tools'])})
+ wire.append({'case':root.name,'task_completed':report['task_completed'],'mode':report['mode'],'budget':report['observations']['tool_output_budget'],'requests':len(rows),'compactions':sum(r['kind']=='compaction' for r in rows),'whole_json_tokens_proxy':sum(r['whole_json_tokens_proxy'] for r in rows),'request_details':rows,**resources(root)})
+for root in sorted(lab.glob('final-sdk-*')):
+ if not root.is_dir():continue
+ report=json.loads((root/'report.json').read_text());assert report['status']=='passed' and not report['cleanup_failures'] and not report['receipt_failures']
+ sdk.append({'case':root.name,'provider':report['provider'],'symbol':report['symbol'],'detail':report['detail'],'warm':report['warm'],'tool_calls':len(report['calls']),'exported_bundle_bytes':(root/'evidence.json').stat().st_size,'disassembly_calls':[r for r in report['calls'] if r['phase'].startswith('disassembly:')],**resources(root)})
+receipt={'measurement_scope':'Actual public SDK and Codex0.163 app-server with real GDB/BFD. Local deterministic SSE fixture and observed-ID checkpoint; no AI inference, model usage/quality/acceptance or billing claim. Whole compact JSON serialization tokens are proxies and include all schemas/replays/compaction. Incomplete default baseline is excluded from completed-workflow gains. Fresh processes have unflushed OS caches; warm SDK cases repeat the selected disassembly in the same server. Native target never executed.','tokenizer':{'package':tiktoken.__version__,'encoding':enc.name},'runtime_pins':{variant:{k:v for k,v in json.loads((lab/(variant+'-manifest.json')).read_text()).items() if k!='files'} for variant in ['baseline','candidate']},'native':json.loads((lab/'native-prerequisite-manifest.json').read_text()),'wire':wire,'sdk':sdk}
+args.output.write_text(json.dumps(receipt,indent=2)+'\n');print(json.dumps({'wire':[{k:v for k,v in row.items() if k!='request_details'} for row in wire],'sdk':[{k:v for k,v in row.items() if k!='disassembly_calls'} for row in sdk]},indent=2))

experiments/analysis-efficiency/command-observation-oracle.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+import struct,hashlib,json,argparse
+parser=argparse.ArgumentParser(description='Independent ELF64 little-endian native-command byte oracle');parser.add_argument('--input',type=Path,required=True);parser.add_argument('--output',type=Path,required=True);args=parser.parse_args();path=args.input;b=path.read_bytes();assert b[:6]==b'\x7fELF\x02\x01'
+shoff=struct.unpack_from('<Q',b,40)[0];entsize,n,strings=struct.unpack_from('<HHH',b,58)
+assert entsize==64 and shoff+n*entsize<=len(b)
+sections=[struct.unpack_from('<IIQQQQIIQQ',b,shoff+i*entsize) for i in range(n)]
+rows=[]
+for si,s in enumerate(sections):
+ if s[1]!=11:continue
+ text=sections[s[6]];names=b[text[4]:text[4]+text[5]]
+ assert s[9]==24 and s[5]%s[9]==0
+ for ordinal,off in enumerate(range(s[4],s[4]+s[5],24)):
+  name,info,other,section,value,size=struct.unpack_from('<IBBHQQ',b,off)
+  end=names.find(b'\0',name);assert end>=name
+  if names[name:end] not in [b'execute_command',b'execute_command_internal'] or info&15!=2:continue
+  target=sections[section];file_offset=target[4]+value-target[3]
+  assert target[4]<=file_offset and file_offset+size<=target[4]+target[5]
+  rows.append({'symbol_table_section':si,'symbol_ordinal':ordinal,'name':names[name:end].decode('ascii'),'address':value,'bytes':size,'file_offset':file_offset,'first_16_bytes_hex':b[file_offset:file_offset+16].hex()})
+assert rows
+result={'path':str(path),'artifact_sha256':hashlib.sha256(b).hexdigest(),'artifact_bytes':len(b),'elf_class':64,'byte_order':'little','elf_machine':struct.unpack_from('<H',b,18)[0],'symbols':rows,'scope':'Independent ELF64 little-endian dynamic symbol/section mapping; no target execution or REA parser.'}
+args.output.write_text(json.dumps(result,indent=2)+'\n')
+print(json.dumps(result))

experiments/analysis-efficiency/command-observation-queries.mjs

@@ -0,0 +1,212 @@
+import assert from "node:assert/strict";
+import { createHash } from "node:crypto";
+import { createWriteStream } from "node:fs";
+import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises";
+import { createRequire } from "node:module";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { finished } from "node:stream/promises";
+import { pathToFileURL } from "node:url";
+import { parseArgs } from "node:util";
+import { createRetainedQueryCaller } from "./retained-query-calls.mjs";
+import { finalizeRetainedQueryReceipt } from "./retained-query-finalization.mjs";
+
+const { values } = parseArgs({
+  options: {
+    runtime: { type: "string" },
+    input: { type: "string" },
+    oracle: { type: "string" },
+    output: { type: "string" },
+    provider: { type: "string" },
+    symbol: { type: "string" },
+    detail: { type: "string", default: "complete" },
+    warm: { type: "boolean", default: false },
+  },
+});
+for (const key of ["runtime", "input", "oracle", "output"])
+  assert.equal(values[key], resolve(values[key]));
+assert.ok(["gdb", "objdump"].includes(values.provider));
+assert.ok(["complete", "summary"].includes(values.detail));
+const oracle = JSON.parse(await readFile(values.oracle, "utf8"));
+assert.ok(oracle.symbols.some((row) => row.name === values.symbol));
+const inputBytes = await readFile(values.input);
+assert.equal(
+  createHash("sha256").update(inputBytes).digest("hex"),
+  oracle.artifact_sha256,
+);
+const require = createRequire(join(values.runtime, "package.json"));
+const { Client } = await import(
+  pathToFileURL(require.resolve("@modelcontextprotocol/client"))
+);
+const { StdioClientTransport } = await import(
+  pathToFileURL(require.resolve("@modelcontextprotocol/client/stdio"))
+);
+await mkdir(values.output, { mode: 0o700 });
+const account = await mkdtemp(join(tmpdir(), "rea-command-views-"));
+const client = new Client({ name: "command-observation-task", version: "1" });
+const transport = new StdioClientTransport({
+  command: process.execPath,
+  args: [join(values.runtime, "scripts/rea.mjs"), "mcp"],
+  env: {
+    HOME: account,
+    PATH: process.env.PATH,
+    NODE_OPTIONS: "--max-old-space-size=2048",
+    UV_THREADPOOL_SIZE: "1",
+    REA_LOG_LEVEL: "silent",
+  },
+  stderr: "pipe",
+});
+const serverLog = createWriteStream(join(values.output, "server.log"));
+const serverLogFinished = finished(serverLog);
+serverLogFinished.catch(() => {});
+transport.stderr?.pipe(serverLog);
+const transcript = [];
+const report = {
+  question:
+    "Read architecture and the complete byte-annotated disassembly of one selected native function; export complete Evidence. No inferior execution.",
+  provider: values.provider,
+  symbol: values.symbol,
+  detail: values.detail,
+  warm: values.warm,
+  calls: [],
+  observations: [],
+};
+const call = createRetainedQueryCaller(client, (receipt, event) => {
+  report.calls.push(receipt);
+  transcript.push(event);
+});
+let sessionId;
+try {
+  await client.connect(transport);
+  const catalog = await client.listTools();
+  await writeFile(join(values.output, "catalog.json"), JSON.stringify(catalog));
+  const definitions = new Map(catalog.tools.map((tool) => [tool.name, tool]));
+  const originalCall = client.callTool.bind(client);
+  client.callTool = (input, options) =>
+    originalCall(input, {
+      ...options,
+      timeout: 60000,
+      toolDefinition: definitions.get(input.name),
+    });
+  const observe = async (name, input, phase = name) => {
+    const response = await call(name, input, phase);
+    assert.ok(!response.isError, JSON.stringify(response));
+    report.observations.push({
+      name,
+      phase,
+      evidence: response.structuredContent,
+    });
+    return response.structuredContent;
+  };
+  const completeParents = [];
+  const disassemble = async (phase) => {
+    const selection = values.detail === "summary" ? { detail: "summary" } : {};
+    const result =
+      values.provider === "gdb"
+        ? await observe(
+            "gdb_console",
+            {
+              session_id: sessionId,
+              command: `disassemble /r ${values.symbol}`,
+              ...selection,
+            },
+            phase,
+          )
+        : await observe(
+            "inspect_with_objdump",
+            {
+              path: values.input,
+              operation: "disassemble",
+              symbol: values.symbol,
+              ...selection,
+            },
+            phase,
+          );
+    completeParents.push(result);
+  };
+  if (values.provider === "gdb") {
+    const started = await observe("start_gdb_session", {});
+    sessionId = started.session_id;
+    for (const command of [
+      "set auto-load off",
+      "set debuginfod enabled off",
+      `file ${values.input}`,
+      "show architecture",
+    ])
+      await observe("gdb_console", { session_id: sessionId, command });
+  } else
+    await observe("inspect_with_objdump", {
+      path: values.input,
+      operation: "file_headers",
+    });
+  await disassemble("disassembly:fresh");
+  if (values.warm) await disassemble("disassembly:repeat");
+  if (sessionId) {
+    await observe("close_gdb_session", { session_id: sessionId });
+    sessionId = undefined;
+  }
+  await observe("export_evidence_bundle", {
+    path: join(values.output, "evidence.json"),
+  });
+  const bundle = JSON.parse(
+    await readFile(join(values.output, "evidence.json"), "utf8"),
+  );
+  for (const delivered of completeParents) {
+    const parent =
+      values.detail === "complete"
+        ? delivered
+        : bundle.records.find(
+            (record) =>
+              record.evidence_id ===
+              delivered.normalized_result.parent_evidence_id,
+          );
+    assert.ok(parent);
+    assert.equal(parent.normalized_result.completion_status, "complete");
+    assert.equal(parent.normalized_result.output_truncated, false);
+    if (values.provider === "gdb") assert.equal(parent.subject, null);
+    else assert.equal(parent.subject.digest.sha256, oracle.artifact_sha256);
+    if (values.detail === "summary") {
+      const selected = delivered.normalized_result.summary;
+      const { available_counts: counts, ...facts } = selected;
+      const { console, target, log, ...records } = parent.normalized_result;
+      assert.deepEqual(
+        values.provider === "gdb" ? facts : selected,
+        values.provider === "gdb" ? records : parent.normalized_result,
+      );
+      if (values.provider === "gdb")
+        assert.equal(counts.console_utf16_units, console.length);
+      assert.equal(delivered.raw_result, null);
+    }
+  }
+  report.status = "passed";
+} catch (error) {
+  report.status = "failed";
+  report.failure = String(error?.stack ?? error);
+  process.exitCode = 1;
+} finally {
+  if (sessionId)
+    try {
+      const closed = await call(
+        "close_gdb_session",
+        { session_id: sessionId },
+        "cleanup:gdb",
+      );
+      assert.ok(!closed.isError);
+    } catch (error) {
+      report.gdb_cleanup_failure = String(error);
+      process.exitCode = 1;
+    }
+  if (
+    !(await finalizeRetainedQueryReceipt({
+      report,
+      client,
+      transport,
+      account,
+      serverLog,
+      serverLogFinished,
+      output: values.output,
+      transcript,
+    }))
+  )
+    process.exitCode = 1;
+}

experiments/analysis-efficiency/command-observation-verify.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+import json,re,hashlib,argparse
+parser=argparse.ArgumentParser(description='Match every native instruction byte against independently mapped ELF file ranges');parser.add_argument('--root',type=Path,required=True);args=parser.parse_args();lab=args.root;oracle=json.loads((lab/'elf-oracle.json').read_text());image=Path(oracle['path']).read_bytes();assert hashlib.sha256(image).hexdigest()==oracle['artifact_sha256'];rows=[]
+for directory in sorted(lab.iterdir()):
+ bundle=directory/'evidence.json'
+ if not bundle.is_file():continue
+ for parent in json.loads(bundle.read_text())['records']:
+  if parent['operation'] not in ['gdb_console','inspect_with_objdump']:continue
+  params=parent['parameters'];gdb=parent['operation']=='gdb_console'
+  symbol=params.get('command','').removeprefix('disassemble /r ') if gdb else params.get('symbol')
+  if symbol not in {s['name'] for s in oracle['symbols']}:continue
+  spec=next(s for s in oracle['symbols'] if s['name']==symbol)
+  report=parent['normalized_result'];assert report['completion_status']=='complete' and not report['output_truncated']
+  if gdb:
+   assert parent['subject'] is None
+   console='\n'.join(line for line in report['records'] if line.startswith('~'));assert console==report['console']
+   text=''.join(json.loads(line[1:]) for line in report['records'] if line.startswith('~'))
+   pattern=r'^\s*(?:=>\s*)?0x([0-9a-f]+)[^:]*:\s*((?:[0-9a-f]{2}(?:\s+|$))+)'
+  else:
+   assert parent['subject']['digest']['sha256']==oracle['artifact_sha256'];text=report['stdout'];pattern=r'^\s*([0-9a-f]+):\s*((?:[0-9a-f]{2}(?:\s+|$))+)'
+  address=spec['address'];collected=bytearray();count=0
+  for line in text.splitlines():
+   m=re.match(pattern,line)
+   if not m:continue
+   actual=int(m[1],16);raw=bytes.fromhex(m[2]);assert actual==address+len(collected),(directory,actual,address+len(collected));collected.extend(raw);count+=1
+  expected=image[spec['file_offset']:spec['file_offset']+spec['bytes']];assert collected==expected,(directory,len(collected),len(expected));rows.append({'case':directory.name,'mode':'gdb' if gdb else 'objdump','symbol':symbol,'byte_bearing_lines':count,'bytes':len(collected),'exact_file_bytes':True,'parent_evidence_id':parent['evidence_id']})
+assert rows
+(lab/'native-final-verification.json').write_text(json.dumps({'status':'passed','artifact_sha256':oracle['artifact_sha256'],'records':rows},indent=2)+'\n');print(json.dumps(rows))

experiments/analysis-efficiency/command-observation-views.md

@@ -0,0 +1,139 @@
+# Retained command observations
+
+Addresses [#5584](https://github.com/morluto/rea/issues/5584). GDB command Evidence
+repeats ordered MI records in stream groups and a raw mirror; objdump repeats
+stdout/stderr in its raw mirror. Explicit `detail: summary` retains the complete
+parent and projects existing owner facts once. Default complete Evidence is
+unchanged. This adds no tool, MI decoder, live-state cache or command authority.
+
+GDB summaries preserve **every ordered record**, including asynchronous and
+unrecognized records, correlated MI result, command, completion/truncation,
+invalid-UTF-8 and process-exit facts. They omit the convenience console/target/log
+groupings and report their original UTF-16 sizes. Objdump summaries preserve
+opaque stdout/stderr, exit/signal, completion and pre/post digest-change facts.
+Both carry source parameters/provider/profile, parent identity, limitations and
+unknowns. Subjectless observations stay subjectless. Historical observations do
+not establish current target state or global coverage. The existing
+`inspect_analysis_view` summary operation works after debugger close.
+
+The domain codec now owns the command fact shapes used by contracts and GDB's
+capture type; protocol interpretation stays with its provider. MCP requires
+retention before command execution. CLI uses the existing shared projection and
+nonoverwriting bundle publisher after provider cleanup:
+
+```sh
+rea inspect-with-objdump /usr/bin/bash disassemble --symbol execute_command \
+  --detail summary --evidence_output /tmp/bash-command-evidence.json --json
+```
+
+## Real input and correctness
+
+Linux x86-64, Node 24.18.0/npm 11.16.0; GNU GDB 13.1 (Debian 13.1-3),
+GNU objdump 2.40, installed Bash ELF 1,265,648 bytes, SHA-256
+`55b89ab22bee4792a210f493a53fb066accd5d30b69837c28d98be5ff863efcf`.
+The [receipt](command-observation-views.json) pins native binaries, baseline main
+`c2858d2aa0916f0dd2d073d848b8c2195d9bf6b3`, source patch and packaged runtime digests.
+No target execution, attach, GUI or database mutation occurred.
+
+An [independent ELF reader](command-observation-oracle.py) maps dynamic symbols
+to exact file ranges: `execute_command`, 239 bytes at `0x4ba50`, and
+`execute_command_internal`, 15,428 bytes at `0x47e00`. The
+[verifier](command-observation-verify.py) checks every disassembled byte and its
+contiguous address against those ranges. GDB reports 66/3,365 instructions;
+objdump byte lines also include continuations, so their count is not an
+instruction count. Ordered summary facts match each exported complete parent.
+GDB Evidence has a null subject; the external oracle does not manufacture a
+target binding. Actual CLI default objdump Evidence matches the baseline exactly,
+including identity; symbol/pathless summaries preserve it in bundles. CLI checks
+also cover rejected overwrite, missing destination before dispatch, and a real
+GDB command with a caller-owned PTY that remains available after debugger exit.
+
+## Whole tasks and resource costs
+
+The [SDK driver](command-observation-queries.mjs) discovers all 210 tools, reads
+architecture and symbol-scoped disassembly, closes GDB where applicable, exports
+complete Evidence and finalizes owned resources. GDB uses 8 calls; objdump uses 3.
+Warm cases repeat disassembly in the same process, adding one call. Fresh-process
+cases have **unflushed OS caches**, not disk-cold inputs. Native commands are
+rerun; results are not reused as current live state.
+
+The [actual Codex 0.163 app-server driver](command-observation-wire.py) uses a
+loopback deterministic SSE fixture, no AI inference or paid service. All 210
+function schemas are present in every normal model-facing request. Matched tasks
+use full input/output profiles; only disassembly selects summary. The
+[counter](command-observation-costs.py) counts compact JSON serialization with
+`tiktoken 0.12.0 / cl100k_base` across **every** request, including schemas,
+replayed results and the actual client's local compaction request. The fixture
+answers compaction with observed identifiers and the retained-body location.
+These are serialization-token proxies, not model rendering, acceptance, answer
+quality, actual usage or billing. Compaction behavior is qualified only for this
+client/configuration; no inference quality is established.
+
+| Completed GDB task             | JSON token proxy before → after | Requests |  Wall seconds |   CPU seconds | Family peak MiB |
+| ------------------------------ | ------------------------------: | -------: | ------------: | ------------: | --------------: |
+| Small, 30,000 per-tool budget  |               605,133 → 580,070 |    9 → 9 | 7.906 → 8.618 | 7.195 → 7.832 |   788.9 → 785.9 |
+| Large, 350,000 per-tool budget |           1,269,405 → 1,091,290 |   10 → 9 | 9.551 → 9.127 | 8.832 → 8.370 |   805.2 → 770.8 |
+
+Whole-task context proxies decrease **4.14% / 14.03%**, rather than the much
+larger reply-byte reduction. The large baseline makes one compaction request;
+the candidate makes none. The small task is slower. These single paired samples
+do not establish a general speedup. With the client's default output budget,
+the small baseline delivers truncated invalid inner JSON and does **not**
+complete the question; summary completes it. That incomplete baseline is
+excluded from the percentage comparison.
+
+| SDK task      | Reply bytes before → after | Whole wall seconds |   CPU seconds | Family peak MiB |
+| ------------- | -------------------------: | -----------------: | ------------: | --------------: |
+| GDB small     |            54,123 → 18,009 |      6.631 → 5.447 | 6.102 → 5.240 |   536.3 → 535.8 |
+| GDB large     |        2,763,983 → 700,453 |      6.965 → 6.660 | 6.729 → 6.451 |   595.0 → 545.8 |
+| Objdump small |            22,467 → 15,990 |      7.599 → 7.699 | 7.110 → 7.264 |   527.8 → 532.5 |
+| Objdump large |          964,889 → 487,219 |      7.737 → 7.499 | 7.334 → 6.935 |   537.0 → 536.6 |
+
+Same-process repeat disassembly takes 1,100 → 1,016 ms for GDB and
+1,070 → 939 ms for objdump in these observations. Whole warm-task CPU improves
+7.613 → 7.162 s for GDB but increases 7.819 → 7.859 s for objdump. Timings include
+existing ownership checks; no check was removed or cached.
+
+Retaining a summary also retains an additional derived record: large GDB bundles
+grow 1,310,073 → 1,640,022 bytes; objdump 474,288 → 712,656. This trades retained
+storage for smaller delivery; long-session heap growth is unmeasured. Retained-data lifetime follows the existing investigation ledger and cleanup
+owners. Output still scales
+with the requested complete disassembly; summary is not a fixed-size preview.
+
+Each experiment runs in one owned systemd scope, one CPU, no swap, 4 GiB hard /
+3 GiB early memory thresholds and host-memory admission/stop thresholds. Recorded
+payload exit and cleanup are verified, with zero OOM/resource events. Final
+whole-scope counters disappear after teardown and remain explicitly unverified.
+The receipt includes CPU time, wall time, peak family memory, warm cases and
+bundle growth. Build/check scopes use separate bounded budgets. Host-memory
+check stops and initial fixture failures are preserved locally and are not passes.
+
+## Coverage and reproduction
+
+320 focused tests cover the new SDK schemas/views plus existing GDB queue,
+malformed output, cancellation, process exit, target identity and cleanup cases,
+objdump symbol selection and command admission. Real provider evidence is from
+the CLI/SDK/client runs above, not simulated-provider tests.
+
+Run the oracle for the selected Bash image, then use the SDK driver with
+`--runtime`, `--input`, `--oracle`, `--output`, `--provider gdb|objdump`,
+`--symbol`, `--detail complete|summary` and optional `--warm`. The wire driver
+also takes the fixed Codex/Node paths, `--mode`, `--symbol` and `--budget`; it always performs the
+export (eight tool calls). Run each through `resource-run.py --observe-payload-exit`, preserving
+its sibling `.log`; the verifier and counter accept `--root` for those results.
+
+Default Rizin execution returned ENOENT with no configured alternative. Cutter
+reported no live bridge; no configured selector or default bridge directory was
+present. Real Rizin/Cutter performance and GUI workflows remain unverified.
+Their command semantics were not changed. Existing objdump native symbol
+selection was reused. No claim covers arbitrary GDB target execution or another
+host client's output budget.
+
+Review follow-up adds nine CLI publication regressions (16 CLI tests including the
+existing cleanup cases), three deterministic delivery-oracle tests, and a focused
+strict typecheck of the new CLI boundary and its transitive source. Real client
+small/default and large/generous summary runs pass the strengthened oracle and
+export complete parents; only the explicit complete/default baseline may finish
+with an incomplete answer. The unused export switch was removed: export is
+mandatory in this question. Product source and the original paired measurements
+are unchanged; the receipt records these additional validation runs separately.

experiments/analysis-efficiency/command-observation-wire.py

@@ -0,0 +1,301 @@
+"""Actual Codex wire capture with a deterministic local SSE backend; no AI inference."""
+import argparse
+import http.server
+import threading
+import asyncio
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import time
+import traceback
+import subprocess
+from command_observation_delivery import verify_command_delivery
+
+
+async def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--runtime", type=Path, required=True)
+    parser.add_argument("--codex", type=Path, required=True)
+    parser.add_argument("--node", type=Path, required=True)
+    parser.add_argument("--input-profile", choices=["full", "compact"], default="full")
+    parser.add_argument("--output-profile", choices=["full", "off"], default="full")
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--oracle", type=Path, required=True)
+    parser.add_argument("--expected-tools", type=int, default=210)
+    parser.add_argument("--mode", choices=["complete", "summary"], required=True)
+    parser.add_argument("--question", choices=["disassembly"], default="disassembly")
+    parser.add_argument("--symbol", choices=["execute_command", "execute_command_internal"], required=True)
+    parser.add_argument("--budget", type=int, default=30000)
+    args = parser.parse_args()
+    oracle=json.loads(args.oracle.read_text())
+    target=str(args.input)
+    assert args.input.is_absolute() and args.oracle.is_absolute() and args.output.is_absolute()
+    assert hashlib.sha256(args.input.read_bytes()).hexdigest()==oracle["artifact_sha256"]
+    def delivered(pieces):
+        assert isinstance(pieces,str) and pieces.startswith("Wall time: "), pieces[:300]
+        _,sep,payload=pieces.partition("Output:\n");assert sep
+        try: value=json.loads(payload)
+        except json.JSONDecodeError: return {"_invalid_json_payload":payload}
+        return value["result"] if "result" in value and "normalized_result" not in value else value
+    def outputs_of(body):
+        return {item["call_id"]:delivered(item["output"]) for item in body.get("input",[]) if item.get("type")=="function_call_output"}
+    steps=8
+    args.output.mkdir(mode=0o700)
+    audit_home = args.output / "account"
+    codex_state = audit_home / ".codex"
+    mcp_home = audit_home / "mcp"
+    workspace = audit_home / "workspace"
+    for directory in [codex_state, mcp_home, workspace]:
+        directory.mkdir(parents=True)
+    captured = []
+    known_outputs = {}
+    state = {"normal_requests":0,"compactions":0}
+    class Capture(http.server.BaseHTTPRequestHandler):
+        def log_message(self, *_):
+            pass
+        def do_POST(self):
+            size = int(self.headers["Content-Length"])
+            assert 0 < size <= 16 * 1024 * 1024
+            assert self.headers.get("Content-Encoding", "identity") == "identity"
+            body = json.loads(self.rfile.read(size))
+            captured.append({"path": self.path, "body": body})
+            (args.output / "model-requests.json").write_text(json.dumps(captured) + "\n")
+            http_index = len(captured)
+            known_outputs.update(outputs_of(body))
+            names = [t.get("name", "") for ns in body.get("tools", []) if ns.get("type") == "namespace" and ns.get("name") == "mcp__rea_audit" for t in ns.get("tools", [])]
+            if not names:
+                last=body.get('input',[])[-1]
+                assert 'CONTEXT CHECKPOINT COMPACTION' in json.dumps(last)
+                state['compactions']+=1;assert state['compactions']<=1
+                evidence=known_outputs['call_local_6'];assert 'normalized_result' in evidence
+                captured_result=evidence['normalized_result'];command_report=captured_result.get('summary',captured_result)
+                checkpoint={'fixture':'deterministic protocol checkpoint; no AI inference or model-quality claim','task':'Full native Bash function disassembly has been delivered to this compaction request. Retain complete Evidence and finish owned cleanup.','path':target,'artifact_sha256_external_oracle':oracle['artifact_sha256'],'symbol':args.symbol,'session_id':known_outputs['call_local_1']['session_id'],'parent_evidence_id':captured_result.get('parent_evidence_id',evidence['evidence_id']),'subject':evidence['subject'],'completion_status':command_report['completion_status'],'output_truncated':command_report['output_truncated'],'native_console_utf16_units':command_report.get('available_counts',{}).get('console_utf16_units',len(command_report.get('console','').encode('utf-16-le'))//2),'full_body_location':'same-session retained Evidence; export next to '+str(args.output/'evidence.json'),'remaining':['close_gdb_session','export_evidence_bundle']}
+                text=json.dumps(checkpoint)
+                item={"type":"message","id":f"msg_checkpoint_{state['compactions']}","role":"assistant","status":"completed","content":[{"type":"output_text","text":text,"annotations":[]}]}
+                captured[-1]['kind']='deterministic-compaction'
+            else:
+                state['normal_requests']+=1;index=state['normal_requests'];assert index<=steps+1
+                captured[-1]['kind']='tool-planning-or-final'
+                if index<=steps:
+                    if index==1:
+                        operation="start_gdb_session";arguments={}
+                    else:
+                        session=known_outputs["call_local_1"]["session_id"]
+                        if index<=6:
+                            command=['set auto-load off','set debuginfod enabled off',f'file {target}','show architecture',f'disassemble /r {args.symbol}'][index-2]
+                            operation="gdb_console";arguments={"session_id":session,"command":command}
+                            if index==6 and args.mode=="summary":arguments["detail"]="summary"
+                        elif index==7:
+                            operation="close_gdb_session";arguments={"session_id":session}
+                        else:
+                            operation="export_evidence_bundle";arguments={"path":str(args.output/'evidence.json')}
+                    selected=[name for name in names if name.endswith(operation)];assert len(selected)==1
+                    item={"type":"function_call","id":f"fc_local_{index}","call_id":f"call_local_{index}","name":selected[0],"namespace":"mcp__rea_audit","arguments":json.dumps(arguments),"status":"completed"}
+                else:
+                    item={"type":"message","id":"msg_local_done","role":"assistant","status":"completed","content":[{"type":"output_text","text":"Scripted local GDB disassembly task complete.","annotations":[]}]}
+            response_id=f"resp_local_{http_index}"
+            base = {"id": response_id, "object": "response", "model": body.get("model"), "status": "completed", "output": [item]}
+            events = [
+                {"type": "response.created", "response": {**base, "status": "in_progress", "output": []}},
+                {"type": "response.output_item.added", "output_index": 0, "item": {**item, "status": "in_progress"}},
+                {"type": "response.output_item.done", "output_index": 0, "item": item},
+                {"type": "response.completed", "response": base},
+            ]
+            wire = "".join("event: " + e["type"] + "\ndata: " + json.dumps(e) + "\n\n" for e in events).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Content-Length", str(len(wire)))
+            self.end_headers()
+            self.wfile.write(wire)
+    endpoint = http.server.HTTPServer(("127.0.0.1", 0), Capture)
+    endpoint_thread = threading.Thread(target=endpoint.serve_forever, daemon=True)
+    endpoint_thread.start()
+    task_path = f"{args.node.parent}:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+    mcp_env = {"HOME": str(mcp_home), "PATH": task_path,
+               "NODE_OPTIONS": "--max-old-space-size=2048", "UV_THREADPOOL_SIZE": "1",
+               "REA_LOG_LEVEL": "silent", "REA_MCP_INPUT_SCHEMA_PROFILE": args.input_profile,
+               "REA_MCP_OUTPUT_SCHEMA_PROFILE": args.output_profile}
+    config = 'model_provider = "rea_capture"\nmodel = "rea-wire-probe"\n[analytics]\nenabled = false\n[mcp_servers.rea_audit]\n'
+    config += f'command = {json.dumps(str(args.node))}\n'
+    config += f'args = {json.dumps([str(args.runtime / "scripts/rea.mjs"), "mcp"])}\n'
+    config += 'startup_timeout_sec = 60\ntool_timeout_sec = 60\n'
+    config += '[mcp_servers.rea_audit.env]\n'
+    config += ''.join(f'{key} = {json.dumps(value)}\n' for key, value in mcp_env.items())
+    for name in ['start_gdb_session','gdb_console','close_gdb_session','export_evidence_bundle']:
+        config += f'[mcp_servers.rea_audit.tools.{name}]\napproval_mode = "approve"\n'
+        if args.budget: config += f'output_token_limit = {args.budget}\n'
+    config += f'\n[model_providers.rea_capture]\nname = "Local scripted capture"\nbase_url = "http://127.0.0.1:{endpoint.server_port}/v1"\nwire_api = "responses"\nrequires_openai_auth = false\nsupports_websockets = false\nrequest_max_retries = 0\nstream_max_retries = 0\n'
+    (codex_state / "config.toml").write_text(config)
+    (args.output / "declared-config.toml").write_text(config)
+    environment = {"HOME": str(audit_home), "CODEX_HOME": str(codex_state), "PATH": task_path,
+                   "NODE_OPTIONS": "--max-old-space-size=2048", "UV_THREADPOOL_SIZE": "1"}
+    report = {"question": "Return architecture and complete byte-annotated disassembly for one selected native Bash function and export complete Evidence. No target execution.",
+              "client": "Codex app-server 0.163.0", "model_turns_started": 1, "backend": "local deterministic SSE fixture; no AI model or paid service",
+              "actual_model_usage": None, "input_profile": args.input_profile,
+              "output_profile": args.output_profile, "calls": [], "observations": [],
+              "unexpected_server_requests": [], "cleanup_failures": []}
+    client_version = subprocess.check_output([str(args.codex), "--version"], env=environment, cwd=workspace, text=True, timeout=10).strip()
+    assert client_version == "codex-cli 0.163.0", client_version
+    transcript = []
+    pending = {}
+    completed_turn = asyncio.Event()
+    next_id = 0
+    stderr = (args.output / "server.log").open("wb")
+    process = await asyncio.create_subprocess_exec(str(args.codex), "app-server", "--stdio",
+        stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=stderr,
+        cwd=workspace, env=environment, limit=16 * 1024 * 1024)
+
+    async def send(message):
+        encoded = (json.dumps(message, separators=(",", ":")) + "\n").encode()
+        transcript.append({"direction": "client", "bytes": len(encoded), "message": message})
+        process.stdin.write(encoded)
+        await process.stdin.drain()
+
+    async def read_messages():
+        try:
+            while line := await process.stdout.readline():
+                value = json.loads(line)
+                transcript.append({"direction": "server", "bytes": len(line), "message": value})
+                if value.get("method") == "turn/completed":
+                    report["turn"] = value["params"]["turn"]
+                    completed_turn.set()
+                if "id" in value and "method" not in value:
+                    future = pending.pop(value["id"], None)
+                    if future and not future.done():
+                        future.set_result(value)
+                elif "id" in value:
+                    report["unexpected_server_requests"].append(value["method"])
+                    await send({"id": value["id"], "error": {"code": -32601, "message": "Unexpected request in scripted metadata probe"}})
+            raise RuntimeError("app-server stdout closed")
+        except Exception as error:
+            for future in pending.values():
+                if not future.done():
+                    future.set_exception(error)
+
+    reader = asyncio.create_task(read_messages())
+
+    async def request(method, params):
+        nonlocal next_id
+        assert method in ["initialize", "thread/start", "mcpServerStatus/list", "mcpServer/tool/call", "turn/start"]
+        next_id += 1
+        message = {"id": next_id, "method": method, "params": params}
+        future = asyncio.get_running_loop().create_future()
+        pending[next_id] = future
+        start = time.monotonic()
+        await send(message)
+        reply = await asyncio.wait_for(future, 90)
+        report["calls"].append({"method": method, "elapsed_ms": (time.monotonic() - start) * 1000,
+                                "error": reply.get("error")})
+        if "error" in reply:
+            raise RuntimeError(json.dumps(reply["error"])[:4000])
+        return reply["result"]
+
+    try:
+        await request("initialize", {"clientInfo": {"name": "rea_host_probe", "title": "REA host compatibility probe", "version": "1"},
+                                     "capabilities": {"experimentalApi": True, "explicitGatewayOauth": True}})
+        await send({"method": "initialized"})
+        created = await request("thread/start", {"cwd": str(workspace), "ephemeral": True,
+                    "approvalPolicy": "never", "sandbox": "read-only"})
+        thread_id = created["thread"]["id"]
+        assert not created["thread"].get("turns", [])
+        inventory = await request("mcpServerStatus/list", {"threadId": thread_id,
+                    "serverName": "rea_audit", "detail": "toolsAndAuthOnly", "limit": 10})
+        assert inventory.get("nextCursor") is None
+        assert len(inventory["data"]) == 1
+        server = inventory["data"][0]
+        assert server["name"] == "rea_audit" and server.get("toolsError") is None
+        tools = list(server["tools"].values())
+        names = [tool["name"] for tool in tools]
+        assert len(names) == len(set(names)) == args.expected_tools
+        assert "gdb_console" in names
+        output_schemas = sum(tool.get("outputSchema") is not None for tool in tools)
+        assert output_schemas == (args.expected_tools if args.output_profile == "full" else 0)
+        raw = json.dumps(inventory, separators=(",", ":"), ensure_ascii=False).encode()
+        (args.output / "catalog.json").write_bytes(raw)
+        report["catalog"] = {"tools": len(tools), "output_schemas": output_schemas,
+                             "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest(),
+                             "server_info": server.get("serverInfo"), "runtime_status": server.get("runtimeStatus")}
+        await request("turn/start", {"threadId":thread_id,"input":[{"type":"text","text":f"Read architecture and the complete byte-annotated disassembly of {args.symbol} in the selected Bash executable. Retain and export complete Evidence, then close the owned debugger. The local fixture scripts the required tools; no AI inference or target execution is used."}]})
+        await asyncio.wait_for(completed_turn.wait(), 90)
+        assert report["turn"]["status"] == "completed", report["turn"]
+        assert state["normal_requests"]==steps+1
+        assert len(captured)==steps+1+state["compactions"]
+        outputs=known_outputs
+        assert len(outputs)==steps
+        bundle=json.loads((args.output/'evidence.json').read_text())
+        evidence=next(e for e in bundle['records'] if e['operation']=='gdb_console' and e['parameters']['command']==f'disassemble /r {args.symbol}')
+        assert evidence['subject'] is None
+        assert evidence['normalized_result']['completion_status']=='complete' and not evidence['normalized_result']['output_truncated']
+        delivered_result=outputs['call_local_6']
+        delivery_complete=verify_command_delivery(delivered_result,evidence,args.mode,args.budget)
+        if delivery_complete and args.mode=='summary':
+            assert next(e for e in bundle['records'] if e['evidence_id']==delivered_result['evidence_id'])==delivered_result
+        report['task_completed']=delivery_complete
+        report['observations']={'parent_evidence_id':evidence['evidence_id'],'exported':True,'native_console_utf16_units':len(evidence['normalized_result']['console'].encode('utf-16-le'))//2,'client_delivered_full_evidence':delivery_complete,'selected_symbol':args.symbol,'tool_output_budget':args.budget or 'client default','subject':evidence['subject']}
+        report["captured_requests"]=len(captured)
+        report["compaction_requests"]=state["compactions"]
+        report["compaction_backend"]="Deterministic checkpoint of observed identifiers and retained-body location; no AI inference or quality claim."
+        report["mode"]=args.mode
+        report["request_tool_counts"] = [len(c["body"].get("tools", [])) for c in captured]
+        assert not report["unexpected_server_requests"]
+        assert hashlib.sha256(args.input.read_bytes()).hexdigest()==oracle["artifact_sha256"]
+        report["status"] = "passed" if report["task_completed"] else "expected-incomplete-task"
+    except Exception:
+        report["status"] = "failed"
+        report["failure"] = traceback.format_exc()[-6000:]
+    finally:
+        process.stdin.close()
+        try:
+            await asyncio.wait_for(process.wait(), 15)
+        except asyncio.TimeoutError:
+            report["cleanup_failures"].append("app-server did not exit after stdin close; terminated owned process")
+            process.terminate()
+            try:
+                await asyncio.wait_for(process.wait(), 5)
+            except asyncio.TimeoutError:
+                process.kill()
+                await process.wait()
+        await reader
+        # app-server can acknowledge exit before its MCP child finishes EOF
+        # cleanup. Wait on the allocated resource scope; never target a PID.
+        try:
+            groups = [line[3:] for line in Path("/proc/self/cgroup").read_text().splitlines()
+                      if line.startswith("0::")]
+            assert len(groups) == 1
+            group = Path("/sys/fs/cgroup") / groups[0].lstrip("/")
+            assert re.fullmatch(r"rea-efficiency-[0-9a-f]{32}\.scope", group.name)
+            observer = os.getppid()
+            observer_args = Path(f"/proc/{observer}/cmdline").read_bytes().split(b"\0")
+            assert any(Path(os.fsdecode(value)).name == "resource_payload.py" for value in observer_args if value)
+            expected = {os.getpid(), observer}
+            until = time.monotonic() + 10
+            while True:
+                actors = {int(value) for value in (group / "cgroup.procs").read_text().splitlines()}
+                assert expected <= actors
+                remaining = sorted(actors - expected)
+                if not remaining or time.monotonic() >= until:
+                    break
+                await asyncio.sleep(0.05)
+            report["owned_scope_drain"] = {"remaining_processes": remaining}
+            if remaining:
+                report["cleanup_failures"].append("owned MCP child did not settle after app-server exit")
+        except (OSError, ValueError, AssertionError) as error:
+            report["owned_scope_drain"] = {"status": "unavailable", "reason": type(error).__name__}
+            report["cleanup_failures"].append("allocated resource-scope cleanup could not be verified")
+        endpoint.shutdown()
+        endpoint.server_close()
+        endpoint_thread.join(5)
+        stderr.close()
+        report["appserver_exit_code"] = process.returncode
+        if process.returncode != 0:
+            report["cleanup_failures"].append(f"app-server exit code: {process.returncode}")
+        (args.output / "transcript.json").write_text(json.dumps(transcript) + "\n")
+        (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+        print(json.dumps({key: report.get(key) for key in ["status", "failure", "catalog", "observations", "cleanup_failures", "appserver_exit_code"]}))
+    return 0 if report["status"] in ["passed","expected-incomplete-task"] and not report["cleanup_failures"] else 1
+
+
+raise SystemExit(asyncio.run(main()))

experiments/analysis-efficiency/command_observation_delivery.py

@@ -0,0 +1,39 @@
+"""Acceptance of the captured GDB answer, separate from native execution success."""
+
+
+def verify_command_delivery(delivered, parent, mode, budget):
+    assert mode in ('complete', 'summary')
+    if '_invalid_json_payload' in delivered:
+        assert mode == 'complete' and budget == 0, (
+            'Incomplete delivery is allowed only for the '
+            'complete/default-budget baseline'
+        )
+        return False
+    if mode == 'complete':
+        assert delivered == parent
+        return True
+
+    selected = delivered['normalized_result']
+    assert selected['parent_evidence_id'] == parent['evidence_id']
+    assert selected['artifact'] is None
+    profile = parent['analysis_profile']
+    expected_profile = None if profile is None else profile['digest']
+    assert selected['source_analysis_profile_digest'] == expected_profile
+    assert selected['source_parameters'] == parent['parameters']
+    assert selected['source_provider'] == parent['provider']
+    assert delivered['raw_result'] is None and delivered['subject'] is None
+    expected = {
+        key: value for key, value in parent['normalized_result'].items()
+        if key not in ('console', 'target', 'log')
+    }
+    actual = {
+        key: value for key, value in selected['summary'].items()
+        if key != 'available_counts'
+    }
+    assert actual == expected
+    counts = {'records': len(expected['records'])}
+    for name in ('console', 'target', 'log'):
+        text = parent['normalized_result'][name]
+        counts[name + '_utf16_units'] = len(text.encode('utf-16-le')) // 2
+    assert selected['summary']['available_counts'] == counts
+    return True

experiments/analysis-efficiency/test_command_observation_delivery.py

@@ -0,0 +1,81 @@
+"""A truncated summary must never qualify as a completed investigation."""
+from copy import deepcopy
+import unittest
+from command_observation_delivery import verify_command_delivery
+
+
+class CommandDelivery(unittest.TestCase):
+    def test_truncation_is_accepted_only_for_complete_default_baseline(self):
+        truncated = {'_invalid_json_payload': '{truncated'}
+        self.assertFalse(verify_command_delivery(truncated, {}, 'complete', 0))
+        for mode, budget in [
+            ('summary', 0), ('summary', 30000), ('complete', 30000),
+        ]:
+            with self.subTest(mode=mode, budget=budget):
+                with self.assertRaises(AssertionError):
+                    verify_command_delivery(truncated, {}, mode, budget)
+
+    def test_valid_complete_answer_can_complete_under_either_budget(self):
+        parent = {'evidence_id': 'parent'}
+        for budget in [0, 30000]:
+            self.assertTrue(
+                verify_command_delivery(parent, parent, 'complete', budget)
+            )
+        with self.assertRaises(AssertionError):
+            verify_command_delivery({'error': 'failed command'}, parent, 'complete', 0)
+
+    def test_summary_must_preserve_every_record_and_accurate_stream_counts(self):
+        report = {
+            'command': 'disassemble /r entry', 'mi': '7^done',
+            'records': ['~"🧪\\n"', '=async', 'unknown', '7^done'],
+            'console': '🧪', 'target': '', 'log': '',
+            'completion_status': 'complete', 'output_truncated': False,
+        }
+        parent = {
+            'evidence_id': 'parent',
+            'parameters': {'command': report['command']},
+            'provider': {'id': 'gdb'}, 'analysis_profile': None,
+            'normalized_result': report,
+        }
+        facts = {
+            key: value for key, value in report.items()
+            if key not in ('console', 'target', 'log')
+        }
+        selected = {
+            'evidence_id': 'derived', 'raw_result': None, 'subject': None,
+            'normalized_result': {
+                'parent_evidence_id': 'parent', 'artifact': None,
+                'source_analysis_profile_digest': None,
+                'source_parameters': parent['parameters'],
+                'source_provider': parent['provider'],
+                'summary': {
+                    **facts,
+                    'available_counts': {
+                        'records': 4, 'console_utf16_units': 2,
+                        'target_utf16_units': 0, 'log_utf16_units': 0,
+                    },
+                },
+            },
+        }
+        self.assertTrue(verify_command_delivery(selected, parent, 'summary', 0))
+        for damage in ['records', 'counts', 'parameters', 'profile']:
+            bad = deepcopy(selected)
+            result = bad['normalized_result']
+            if damage == 'records':
+                result['summary']['records'].pop(1)
+            elif damage == 'counts':
+                result['summary']['available_counts']['console_utf16_units'] = 1
+            elif damage == 'profile':
+                result['source_analysis_profile_digest'] = 'wrong'
+            else:
+                result['source_parameters'] = {}
+            with self.subTest(damage=damage):
+                with self.assertRaises(AssertionError):
+                    verify_command_delivery(bad, parent, 'summary', 0)
+        parent['analysis_profile'] = {'digest': 'recorded-profile'}
+        selected['normalized_result']['source_analysis_profile_digest'] = 'recorded-profile'
+        self.assertTrue(verify_command_delivery(selected, parent, 'summary', 0))
+
+
+if __name__ == '__main__':
+    unittest.main()