fix(kuna): preserve ownership through cancellation and failed verification (#4029)

Let individual requests cancel their shared assembly-index wait while the session aborts and drains the owned producer. Preserve verifier workspaces until provider, transport and process cleanup are confirmed; report failed runs with original errors and lineage evidence.

Co-authored-by: gensokyo-dev <342580529+gensokyo-dev@users.noreply.github.com>
Co-authored-by: Kaoru0822 <208281322+kaoru0822-kitauji@users.noreply.github.com>

docs/kuna-provider.md

@@ -51,6 +51,8 @@ Pseudocode, assembly, `analyze_function` and data-name answers record where thei
 
 The background export belongs to the session. `close_binary`, or the end of a CLI command, cancels it unless another session still waits for it. A one-shot CLI decompile therefore answers from a single-function run; any request that needs the export, such as `procedure_assembly` or `analyze_function`, keeps it.
 
+The assembly label index is scanned once per export reader. Cancelling a request stops its wait without interrupting other readers; closing the session aborts and drains the shared scan before releasing its files.
+
 Regex procedure search uses ECMAScript Unicode matching in a cancellable worker with a five-second matching deadline. Literal matching remains the default. Cancellation or a matching deadline leaves the session available.
 
 Single-function subprocess output and each cached source facet share a 16 MiB buffered-text capacity. Cached spans are checked before allocation; larger text remains in the local export and the error identifies the file and range. `analyze_function` initially returns the complete dossier, which can still be large; later retained views do not remove that first transfer. Prefer the individual text tools for a focused question.
@@ -82,3 +84,5 @@ REA_KUNA_COMMAND=/absolute/path/to/kuna GHIDRA_INSTALL_DIR=/absolute/path/to/ghi
 ```
 
 The lane compiles a small source-owned fixture with the host C compiler and verifies the MCP workflow end to end. `--targets DIR --functions a,b` adds caller-built binaries. `--compare-ghidra` (with `GHIDRA_INSTALL_DIR`) runs those same functions through Ghidra; it requires `--targets` and `--functions` and refuses empty matrices or requested function names missing from every target. See [testing](testing.md#kuna-lane) for coverage.
+
+Failed runs also report process-lineage evidence and exit unsuccessfully. If provider or transport cleanup is unconfirmed, the report says `cleanup: "incomplete"` and records `retained_workspace`; that directory is preserved for investigation and cleanup retries.

scripts/verify-real-kuna.mjs

@@ -25,6 +25,7 @@ import { fileURLToPath } from "node:url";
 import { parseArgs, promisify } from "node:util";
 import { Client } from "@modelcontextprotocol/client";
 import { StdioClientTransport } from "@modelcontextprotocol/client/stdio";
+import { execFileOutput } from "../dist/process/ExecFileOutput.js";
 import { requireMcpOperationResult } from "./lib/mcp-verifier-results.mjs";
 import {
   assertVerifierCleanupComplete,
@@ -62,10 +63,13 @@ if (values["compare-ghidra"]) {
     "Ghidra comparison prerequisite missing: GHIDRA_INSTALL_DIR (bring your own Ghidra 12.1.4 and JDK 21)",
   );
 }
-const kunaVersion = (await exec(kuna, ["--version"])).stdout.trim();
 const run = createVerifierRun();
 const workspace = await mkdtemp(join(tmpdir(), "rea-kuna-lane-"));
 const cacheRoot = join(workspace, "cache");
+// A server retains its workspace until both provider and transport cleanup
+// have positively completed, even if later lineage observation is empty.
+const pendingServers = new Set();
+let kunaVersion;
 
 const FIXTURE_SOURCE = String.raw`
 #include <stdio.h>
@@ -139,6 +143,8 @@ const sampleTree = async (root) => {
 
 /** One MCP server with a selected provider, sampled while it runs. */
 const openServer = async (provider) => {
+  const owner = Symbol(provider.id);
+  pendingServers.add(owner);
   const transport = new StdioClientTransport({
     command: process.execPath,
     args: [entrypoint, "--mcp"],
@@ -159,11 +165,10 @@ const openServer = async (provider) => {
     stderr: "ignore",
   });
   const client = new Client({ name: "rea-real-kuna", version: "1" });
-  await client.connect(transport);
-  const pid = transport.pid;
   let peakKiB = 0;
   const cpu = new Map();
-  const sample = async () => {
+  const sampleOnce = async () => {
+    const pid = transport.pid;
     if (pid === null) return;
     const observed = await sampleTree(pid).catch(() => undefined);
     if (observed === undefined) return;
@@ -171,7 +176,14 @@ const openServer = async (provider) => {
     for (const [member, seconds] of observed.cpu)
       cpu.set(member, Math.max(cpu.get(member) ?? 0, seconds));
   };
-  const timer = setInterval(() => void sample(), 200);
+  let sampling;
+  const sample = () => {
+    sampling ??= sampleOnce().finally(() => {
+      sampling = undefined;
+    });
+    return sampling;
+  };
+  let timer;
   const call = async (name, arguments_ = {}) => {
     const started = performance.now();
     const result = await client.callTool(
@@ -187,31 +199,46 @@ const openServer = async (provider) => {
       structured: result.structuredContent,
     };
   };
-  const close = async () => {
-    await sample();
+  const close = async (priorFailure) => {
     clearInterval(timer);
-    try {
-      await call("close_binary");
-    } finally {
-      await client.close();
-      await transport.close();
-    }
+    await sample();
+    const failures = priorFailure === undefined ? [] : [priorFailure];
+    // Attempt each cleanup independently. Disposing a transport is not proof
+    // that close_binary released retained provider processes or files.
+    for (const cleanup of [
+      () => call("close_binary"),
+      () => client.close(),
+      () => transport.close(),
+    ])
+      try {
+        await cleanup();
+      } catch (cause) {
+        failures.push(cause);
+      }
+    // A prior verification error must survive successful cleanup. Only the
+    // cleanup failures retain this server's files.
+    if (failures.length > (priorFailure === undefined ? 0 : 1))
+      throw new AggregateError(
+        failures,
+        "MCP server cleanup was not confirmed",
+      );
+    pendingServers.delete(owner);
+    if (priorFailure !== undefined) throw priorFailure;
     return {
       peak_rss_mib: Math.round(peakKiB / 1024),
       cpu_seconds: [...cpu.values()].reduce((total, value) => total + value, 0),
     };
   };
   try {
+    await client.connect(transport);
+    timer = setInterval(() => void sample(), 200);
     await call("open_binary", {
       path: provider.path,
       provider_id: provider.id,
     });
   } catch (cause) {
-    // A rejected target still owns the client, transport and sampler.
-    clearInterval(timer);
-    await client.close().catch(() => undefined);
-    await transport.close().catch(() => undefined);
-    throw cause;
+    // A rejected connection/target still owns its transport and workspace.
+    await close(cause);
   }
   return { call, close };
 };
@@ -231,6 +258,7 @@ const verifyKuna = async (path, functions) => {
     .digest("hex");
   const cold = await openServer({ path, id: "kuna" });
   const answers = [];
+  let coldFailure;
   try {
     const listed = await cold.call("list_procedures");
     for (const [name, address] of functions) {
@@ -309,13 +337,16 @@ const verifyKuna = async (path, functions) => {
         ...quality(exported.value),
       });
     }
+  } catch (cause) {
+    coldFailure = cause;
   } finally {
-    answers.push({ cold_server: await cold.close() });
+    answers.push({ cold_server: await cold.close(coldFailure) });
   }
   const staging = await readdir(join(cacheRoot, "staging"));
   assert.deepEqual(staging, [], "Kuna staging survived a completed export");
   // A fresh server reuses the export immediately.
   const warm = await openServer({ path, id: "kuna" });
+  let warmFailure;
   try {
     for (const answer of answers)
       if (answer.address !== undefined) {
@@ -328,8 +359,10 @@ const verifyKuna = async (path, functions) => {
         );
         answer.warm_fresh_server_ms = reread.ms;
       }
+  } catch (cause) {
+    warmFailure = cause;
   } finally {
-    answers.push({ warm_server: await warm.close() });
+    answers.push({ warm_server: await warm.close(warmFailure) });
   }
   return answers;
 };
@@ -338,6 +371,7 @@ const verifyKuna = async (path, functions) => {
 const measureGhidra = async (path, functions) => {
   const server = await openServer({ path, id: "ghidra" });
   const answers = [];
+  let failure;
   try {
     for (const [name, address] of functions) {
       const cold = await server.call("procedure_pseudo_code", {
@@ -355,8 +389,10 @@ const measureGhidra = async (path, functions) => {
         ...quality(cold.value),
       });
     }
+  } catch (cause) {
+    failure = cause;
   } finally {
-    answers.push({ server: await server.close() });
+    answers.push({ server: await server.close(failure) });
   }
   return answers;
 };
@@ -388,7 +424,15 @@ const verifyVariant = async ({ flavor, directory, symbols }, variant) => {
 };
 
 const results = [];
+const failures = [];
 try {
+  kunaVersion = (
+    await execFileOutput(kuna, ["--version"], {
+      timeout: 10_000,
+      maxBuffer: 64 * 1024,
+      stopSignal: "SIGTERM",
+    })
+  ).stdout.trim();
   let callerPairs = [];
   if (values.targets !== undefined) {
     const names = (values.functions ?? "").split(",").filter(Boolean);
@@ -434,13 +478,34 @@ try {
   for (const pair of callerPairs)
     for (const variant of [pair.file, `${pair.file}.stripped`])
       results.push(await verifyVariant(pair, variant));
-} finally {
-  await rm(workspace, { recursive: true, force: true });
+} catch (cause) {
+  failures.push(cause);
 }
+// Failure paths need the same process evidence as successful verification.
 const lifecycle = await completeVerifierRun(run);
-assertVerifierCleanupComplete(lifecycle.process_lineage);
+let cleanupComplete = pendingServers.size === 0;
+try {
+  assertVerifierCleanupComplete(lifecycle.process_lineage);
+} catch (cause) {
+  cleanupComplete = false;
+  failures.push(cause);
+}
+if (cleanupComplete)
+  try {
+    await rm(workspace, { recursive: true, force: true });
+  } catch (cause) {
+    cleanupComplete = false;
+    failures.push(cause);
+  }
+const describeFailure = (cause) => ({
+  message: cause instanceof Error ? cause.message : String(cause),
+  ...(cause instanceof AggregateError
+    ? { causes: cause.errors.map(describeFailure) }
+    : {}),
+});
 const report = {
   lane: "kuna",
+  status: failures.length === 0 ? "passed" : "failed",
   kuna_version: kunaVersion,
   platform: `${process.platform}/${process.arch}`,
   verifier_run: lifecycle,
@@ -451,8 +516,15 @@ const report = {
     "Quality signals count gotos and warning markers only; they are not a correctness judgement and not a DecBench reproduction.",
     "Response bytes are UTF-8 serialization sizes. Model token usage, tool schema costs and agent task completion were not measured.",
   ],
-  cleanup: "complete",
+  cleanup: cleanupComplete ? "complete" : "incomplete",
+  ...(cleanupComplete ? {} : { retained_workspace: workspace }),
+  ...(failures.length === 0 ? {} : { failures: failures.map(describeFailure) }),
 };
 if (values.report !== undefined)
   await writeFile(values.report, `${JSON.stringify(report, null, 2)}\n`);
 process.stdout.write(`${JSON.stringify(report)}\n`);
+if (failures.length > 0)
+  throw new AggregateError(
+    failures,
+    "Kuna verification failed; see the report for cleanup evidence",
+  );

src/kuna/KunaProjectReader.ts

@@ -5,10 +5,12 @@ import { join } from "node:path";
 import type { AnalysisError } from "../domain/analysisErrorBase.js";
 import {
   AnalysisCapabilityUnavailableError,
+  AnalysisCancelledError,
   AnalysisResourceConstraintError,
 } from "../domain/analysisErrorCore.js";
 import { ProviderAdapterError } from "../domain/providerAdapterError.js";
 import { err, ok, type Result } from "../domain/result.js";
+import { ABORTED, waitForAbortable } from "../process/AbortablePromise.js";
 import type { KunaProject } from "./KunaProjectCache.js";
 import {
   scanKunaAssembly,
@@ -69,7 +71,13 @@ export class KunaProjectReader {
   #blocks: ReadonlyMap<string, KunaProjectFunction> | undefined;
   #assembly: Promise<Result<KunaAssemblyIndex, string>> | undefined;
 
-  constructor(readonly project: KunaProject) {}
+  constructor(
+    readonly project: KunaProject,
+    readonly lifetime: {
+      readonly signal: AbortSignal;
+      readonly own: <T>(work: Promise<T>) => Promise<T>;
+    },
+  ) {}
 
   async pseudocode(
     address: string,
@@ -119,8 +127,9 @@ export class KunaProjectReader {
   async assembly(
     address: string,
     operation: string,
+    signal?: AbortSignal,
   ): Promise<Result<KunaProjectText, AnalysisError>> {
-    const index = await this.#index(operation);
+    const index = await this.#index(operation, signal);
     if (!index.ok) return index;
     const span = index.value.code.get(address);
     if (span === undefined)
@@ -155,20 +164,26 @@ export class KunaProjectReader {
   async dataName(
     address: string,
     operation: string,
+    signal?: AbortSignal,
   ): Promise<Result<string | null, AnalysisError>> {
-    const index = await this.#index(operation);
+    const index = await this.#index(operation, signal);
     return index.ok ? ok(index.value.data.get(address) ?? null) : index;
   }
 
   async #index(
     operation: string,
+    signal: AbortSignal | undefined,
   ): Promise<Result<KunaAssemblyIndex, AnalysisError>> {
-    this.#assembly ??= scanKunaAssembly(
-      createReadStream(
-        join(this.project.directory, this.project.manifest.files.asm.path),
+    this.#assembly ??= this.lifetime.own(
+      scanKunaAssembly(
+        createReadStream(
+          join(this.project.directory, this.project.manifest.files.asm.path),
+          { signal: this.lifetime.signal },
+        ),
       ),
     );
-    const index = await this.#assembly;
+    const index = await waitForAbortable(this.#assembly, signal);
+    if (index === ABORTED) return err(new AnalysisCancelledError(operation));
     return index.ok
       ? index
       : err(

src/kuna/KunaSessionClient.ts

@@ -152,10 +152,8 @@ export class KunaSessionClient implements AnalysisClient {
   readonly #launcher: KunaProcessLauncher | undefined;
   /** Owns background work: aborting it cancels this session's export interest. */
   readonly #session = new AbortController();
-  /** Requests keep their workspaces and process cleanup owned until they settle. */
-  readonly #requests = new Set<
-    Promise<Result<AnalysisExecution, AnalysisError>>
-  >();
+  /** Requests and shared readers retain their resources until their work settles. */
+  readonly #active = new Set<Promise<unknown>>();
   /** Single-function runs that could not be proven stopped, with their retry owners. */
   readonly #retained: {
     readonly error: ProviderCleanupError;
@@ -216,19 +214,22 @@ export class KunaSessionClient implements AnalysisClient {
           "This Kuna session is closed.",
         ),
       );
-    const request = this.#execute(operation, parameters, {
-      ...options,
-      signal:
-        options.signal === undefined
-          ? this.#session.signal
-          : AbortSignal.any([this.#session.signal, options.signal]),
-    });
-    this.#requests.add(request);
-    try {
-      return await request;
-    } finally {
-      this.#requests.delete(request);
-    }
+    return this.#own(
+      this.#execute(operation, parameters, {
+        ...options,
+        signal:
+          options.signal === undefined
+            ? this.#session.signal
+            : AbortSignal.any([this.#session.signal, options.signal]),
+      }),
+    );
+  }
+
+  #own<T>(work: Promise<T>): Promise<T> {
+    this.#active.add(work);
+    const release = () => this.#active.delete(work);
+    void work.then(release, release);
+    return work;
   }
 
   async #execute(
@@ -320,7 +321,8 @@ export class KunaSessionClient implements AnalysisClient {
     }
     // Cancellation starts process shutdown; its completion proves whether it
     // is safe to remove private copies and exposes any retained cleanup owner.
-    await Promise.allSettled(this.#requests);
+    // An admitted request may create a shared reader after shutdown begins.
+    while (this.#active.size > 0) await Promise.allSettled(this.#active);
     await this.#project;
     await this.#inventory;
     const failures: ProviderCleanupError[] = [];
@@ -541,7 +543,7 @@ export class KunaSessionClient implements AnalysisClient {
       }
       case "analyze_function":
         return this.#withProject(operation, installation, options, (reader) =>
-          this.#dossier(operation, entry.value, reader),
+          this.#dossier(operation, entry.value, reader, options.signal),
         );
     }
   }
@@ -618,7 +620,11 @@ export class KunaSessionClient implements AnalysisClient {
       installation,
       options,
       async (reader) => {
-        const name = await reader.dataName(address, "address_name");
+        const name = await reader.dataName(
+          address,
+          "address_name",
+          options.signal,
+        );
         if (!name.ok) return name;
         return ok({
           result: name.value,
@@ -697,7 +703,11 @@ export class KunaSessionClient implements AnalysisClient {
       installation,
       options,
       async (reader) => {
-        const text = await reader.assembly(entry.address, operation);
+        const text = await reader.assembly(
+          entry.address,
+          operation,
+          options.signal,
+        );
         return text.ok
           ? ok({
               ...text.value,
@@ -712,8 +722,9 @@ export class KunaSessionClient implements AnalysisClient {
     operation: string,
     entry: KunaInventoryFunction,
     reader: KunaProjectReader,
+    signal: AbortSignal | undefined,
   ): Promise<Result<Answer, AnalysisError>> {
-    const assembly = await reader.assembly(entry.address, operation);
+    const assembly = await reader.assembly(entry.address, operation, signal);
     if (!assembly.ok) return assembly;
     const pseudocode = await reader.pseudocode(entry.address, operation);
     if (!pseudocode.ok) return pseudocode;
@@ -914,7 +925,10 @@ export class KunaSessionClient implements AnalysisClient {
   #readerFor(project: KunaProject): KunaProjectReader {
     return this.#reader?.project === project
       ? this.#reader
-      : new KunaProjectReader(project);
+      : new KunaProjectReader(project, {
+          signal: this.#session.signal,
+          own: (work) => this.#own(work),
+        });
   }
 
   /** The export's failure once it has settled; undefined while it builds or after success. */

tests/boundary/providers/kuna/kunaReaderCancellation.test.ts

@@ -0,0 +1,130 @@
+import { readFile, readdir } from "node:fs/promises";
+import { join } from "node:path";
+import { addAbortSignal, PassThrough } from "node:stream";
+
+import { afterEach, expect, it as test, vi } from "vitest";
+
+import { run, session } from "./kunaSessionFixture.js";
+
+const held = vi.hoisted(() => ({
+  path: "",
+  streams: [] as PassThrough[],
+}));
+
+vi.mock("node:fs", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs")>();
+  return {
+    ...actual,
+    createReadStream: (...args: Parameters<typeof actual.createReadStream>) => {
+      if (String(args[0]) !== held.path)
+        return actual.createReadStream(...args);
+      const stream = new PassThrough();
+      const options = args[1];
+      if (typeof options === "object" && options?.signal != null)
+        addAbortSignal(options.signal, stream);
+      held.streams.push(stream);
+      return stream;
+    },
+  };
+});
+
+afterEach(() => {
+  held.path = "";
+  for (const stream of held.streams) stream.destroy();
+  held.streams = [];
+});
+
+const ready = async () => {
+  const { cacheRoot, open } = await session("rea-kuna-reader-cancel-");
+  const client = open();
+  await expect
+    .poll(
+      async () =>
+        (await run(client, "procedure_pseudo_code", { procedure: "main" }))
+          .rawResult,
+    )
+    .toMatchObject({ source: { kind: "project-export" } });
+  const [key] = await readdir(join(cacheRoot, "projects"));
+  held.path = join(cacheRoot, "projects", key ?? "", "project", "program.asm");
+  return { client, bytes: await readFile(held.path) };
+};
+
+const it = test.skipIf(process.platform === "win32");
+
+it.each([
+  { operation: "procedure_assembly", parameters: { procedure: "main" } },
+  { operation: "analyze_function", parameters: { procedure: "main" } },
+  { operation: "address_name", parameters: { address: "0x3010" } },
+] as const)(
+  "cancels $operation without cancelling another index reader",
+  async ({ operation, parameters }) => {
+    const { client, bytes } = await ready();
+    const controller = new AbortController();
+    let finished = false;
+    const cancelled = client
+      .execute(operation, parameters, { signal: controller.signal })
+      .then((value) => {
+        finished = true;
+        return value;
+      });
+    const other = run(client, "procedure_assembly", { procedure: "parse" });
+    try {
+      await expect.poll(() => held.streams.length).toBe(1);
+      controller.abort();
+      await expect.poll(() => finished, { timeout: 1000 }).toBe(true);
+      expect(await cancelled).toMatchObject({
+        ok: false,
+        error: { _tag: "AnalysisCancelledError" },
+      });
+      expect(held.streams[0]?.destroyed).toBe(false);
+      expect((await run(client, "health")).result).toBeDefined();
+      held.streams[0]?.end(bytes);
+      expect((await other).result).toContain("parse:  ; 0x1040");
+      expect(held.streams).toHaveLength(1);
+    } finally {
+      held.streams[0]?.end(bytes);
+      await Promise.allSettled([cancelled, other]);
+      expect(await client.close()).toEqual({ ok: true, value: null });
+    }
+  },
+);
+
+it("close aborts and drains the shared scan after its only caller cancels", async () => {
+  const { client } = await ready();
+  const controller = new AbortController();
+  let result: unknown;
+  const request = client
+    .execute(
+      "procedure_assembly",
+      { procedure: "main" },
+      { signal: controller.signal },
+    )
+    .then((value) => {
+      result = value;
+      return value;
+    });
+  let closing: ReturnType<typeof client.close> | undefined;
+  try {
+    await expect.poll(() => held.streams.length).toBe(1);
+    controller.abort();
+    await expect
+      .poll(() => result, { timeout: 1000 })
+      .toMatchObject({ ok: false });
+    expect(held.streams[0]?.destroyed).toBe(false);
+    let closed: unknown;
+    closing = client.close().then((value) => {
+      closed = value;
+      return value;
+    });
+    await expect
+      .poll(() => closed, { timeout: 1000 })
+      .toEqual({ ok: true, value: null });
+    expect(held.streams[0]?.destroyed).toBe(true);
+    expect(held.streams[0]?.closed).toBe(true);
+  } finally {
+    held.streams[0]?.destroy();
+    await request;
+    await closing;
+    await client.close();
+  }
+});

tests/boundary/providers/kuna/kunaVerifier.test.ts

@@ -1,6 +1,6 @@
 import { spawnSync } from "node:child_process";
-import { mkdir, writeFile } from "node:fs/promises";
-import { join } from "node:path";
+import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
+import { join, resolve } from "node:path";
 
 import { expect, it as test } from "vitest";
 
@@ -71,3 +71,115 @@ it.each([
     ]);
   },
 );
+
+it.each([
+  { rejectClose: false, rejectRead: true },
+  { rejectClose: true, rejectRead: false },
+  { rejectClose: true, rejectRead: true },
+])(
+  "reports failure-path lineage (cleanup rejection: $rejectClose, read rejection: $rejectRead)",
+  async ({ rejectClose, rejectRead }) => {
+    const root = await createTestTempDirectory("rea-kuna-verifier-cleanup-");
+    const kuna = await writeFakeKuna(root, {
+      functions: [
+        { name: "rea_checksum", addr: "0x1000" },
+        { name: "rea_classify", addr: "0x1040" },
+      ],
+    });
+    const nm = join(root, "nm");
+    // The host compiler creates a real fixture, but provider output and symbol
+    // discovery are simulated to isolate the verifier's cleanup contract.
+    await writeFile(
+      nm,
+      `#!${process.execPath}\nprocess.stdout.write("00001000 T rea_checksum\\n00001040 T rea_classify\\n");\n`,
+      { mode: 0o755 },
+    );
+    const proxy = join(root, "proxy.mjs");
+    const cacheRecord = join(root, "cache-path");
+    await writeFile(
+      proxy,
+      `
+import { spawn } from "node:child_process";
+import { mkdir, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+import { createInterface } from "node:readline";
+const backend = spawn(process.execPath, [${JSON.stringify(resolve("scripts/rea.mjs"))}, "--mcp"], { stdio: ["pipe", "pipe", "ignore"] });
+const exited = new Promise((resolve, reject) => { backend.once("error", reject); backend.once("exit", resolve); });
+backend.stdout.pipe(process.stdout);
+const input = createInterface({ input: process.stdin });
+for await (const line of input) {
+  const message = JSON.parse(line);
+  const name = message.params?.name;
+  if (name === "close_binary" && ${rejectClose}) {
+    const marker = join(process.env.REA_KUNA_CACHE_DIR, "staging", "retained-marker");
+    await mkdir(marker, { recursive: true });
+    await writeFile(join(marker, "evidence"), "retained cleanup owner");
+    await writeFile(${JSON.stringify(cacheRecord)}, process.env.REA_KUNA_CACHE_DIR);
+    process.stdout.write(JSON.stringify({ jsonrpc: "2.0", id: message.id, result: { isError: true, content: [{ type: "text", text: "injected cleanup rejection" }] } }) + "\\n");
+  } else if (name === "list_procedures" && ${rejectRead}) {
+    // A verification failure with successful provider cleanup must still
+    // produce lineage evidence; it need not retain an otherwise clean root.
+    process.stdout.write(JSON.stringify({ jsonrpc: "2.0", id: message.id, result: { isError: true, content: [{ type: "text", text: "injected verification failure" }] } }) + "\\n");
+  } else backend.stdin.write(line + "\\n");
+}
+backend.stdin.end();
+await exited;
+`,
+    );
+    const reportPath = join(root, "report.json");
+    const result = spawnSync(
+      process.execPath,
+      [
+        "scripts/verify-real-kuna.mjs",
+        "--entrypoint",
+        proxy,
+        "--report",
+        reportPath,
+      ],
+      {
+        encoding: "utf8",
+        timeout: 25_000,
+        env: {
+          PATH: process.env.PATH,
+          HOME: root,
+          REA_KUNA_COMMAND: kuna,
+          REA_NM: nm,
+        },
+      },
+    );
+    expect(result.error).toBeUndefined();
+    expect(result.status).toBe(1);
+    const report = JSON.parse(await readFile(reportPath, "utf8"));
+    expect(report).toMatchObject({
+      status: "failed",
+      cleanup: rejectClose ? "incomplete" : "complete",
+      verifier_run: {
+        process_lineage: { status: "verified", descendants: [] },
+      },
+    });
+    expect(JSON.stringify(report.failures)).toContain(
+      rejectClose
+        ? "injected cleanup rejection"
+        : "injected verification failure",
+    );
+    if (rejectRead)
+      expect(JSON.stringify(report.failures)).toContain(
+        "injected verification failure",
+      );
+    if (rejectClose) {
+      const cache = await readFile(cacheRecord, "utf8");
+      expect(cache).toBe(join(report.retained_workspace, "cache"));
+      try {
+        expect(
+          await readFile(
+            join(cache, "staging", "retained-marker", "evidence"),
+            "utf8",
+          ),
+        ).toBe("retained cleanup owner");
+      } finally {
+        // The test owns this injected file; production intentionally retains it.
+        await rm(report.retained_workspace, { recursive: true, force: true });
+      }
+    } else expect(report.retained_workspace).toBeUndefined();
+  },
+);