Count backtrack exhaustion only after failed search and retain explicit benchmark limits

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

@@ -39,7 +39,7 @@ jobs:
       - 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
+          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 --report-scan-limits
       - name: Extract aggregate measurements only
         if: always()
         shell: pwsh

mcptrace/core.py

@@ -225,8 +225,8 @@ def _carve_buffer(
             anchor = match.start()
             low = bisect.bisect_left(starts, max(0, anchor - maximum + 1))
             high = bisect.bisect_right(starts, anchor)
-            if high - low > 64:
-                stats.candidate_limit_hits += 1
+            limited = high - low > 64
+            if limited:
                 low = high - 64
             found = False
             for index in range(high - 1, low - 1, -1):
@@ -254,6 +254,8 @@ def _carve_buffer(
                     found = True
                     break
             if not found:
+                if limited:
+                    stats.candidate_limit_hits += 1
                 unmatched.append(anchor)
         yield from complete.values()
         if partial:

tests/test_benchmark_reporting.py

@@ -136,6 +136,23 @@ def test_failed_run_remains_a_failure_without_partial_metrics(tmp_path):
     summary = benchmark.summarize_run(tmp_path, 1)
     assert summary["status"] == "failed"
     assert "cases" not in summary
+    assert summary["scan_diagnostics"]["processes"] == 4
+
+
+def test_scan_limits_are_never_hidden_behind_a_passed_status(tmp_path):
+    manifest = report_files(tmp_path)
+    path = tmp_path / "scan-summaries.json"
+    scans = json.loads(path.read_text())
+    scans["0"]["candidate_limit_hits"] = 7
+    path.write_text(json.dumps(scans))
+    with pytest.raises(ValueError, match="explicitly marked"):
+        benchmark.summarize_run(tmp_path, 1)
+    manifest.update(status="completed_with_limits", report_scan_limits=True)
+    (tmp_path / "manifest.json").write_text(json.dumps(manifest))
+    summary = benchmark.summarize_run(tmp_path, 1)
+    assert summary["status"] == "completed_with_limits"
+    assert not summary["checks"]["full_endpoint_scans"]
+    assert summary["scan_diagnostics"]["candidate_limit_hits"] == 7
 
 
 @pytest.mark.parametrize("problem", ["cleanup", "record_count", "timings"])

tests/test_core.py

@@ -169,6 +169,16 @@ def test_backtrack_limit_is_observable():
     assert stats.candidate_limit_hits > 0
 
 
+@pytest.mark.parametrize("encoding", ["utf-8", "utf-16-le"])
+def test_dense_successful_messages_do_not_report_backtrack_exhaustion(encoding):
+    objects = [request(rid=i) for i in range(133)]
+    raw = "\n".encode(encoding).join(encode(obj, encoding) for obj in objects)
+    stats = ScanStats()
+    records = list(scan_chunks([(0, raw)], stats=stats))
+    assert [r.obj for r in records] == objects
+    assert stats.candidate_limit_hits == 0
+
+
 def test_candidate_addresses_are_bounded_without_dropping_artifacts():
     req = carve(request())
     responses = [carve({"jsonrpc": "2.0", "id": 1, "result": {}}) for _ in range(300)]

validation/windows_positive/BENCHMARK.md

@@ -30,7 +30,7 @@ Use a fresh administrative Windows VM for each replica. Install the repository a
 `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/run.py --workdir C:\MCPBenchmark\work --reportdir C:\MCPBenchmark\report --tool-calls 64 --scan-repeats 3 --report-scan-limits
 python validation/windows_positive/benchmark.py summarize --reportdir C:\MCPBenchmark\report --replica 1 --output benchmark-1.json
 ```
 
@@ -44,7 +44,11 @@ The GitHub Actions `Windows memory benchmark` workflow performs the three captur
 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.
+dropped from the three-capture aggregate. Read errors and missing endpoints remain
+failures. `--report-scan-limits` allows bounded-search results to be measured, but
+marks them `completed_with_limits`, preserves every limit counter, and sets
+`full_endpoint_scans=false`; the aggregate reports how many captures were limited.
+The normal positive-validation command still rejects limited scans by default.
 
 ## Interpretation
 

validation/windows_positive/README.md

@@ -61,3 +61,7 @@ GitHub Actions의 `Windows positive memory validation` 워크플로도 같은
 `--tool-calls`로 조건별 도구 호출 수를, `--scan-repeats`로 동일 이미지의 CLI 분석 반복 수를
 지정할 수 있습니다. 기본값은 기존 검증과 같은 8회/1회입니다. 별도 VM 3개에서 64회 호출과
 3회 분석을 수행하는 [반복 벤치마크 절차](BENCHMARK.md)도 제공합니다.
+
+벤치마크의 `--report-scan-limits`는 실제 탐색 제한에 도달한 결과를
+`completed_with_limits`로 구분해 집계합니다. 기본 양성 검증은 제한 도달을 계속 실패로 처리하며,
+읽기 오류나 누락된 대상 프로세스는 벤치마크에서도 실패입니다.

validation/windows_positive/benchmark.py

@@ -23,12 +23,31 @@ CASE_FIELDS = (
     "unverifiable_candidate_links",
 )
 ENVIRONMENT_FIELDS = ("platform", "python", "mcp_sdk", "volatility", "logical_cpus")
+COMPLETED = {"passed", "completed_with_limits"}
 
 
 def read_json(path):
     return json.loads(path.read_text(encoding="utf-8"))
 
 
+def scan_diagnostics(scans):
+    return {
+        "processes": len(scans),
+        "limited_processes": sum(
+            bool(s["limits_hit"] or s["anchor_limit_hits"] or s["candidate_limit_hits"])
+            for s in scans.values()
+        ),
+        "error_processes": sum(bool(s["error"]) for s in scans.values()),
+        "skipped_processes": sum(bool(s["skipped_reason"]) for s in scans.values()),
+        **{
+            key: sum(s[key] for s in scans.values())
+            for key in ("invalid_regions", "anchor_limit_hits", "candidate_limit_hits")
+        },
+        "record_budget_processes": sum("max-records" in s["limits_hit"] for s in scans.values()),
+        "byte_budget_processes": sum("max-bytes" in s["limits_hit"] for s in scans.values()),
+    }
+
+
 def summarize_run(report, replica):
     """Exclude recovered content, ground truth, PIDs, paths, and arbitrary log text."""
     path = report / "manifest.json"
@@ -36,13 +55,18 @@ def summarize_run(report, replica):
     output = {
         "schema_version": 1,
         "replica": replica,
-        "status": "passed" if manifest.get("status") == "passed" else "failed",
+        "status": manifest["status"] if manifest.get("status") in COMPLETED 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"),
+        "report_scan_limits": manifest.get("report_scan_limits", False),
     }
-    if output["status"] != "passed":
+    scan_path = report / "scan-summaries.json"
+    scans = read_json(scan_path) if scan_path.exists() else {}
+    if scans:
+        output["scan_diagnostics"] = scan_diagnostics(scans)
+    if output["status"] not in COMPLETED:
         return output
     for key in (
         "raw_memory_deleted",
@@ -52,17 +76,17 @@ def summarize_run(report, replica):
         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()
+        s["error"] or s["skipped_reason"] or s["invalid_regions"] for s in scans.values()
     ):
         raise ValueError("Benchmark requires four complete endpoint scans")
+    limited = output["scan_diagnostics"]["limited_processes"] > 0
+    if limited and (
+        output["status"] != "completed_with_limits" or not output["report_scan_limits"]
+    ):
+        raise ValueError("Limited scans must be explicitly marked completed_with_limits")
+    if not limited and output["status"] != "passed":
+        raise ValueError("Completed status does not agree with scan diagnostics")
     cases = {}
     for case in comparison["cases"]:
         counts = {key: case[key] for key in CASE_FIELDS}
@@ -123,7 +147,7 @@ def summarize_run(report, replica):
             for c in manifest["cases"]
         },
         checks={
-            "full_endpoint_scans": True,
+            "full_endpoint_scans": not limited,
             "native_bytes_verified": True,
             "repeated_exports_identical": True,
             "image_hash_unchanged": True,
@@ -147,11 +171,18 @@ 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):
+    if any(r["status"] not in COMPLETED 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"):
+    for key in (
+        "schema_version",
+        "commit",
+        "environment",
+        "tool_calls_per_case",
+        "scan_repeats",
+        "report_scan_limits",
+    ):
         if any(r[key] != runs[0][key] for r in runs):
             raise ValueError(f"Incompatible benchmark conditions: {key}")
     modes = {}
@@ -173,6 +204,7 @@ def aggregate_runs(runs):
     return {
         "schema_version": 1,
         "independent_captures": 3,
+        "captures_with_scan_limits": sum(r["status"] == "completed_with_limits" for r in runs),
         "commit": runs[0]["commit"],
         "environment": runs[0]["environment"],
         "tool_calls_per_case": runs[0]["tool_calls_per_case"],

validation/windows_positive/run.py

@@ -83,6 +83,7 @@ def main():
     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)
+    parser.add_argument("--report-scan-limits", action="store_true")
     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")
@@ -115,6 +116,7 @@ def main():
         "logical_cpus": os.cpu_count(),
         "tool_calls_per_case": args.tool_calls,
         "scan_repeats": args.scan_repeats,
+        "report_scan_limits": args.report_scan_limits,
         "timings_seconds": timings,
     }
     children, cases = [], []
@@ -273,18 +275,21 @@ def main():
         verification_started = time.perf_counter()
         artifacts, summaries = verify_exports(image, symbols, cache, config, exports)
         timings["native_byte_verification"] = time.perf_counter() - verification_started
+        # Persist diagnostics before deciding whether a scan can be scored.
+        write_json(report / "scan-summaries.json", summaries)
+        limited = False
         for summary in summaries.values():
+            if summary["error"] or summary["skipped_reason"] or summary["invalid_regions"]:
+                raise RuntimeError("An endpoint scan failed; inspect scan summaries")
             if (
-                summary["error"]
-                or summary["limits_hit"]
-                or summary["skipped_reason"]
+                summary["limits_hit"]
                 or summary["anchor_limit_hits"]
                 or summary["candidate_limit_hits"]
             ):
-                raise RuntimeError(
-                    "A selected endpoint scan was incomplete; inspect scan summaries"
-                )
-        write_json(report / "scan-summaries.json", summaries)
+                limited = True
+        manifest["scan_complete"] = not limited
+        if limited and not args.report_scan_limits:
+            raise RuntimeError("A selected endpoint scan was incomplete; inspect scan summaries")
         scores, publish_keys = [], set()
         for case in cases:
             mode = case["mode"]
@@ -326,7 +331,7 @@ def main():
             raise RuntimeError("No complete known tool request/response pair recovered")
         if any(s["incorrect_candidate_links"] for s in scores):
             raise RuntimeError("Incorrect candidate links observed in controlled traffic")
-        manifest["analysis_status"] = "passed"
+        manifest["analysis_status"] = "completed_with_limits" if limited else "passed"
         checkpoint("analysis-completed")
         print(json.dumps(result, ensure_ascii=True, indent=2), flush=True)
     except BaseException as exc:
@@ -354,7 +359,7 @@ def main():
         else:
             manifest["raw_memory_deleted"] = True
             write_json(report / "manifest.json", manifest)
-    manifest["status"] = "passed"
+    manifest["status"] = manifest["analysis_status"]
     timings["pipeline_total"] = time.perf_counter() - pipeline_started
     checkpoint("completed")