Add repeated Windows RAM recovery and CLI runtime benchmark

.github/workflows/windows-memory-benchmark.yml

@@ -0,0 +1,55 @@
+name: Windows memory benchmark
+
+on:
+  workflow_dispatch:
+  push:
+    branches: ['test/windows-memory-benchmark']
+    paths:
+      - 'validation/windows_positive/**'
+      - '.github/workflows/windows-memory-benchmark.yml'
+
+permissions:
+  contents: read
+
+concurrency:
+  group: windows-memory-benchmark-${{ github.ref }}
+  cancel-in-progress: false
+
+jobs:
+  capture:
+    name: Capture ${{ matrix.replica }}
+    runs-on: windows-2022
+    timeout-minutes: 25
+    strategy:
+      fail-fast: false
+      max-parallel: 3
+      matrix:
+        replica: [1, 2, 3]
+    env:
+      PYTHONUTF8: '1'
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 # v7.1.0
+        with:
+          python-version: '3.12'
+      - name: Install plugin and pinned SDK
+        run: python -m pip install -e . -r validation/windows_positive/requirements.txt
+      - name: Acquire real RAM and benchmark three scans
+        shell: pwsh
+        run: |
+          python validation/windows_positive/run.py --workdir "$env:RUNNER_TEMP/mcp-benchmark-work" --reportdir "$env:RUNNER_TEMP/mcp-benchmark-report" --tool-calls 64 --scan-repeats 3
+      - name: Extract aggregate measurements only
+        if: always()
+        shell: pwsh
+        run: |
+          python validation/windows_positive/benchmark.py summarize --reportdir "$env:RUNNER_TEMP/mcp-benchmark-report" --replica ${{ matrix.replica }} --output "$env:RUNNER_TEMP/benchmark-${{ matrix.replica }}.json"
+      - name: Upload aggregate measurements only
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
+        with:
+          name: windows-benchmark-${{ github.run_id }}-${{ matrix.replica }}
+          path: ${{ runner.temp }}/benchmark-${{ matrix.replica }}.json
+          if-no-files-found: warn
+          retention-days: 30

tests/test_benchmark_reporting.py

@@ -0,0 +1,155 @@
+"""Benchmark publication must keep denominators, failures, and content boundaries."""
+
+import copy
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+SPEC = importlib.util.spec_from_file_location(
+    "benchmark", Path(__file__).resolve().parents[1] / "validation/windows_positive/benchmark.py"
+)
+benchmark = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(benchmark)
+
+
+def report_files(tmp_path):
+    manifest = {
+        "status": "passed",
+        "commit": "test-commit",
+        "platform": "Windows-2022Server",
+        "python": "3.12",
+        "mcp_sdk": "1.26.0",
+        "volatility": "2.28.2",
+        "logical_cpus": 2,
+        "tool_calls_per_case": 64,
+        "scan_repeats": 3,
+        "raw_memory_deleted": True,
+        "image_hash_unchanged_after_analysis": True,
+        "repeated_exports_identical": True,
+        "image_sha256": "1" * 64,
+        "physical_ram_bytes": 8 * 1024**3,
+        "image_bytes": 11 * 1024**3,
+        "capture_started_ns": 1_000_000_000,
+        "cases": [
+            {"mode": mode, "ready_ns": 900_000_000, "client_pid": 12345}
+            for mode in ("natural", "retained")
+        ],
+        "timings_seconds": {
+            "acquisition": 10.0,
+            "symbols_and_info": 20.0,
+            "native_byte_verification": 2.0,
+            "pipeline_total": 50.0,
+            "plugin_cli_seconds": [4.0, 3.0, 2.0],
+        },
+        "unrelated_secret": "DO_NOT_PUBLISH",
+    }
+    cases = []
+    for mode in ("natural", "retained"):
+        case = dict.fromkeys(benchmark.CASE_FIELDS, 0)
+        case.update(
+            mode=mode,
+            expected_logical_messages=133,
+            expected_tool_pairs=64,
+            distinct_expected_messages_found=10,
+            complete_tool_pairs_any_endpoint=2,
+            artifact_records=10,
+            records_matching_logged_traffic=10,
+            per_pid={"12345": {"complete_tool_pairs": 2}},
+            recovered_message="DO_NOT_PUBLISH",
+        )
+        cases.append(case)
+    scans = {
+        str(pid): {
+            "error": None,
+            "limits_hit": [],
+            "skipped_reason": None,
+            "anchor_limit_hits": 0,
+            "candidate_limit_hits": 0,
+            "invalid_regions": 0,
+            "displayed_records": 5,
+            "candidate_records": 5,
+            "bytes_read": 1024,
+            "unreadable_bytes": 2048,
+            "regions_scanned": 2,
+        }
+        for pid in range(4)
+    }
+    for name, data in (
+        ("manifest.json", manifest),
+        ("scan-summaries.json", scans),
+        (
+            "comparison.json",
+            {"cases": cases, "artifact_records_byte_verified": 20, "artifact_bytes_verified": 500},
+        ),
+    ):
+        (tmp_path / name).write_text(json.dumps(data))
+    return manifest
+
+
+def test_summary_excludes_raw_content_and_identifiers(tmp_path):
+    report_files(tmp_path)
+    summary = benchmark.summarize_run(tmp_path, 1)
+    encoded = json.dumps(summary)
+    assert "DO_NOT_PUBLISH" not in encoded and "12345" not in encoded
+    assert summary["scan_totals"]["bytes_read"] == 4096
+    assert summary["cases"]["natural"]["expected_logical_messages"] == 133
+    assert summary["checks"]["raw_memory_deleted"]
+
+
+def test_aggregation_uses_capture_denominators_not_scan_repeats(tmp_path):
+    report_files(tmp_path)
+    runs = []
+    for replica, found in enumerate([10, 12, 14], 1):
+        run = benchmark.summarize_run(tmp_path, replica)
+        run["capture_sha256"] = str(replica) * 64
+        run["cases"]["natural"]["distinct_expected_messages_found"] = found
+        runs.append(run)
+    combined = benchmark.aggregate_runs(runs)
+    natural = combined["modes"]["natural"]
+    assert natural["totals"]["expected_logical_messages"] == 399
+    assert natural["totals"]["distinct_expected_messages_found"] == 36
+    assert natural["logical_message_coverage"] == 36 / 399
+    assert natural["messages_found_per_capture"]["mean"] == 12
+    assert combined["plugin_cli_seconds"]["n"] == 9
+    with pytest.raises(ValueError, match="exactly replicas"):
+        benchmark.aggregate_runs(runs[:2])
+    bad = copy.deepcopy(runs)
+    bad[2]["status"] = "failed"
+    with pytest.raises(ValueError, match="failed capture"):
+        benchmark.aggregate_runs(bad)
+    bad = copy.deepcopy(runs)
+    bad[2]["capture_sha256"] = bad[0]["capture_sha256"]
+    with pytest.raises(ValueError, match="independently"):
+        benchmark.aggregate_runs(bad)
+    bad = copy.deepcopy(runs)
+    bad[2]["tool_calls_per_case"] = 8
+    with pytest.raises(ValueError, match="Incompatible"):
+        benchmark.aggregate_runs(bad)
+
+
+def test_failed_run_remains_a_failure_without_partial_metrics(tmp_path):
+    manifest = report_files(tmp_path)
+    manifest["status"] = "failed"
+    (tmp_path / "manifest.json").write_text(json.dumps(manifest))
+    summary = benchmark.summarize_run(tmp_path, 1)
+    assert summary["status"] == "failed"
+    assert "cases" not in summary
+
+
+@pytest.mark.parametrize("problem", ["cleanup", "record_count", "timings"])
+def test_incomplete_validation_cannot_be_published_as_success(tmp_path, problem):
+    manifest = report_files(tmp_path)
+    if problem == "cleanup":
+        manifest["raw_memory_deleted"] = False
+    elif problem == "timings":
+        manifest["timings_seconds"]["plugin_cli_seconds"] = [1.0]
+    else:
+        path = tmp_path / "comparison.json"
+        data = json.loads(path.read_text())
+        data["artifact_records_byte_verified"] = 999
+        path.write_text(json.dumps(data))
+    (tmp_path / "manifest.json").write_text(json.dumps(manifest))
+    with pytest.raises(ValueError):
+        benchmark.summarize_run(tmp_path, 1)

validation/windows_positive/BENCHMARK.md

@@ -0,0 +1,56 @@
+# Repeated Windows RAM benchmark
+
+The benchmark uses three independently acquired Windows Server 2022 VM images.
+Each VM runs two official Python MCP SDK stdio client/server pairs: `natural`
+(ground-truth logging without a separate retained-frame list) and `retained`
+(the same logging plus a retained-frame list and one deliberately unsent control).
+
+Each pair completes 64 echo tool calls, cycling through the existing eight payload
+patterns with a unique sequence and request ID. Initialization, notification, and
+tool discovery bring the ground truth to 133 distinct logical messages per mode
+per capture. The three-capture denominator is 399 messages and 192 tool-call pairs
+per mode. These sample counts are fixed before execution.
+
+Each capture is analyzed three times by separate MCPTrace CLI processes. All four
+selected endpoint PIDs are scanned together. Exported JSONL must be byte-identical
+across the three scans. One set of exports is independently re-read through native
+Volatility Windows process layers to check bytes, SHA-256, addresses, and VAD bounds.
+The original RAM image is hashed before and after analysis, then deleted.
+
+CLI timing includes process startup, Volatility automagic, scanning, JSON rendering,
+and JSONL export. It excludes acquisition, initial `windows.info`/symbol preparation,
+export comparison, and native byte verification. The image has already been hashed
+and inspected before the first scan; no scan is claimed to be a cold-cache result.
+Repeated scans measure timing/repeatability, not additional independent recovery
+samples. Runtime covers four selected processes, not all processes in the RAM image.
+
+## Reproduce
+
+Use a fresh administrative Windows VM for each replica. Install the repository and
+`validation/windows_positive/requirements.txt`, then run from the repository root:
+
+```powershell
+python validation/windows_positive/run.py --workdir C:\MCPBenchmark\work --reportdir C:\MCPBenchmark\report --tool-calls 64 --scan-repeats 3
+python validation/windows_positive/benchmark.py summarize --reportdir C:\MCPBenchmark\report --replica 1 --output benchmark-1.json
+```
+
+Repeat in two other fresh VMs using replicas 2 and 3, then aggregate all three files:
+
+```powershell
+python validation/windows_positive/benchmark.py aggregate benchmark-1.json benchmark-2.json benchmark-3.json --output benchmark.json
+```
+
+The GitHub Actions `Windows memory benchmark` workflow performs the three captures.
+It uploads only allowlisted counts, durations, environment/version information, image
+hashes, and validation flags. RAM, recovered messages, ground truth, process IDs,
+and arbitrary log text are excluded. Failed or missing captures cannot be silently
+dropped from the three-capture aggregate.
+
+## Interpretation
+
+Recovery counts deduplicate messages within each capture and combine both endpoints.
+Candidate links remain candidates; unsent controls are counted separately from logged
+traffic. Mean/min/max describe only these three captures. Logging changes memory,
+and retained frames are a favorable control. No general recall, detection precision,
+latency/pressure survival curve, or Windows 11/Linux benchmark is established by
+this Windows Server 2022 SDK/stdio experiment.

validation/windows_positive/README.md

@@ -57,3 +57,7 @@ GitHub Actions의 `Windows positive memory validation` 워크플로도 같은
 전체 실험은 실행 가능성을 확인한 것으로 기록되므로 **반드시 natural/retained 수치를 나눠 읽어야 합니다.**
 실제로 수행한 Windows Server 2022 캡처와 조건별 수치는
 [양성 검증 보고서](../../docs/windows-positive-memory-validation.md)에 기록했습니다.
+
+`--tool-calls`로 조건별 도구 호출 수를, `--scan-repeats`로 동일 이미지의 CLI 분석 반복 수를
+지정할 수 있습니다. 기본값은 기존 검증과 같은 8회/1회입니다. 별도 VM 3개에서 64회 호출과
+3회 분석을 수행하는 [반복 벤치마크 절차](BENCHMARK.md)도 제공합니다.

validation/windows_positive/benchmark.py

@@ -0,0 +1,214 @@
+"""Publish allowlisted measurements and aggregate three independent Windows captures."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import statistics
+from pathlib import Path
+
+CASE_FIELDS = (
+    "expected_logical_messages",
+    "distinct_expected_messages_found",
+    "expected_tool_pairs",
+    "complete_tool_pairs_any_endpoint",
+    "artifact_records",
+    "records_matching_logged_traffic",
+    "distinct_unsent_controls_found",
+    "records_matching_unsent_controls",
+    "unmatched_artifact_records",
+    "evaluated_candidate_links",
+    "incorrect_candidate_links",
+    "unverifiable_candidate_links",
+)
+ENVIRONMENT_FIELDS = ("platform", "python", "mcp_sdk", "volatility", "logical_cpus")
+
+
+def read_json(path):
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def summarize_run(report, replica):
+    """Exclude recovered content, ground truth, PIDs, paths, and arbitrary log text."""
+    path = report / "manifest.json"
+    manifest = read_json(path) if path.exists() else {}
+    output = {
+        "schema_version": 1,
+        "replica": replica,
+        "status": "passed" if manifest.get("status") == "passed" else "failed",
+        "commit": manifest.get("commit"),
+        "environment": {key: manifest.get(key) for key in ENVIRONMENT_FIELDS},
+        "tool_calls_per_case": manifest.get("tool_calls_per_case"),
+        "scan_repeats": manifest.get("scan_repeats"),
+    }
+    if output["status"] != "passed":
+        return output
+    for key in (
+        "raw_memory_deleted",
+        "image_hash_unchanged_after_analysis",
+        "repeated_exports_identical",
+    ):
+        if manifest.get(key) is not True:
+            raise ValueError(f"Completed capture is missing validation: {key}")
+    comparison = read_json(report / "comparison.json")
+    scans = read_json(report / "scan-summaries.json")
+    if len(scans) != 4 or any(
+        s["error"]
+        or s["limits_hit"]
+        or s["skipped_reason"]
+        or s["anchor_limit_hits"]
+        or s["candidate_limit_hits"]
+        or s["invalid_regions"]
+        for s in scans.values()
+    ):
+        raise ValueError("Benchmark requires four complete endpoint scans")
+    cases = {}
+    for case in comparison["cases"]:
+        counts = {key: case[key] for key in CASE_FIELDS}
+        if any(type(value) is not int or value < 0 for value in counts.values()):
+            raise ValueError("Case measurements must be nonnegative integer counts")
+        if (
+            counts["expected_tool_pairs"] != output["tool_calls_per_case"]
+            or counts["expected_logical_messages"] != 5 + 2 * output["tool_calls_per_case"]
+            or counts["distinct_expected_messages_found"] > counts["expected_logical_messages"]
+            or counts["complete_tool_pairs_any_endpoint"] > counts["expected_tool_pairs"]
+            or counts["incorrect_candidate_links"]
+        ):
+            raise ValueError("Inconsistent ground truth or incorrect candidate links")
+        counts["best_single_pid_tool_pairs"] = max(
+            p["complete_tool_pairs"] for p in case["per_pid"].values()
+        )
+        if case["mode"] in cases:
+            raise ValueError("Duplicate benchmark mode")
+        cases[case["mode"]] = counts
+    if set(cases) != {"natural", "retained"}:
+        raise ValueError("Both benchmark modes must be present")
+    verified = comparison["artifact_records_byte_verified"]
+    if verified != sum(c["artifact_records"] for c in cases.values()) or verified != sum(
+        s["displayed_records"] for s in scans.values()
+    ):
+        raise ValueError("Verified and exported record counts differ")
+    timings = manifest["timings_seconds"]
+    durations = timings["plugin_cli_seconds"]
+    if len(durations) != output["scan_repeats"] or any(
+        not math.isfinite(s) or s <= 0 for s in durations
+    ):
+        raise ValueError("Missing or invalid CLI timing samples")
+    output.update(
+        cases=cases,
+        capture_sha256=manifest["image_sha256"],
+        physical_ram_bytes=manifest["physical_ram_bytes"],
+        image_logical_bytes=manifest["image_bytes"],
+        artifact_records_byte_verified=verified,
+        artifact_bytes_verified=comparison["artifact_bytes_verified"],
+        scan_totals={
+            key: sum(s[key] for s in scans.values())
+            for key in ("bytes_read", "unreadable_bytes", "regions_scanned", "candidate_records")
+        },
+        timings_seconds={
+            **{
+                key: timings[key]
+                for key in (
+                    "acquisition",
+                    "symbols_and_info",
+                    "native_byte_verification",
+                    "pipeline_total",
+                )
+            },
+            "plugin_cli": durations,
+        },
+        ready_to_capture_seconds={
+            c["mode"]: (manifest["capture_started_ns"] - c["ready_ns"]) / 1e9
+            for c in manifest["cases"]
+        },
+        checks={
+            "full_endpoint_scans": True,
+            "native_bytes_verified": True,
+            "repeated_exports_identical": True,
+            "image_hash_unchanged": True,
+            "raw_memory_deleted": True,
+        },
+    )
+    return output
+
+
+def distribution(values):
+    return {
+        "n": len(values),
+        "mean": statistics.mean(values),
+        "median": statistics.median(values),
+        "min": min(values),
+        "max": max(values),
+    }
+
+
+def aggregate_runs(runs):
+    """Require the full predeclared sample; never silently discard a failed capture."""
+    if len(runs) != 3 or {r["replica"] for r in runs} != {1, 2, 3}:
+        raise ValueError("Require exactly replicas 1, 2, and 3")
+    if any(r["status"] != "passed" for r in runs):
+        raise ValueError("Cannot aggregate a failed capture as a successful benchmark")
+    if len({r["capture_sha256"] for r in runs}) != 3:
+        raise ValueError("Require three independently acquired images")
+    for key in ("schema_version", "commit", "environment", "tool_calls_per_case", "scan_repeats"):
+        if any(r[key] != runs[0][key] for r in runs):
+            raise ValueError(f"Incompatible benchmark conditions: {key}")
+    modes = {}
+    for mode in ("natural", "retained"):
+        cases = [r["cases"][mode] for r in runs]
+        totals = {key: sum(c[key] for c in cases) for key in CASE_FIELDS}
+        modes[mode] = {
+            "totals": totals,
+            "logical_message_coverage": (
+                totals["distinct_expected_messages_found"] / totals["expected_logical_messages"]
+            ),
+            "messages_found_per_capture": distribution(
+                [c["distinct_expected_messages_found"] for c in cases]
+            ),
+            "tool_pairs_per_capture": distribution(
+                [c["complete_tool_pairs_any_endpoint"] for c in cases]
+            ),
+        }
+    return {
+        "schema_version": 1,
+        "independent_captures": 3,
+        "commit": runs[0]["commit"],
+        "environment": runs[0]["environment"],
+        "tool_calls_per_case": runs[0]["tool_calls_per_case"],
+        "scan_repeats_per_capture": runs[0]["scan_repeats"],
+        "modes": modes,
+        "plugin_cli_seconds": distribution(
+            [s for r in runs for s in r["timings_seconds"]["plugin_cli"]]
+        ),
+        "per_capture_cli_median_seconds": distribution(
+            [statistics.median(r["timings_seconds"]["plugin_cli"]) for r in runs]
+        ),
+        "artifact_records_byte_verified": sum(r["artifact_records_byte_verified"] for r in runs),
+        "artifact_bytes_verified": sum(r["artifact_bytes_verified"] for r in runs),
+        "runs": sorted(runs, key=lambda r: r["replica"]),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="action", required=True)
+    summarize = subparsers.add_parser("summarize")
+    summarize.add_argument("--reportdir", type=Path, required=True)
+    summarize.add_argument("--replica", type=int, choices=[1, 2, 3], required=True)
+    summarize.add_argument("--output", type=Path, required=True)
+    aggregate = subparsers.add_parser("aggregate")
+    aggregate.add_argument("inputs", nargs=3, type=Path)
+    aggregate.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    result = (
+        summarize_run(args.reportdir, args.replica)
+        if args.action == "summarize"
+        else aggregate_runs([read_json(path) for path in args.inputs])
+    )
+    args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n", encoding="utf-8")
+    print(f"Wrote {args.output.name}")
+
+
+if __name__ == "__main__":
+    main()

validation/windows_positive/run.py

@@ -8,6 +8,7 @@ import ctypes
 import hashlib
 import importlib.metadata
 import json
+import os
 import platform
 import shutil
 import subprocess
@@ -40,12 +41,14 @@ def write_json(path, value):
 
 
 def command(args, stdout, stderr, timeout=300):
+    started = time.perf_counter()
     with stdout.open("wb") as out, stderr.open("wb") as err:
         completed = subprocess.run(args, stdout=out, stderr=err, timeout=timeout, check=False)
     if completed.returncode:
         raise RuntimeError(
             f"Command failed ({completed.returncode}); see {stdout.name}/{stderr.name}"
         )
+    return time.perf_counter() - started
 
 
 def physical_ram():
@@ -78,14 +81,20 @@ def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--workdir", required=True, type=Path)
     parser.add_argument("--reportdir", required=True, type=Path)
+    parser.add_argument("--tool-calls", type=int, default=8)
+    parser.add_argument("--scan-repeats", type=int, default=1)
     args = parser.parse_args()
+    if not 1 <= args.tool_calls <= 512 or not 1 <= args.scan_repeats <= 5:
+        parser.error("Require 1..512 tool calls and 1..5 scan repeats")
     if sys.platform != "win32":
         parser.error("Run this acquisition harness on Windows as administrator")
     work, report = args.workdir.resolve(), args.reportdir.resolve()
     work.mkdir(parents=True, exist_ok=False)
     report.mkdir(parents=True, exist_ok=False)
     repo = Path(__file__).resolve().parents[2]
+    pipeline_started = time.perf_counter()
     image = work / "memory.raw"
+    timings = {"plugin_cli_seconds": []}
     manifest = {
         "status": "running",
         "stage": "preflight",
@@ -103,6 +112,10 @@ def main():
         "physical_ram_bytes": physical_ram(),
         "free_disk_bytes": shutil.disk_usage(work).free,
         "raw_memory_uploaded": False,
+        "logical_cpus": os.cpu_count(),
+        "tool_calls_per_case": args.tool_calls,
+        "scan_repeats": args.scan_repeats,
+        "timings_seconds": timings,
     }
     children, cases = [], []
 
@@ -144,6 +157,8 @@ def main():
                     str(output),
                     "--mode",
                     mode,
+                    "--tool-calls",
+                    str(args.tool_calls),
                 ],
                 stdout=subprocess.DEVNULL,
                 stderr=error_log,
@@ -160,7 +175,10 @@ def main():
                     raise TimeoutError(f"{output.name}: MCP workload did not become ready")
                 time.sleep(0.1)
             ready = json.loads((output / "ready.json").read_text(encoding="utf-8"))
-            if ready["tool_calls_completed"] != 8 or ready["logged_messages"] != 21:
+            if (
+                ready["tool_calls_completed"] != args.tool_calls
+                or ready["logged_messages"] != 5 + 2 * args.tool_calls
+            ):
                 raise RuntimeError("Incomplete MCP ground truth")
             cases.append(ready)
         pids = [c[key] for c in cases for key in ["client_pid", "server_pid"]]
@@ -169,7 +187,7 @@ def main():
         manifest["cases"] = cases
         manifest["capture_started_ns"] = time.time_ns()
         checkpoint("physical-ram-acquisition")
-        command(
+        timings["acquisition"] = command(
             [str(tool), "acquire", str(image)],
             report / "capture.txt",
             report / "capture.stderr",
@@ -203,7 +221,7 @@ def main():
             str(image),
         ]
         # Windows symbols can be fetched during info; subsequent steps are offline.
-        command(
+        timings["symbols_and_info"] = command(
             common + ["-r", "json", "windows.info.Info"],
             report / "info.json",
             report / "info.stderr",
@@ -225,10 +243,36 @@ def main():
             "--dump",
         ]
         manifest["plugin_command"] = trace_cmd
-        command(trace_cmd, work / "all-findings.json", report / "mcptrace.stderr")
-        checkpoint("byte-verification-and-ground-truth-comparison")
+        timings["plugin_cli_seconds"].append(
+            command(trace_cmd, work / "all-findings.json", report / "mcptrace.stderr")
+        )
         exports = [export / f"mcptrace.pid-{pid}.jsonl" for pid in pids]
+        expected_exports = {path.name: sha256(path) for path in exports}
+        for repeat in range(1, args.scan_repeats):
+            repeated_output = work / f"exports-{repeat + 1}"
+            repeated_output.mkdir()
+            repeated_cmd = trace_cmd.copy()
+            repeated_cmd[repeated_cmd.index("-o") + 1] = str(repeated_output)
+            repeated_cmd[repeated_cmd.index("--save-config") + 1] = str(
+                work / f"mcptrace-config-{repeat + 1}.json"
+            )
+            timings["plugin_cli_seconds"].append(
+                command(
+                    repeated_cmd,
+                    work / f"all-findings-{repeat + 1}.json",
+                    report / f"mcptrace-{repeat + 1}.stderr",
+                )
+            )
+            if any(
+                sha256(repeated_output / name) != digest
+                for name, digest in expected_exports.items()
+            ):
+                raise RuntimeError("Repeated scans of the same image produced different JSONL")
+        manifest["repeated_exports_identical"] = True
+        checkpoint("byte-verification-and-ground-truth-comparison")
+        verification_started = time.perf_counter()
         artifacts, summaries = verify_exports(image, symbols, cache, config, exports)
+        timings["native_byte_verification"] = time.perf_counter() - verification_started
         for summary in summaries.values():
             if (
                 summary["error"]
@@ -311,6 +355,7 @@ def main():
             manifest["raw_memory_deleted"] = True
             write_json(report / "manifest.json", manifest)
     manifest["status"] = "passed"
+    timings["pipeline_total"] = time.perf_counter() - pipeline_started
     checkpoint("completed")
 
 

validation/windows_positive/workload.py

@@ -116,7 +116,8 @@ async def client(args):
                     "last message",
                 ]
                 server_pid = None
-                for sequence, text in enumerate(samples):
+                for sequence in range(args.tool_calls):
+                    text = samples[sequence % len(samples)]
                     result = await session.call_tool(
                         "echo_marker",
                         {"run_id": args.run_id, "sequence": sequence, "text": text},
@@ -159,7 +160,7 @@ async def client(args):
                     "run_id": args.run_id,
                     "client_pid": os.getpid(),
                     "server_pid": server_pid,
-                    "tool_calls_completed": len(samples),
+                    "tool_calls_completed": args.tool_calls,
                     "logged_messages": journal.count,
                     "retained_transport_frames": len(journal.frames),
                     "retained_unsent_controls": len(retained_decoys),
@@ -186,7 +187,10 @@ def main():
     parser.add_argument("--output", type=Path)
     parser.add_argument("--mode", choices=["natural", "retained"], default="natural")
     parser.add_argument("--hold-seconds", type=int, default=900)
+    parser.add_argument("--tool-calls", type=int, default=8)
     args = parser.parse_args()
+    if not 1 <= args.tool_calls <= 512:
+        parser.error("--tool-calls must be between 1 and 512")
     if args.role == "server":
         server(args.run_id)
     elif args.output is None: