fix: bound scans match --all history payloads

sdk/typescript/README.md

@@ -440,6 +440,10 @@ checkout so a fixed vulnerability can be checked again.
 cause; `scans match --all` matches all completed scans of the current repository,
 including other worktrees and clones. Saved matches appear in `scans show` and
 are reused unless `--force` is passed. Scans without sealed artifacts are skipped.
+Automatic history matching loads at most 64 scan pairs per workbench page and
+32 findings or 512 KiB per finding page. Each model comparison receives one
+bounded page from each scan; cross-page matches are reconciled before the pair
+is saved. Match results larger than 1 MiB are rejected with an actionable error.
 
 `scans compare BEFORE_SCAN_ID AFTER_SCAN_ID` automatically matches findings by
 root cause, reuses saved matches, and reports findings as new, persisting,

sdk/typescript/_bundled_plugin/scripts/workbench_cli.py

@@ -152,6 +152,11 @@ def parse_args(description: str) -> argparse.Namespace:
     list_unmatched_scan_pairs = subparsers.add_parser("list-unmatched-scan-pairs")
     list_unmatched_scan_pairs.add_argument("--repository", required=True)
     list_unmatched_scan_pairs.add_argument("--force", action="store_true")
+    list_unmatched_scan_pairs.add_argument("--offset", type=non_negative_int, default=0)
+
+    get_scan_matching_inputs = subparsers.add_parser("get-scan-matching-inputs")
+    get_scan_matching_inputs.add_argument("--scan-id", required=True)
+    get_scan_matching_inputs.add_argument("--offset", type=non_negative_int, default=0)
 
     register_cli_scan = subparsers.add_parser("register-cli-scan")
     register_cli_scan.add_argument("--scan-dir", required=True)

sdk/typescript/_bundled_plugin/scripts/workbench_constants.py

@@ -42,6 +42,9 @@
 PATCH_ARTIFACT_MAX_BYTES = 2 * 1024 * 1024
 FINDINGS_RESULT_LIMIT = 20
 FINDINGS_PAGE_MAX = 20
+MATCHING_PAIR_PAGE_MAX = 64
+MATCHING_FINDING_PAGE_MAX = 32
+MATCHING_INPUT_PAGE_BYTES = 512 * 1024
 FINDING_DETAILS_PREVIEW_BYTES = 16_000
 FINDING_ROOT_CAUSE_PREVIEW_BYTES = 2_000
 FINDING_VALIDATION_PREVIEW_BYTES = 3_000

sdk/typescript/_bundled_plugin/scripts/workbench_db.py

@@ -3568,9 +3568,16 @@ def main() -> None:
             result = scan_history.list_unmatched_scan_pairs(
                 connection,
                 args,
-                backfill_finding_details=backfill_legacy_finding_details,
                 read_coverage=coverage_for_comparison,
             )
+        elif args.command == "get-scan-matching-inputs":
+            result = scan_history.get_scan_matching_inputs(
+                connection,
+                args,
+                require_scan=require_scan,
+                read_coverage=coverage_for_comparison,
+                backfill_finding_details=backfill_legacy_finding_details,
+            )
         elif args.command == "register-cli-scan":
             result = register_cli_scan(connection, args)
         elif args.command == "get-scan-recipe":

sdk/typescript/_bundled_plugin/scripts/workbench_scan_history.py

@@ -3,14 +3,20 @@
 import argparse
 import fnmatch
 import json
+import math
 import os
 import sqlite3
 from pathlib import Path, PurePosixPath
 from typing import Any, Callable
 from urllib.parse import urlsplit
 
 from report_projection import SEVERITY_ORDER
-from workbench_constants import FINDINGS_PAGE_MAX
+from workbench_constants import (
+    FINDINGS_PAGE_MAX,
+    MATCHING_FINDING_PAGE_MAX,
+    MATCHING_INPUT_PAGE_BYTES,
+    MATCHING_PAIR_PAGE_MAX,
+)
 from workbench_target import git_output
 
 
@@ -271,7 +277,6 @@ def list_unmatched_scan_pairs(
     connection: sqlite3.Connection,
     args: argparse.Namespace,
     *,
-    backfill_finding_details: Callable[[sqlite3.Connection, sqlite3.Row], None],
     read_coverage: Callable[[sqlite3.Row], dict[str, Any]],
 ) -> dict[str, Any]:
     repository = Path(args.repository).expanduser().resolve()
@@ -302,49 +307,119 @@ def list_unmatched_scan_pairs(
         (row["before_scan_id"], row["after_scan_id"])
         for row in connection.execute("SELECT before_scan_id, after_scan_id FROM scan_comparisons")
     }
-    batches = []
-    skipped = 0
-    backfilled: set[str] = set()
-    for index, after in enumerate(available):
-        previous = [
-            before
-            for before in available[:index]
-            if args.force or (before["id"], after["id"]) not in saved_pairs
-        ]
-        skipped += index - len(previous)
-        if not previous:
-            continue
-        for scan in (*previous, after):
-            if scan["id"] not in backfilled:
-                backfill_finding_details(connection, scan)
-                backfilled.add(scan["id"])
-        batches.append(
-            {
-                "afterFindings": [
-                    _matching_input(row) for row in _scan_findings(connection, after["id"]).values()
-                ],
-                "afterScanId": after["id"],
-                "beforeScans": [
-                    {
-                        "findings": [
-                            _matching_input(row)
-                            for row in _scan_findings(connection, before["id"]).values()
-                        ],
-                        "scanId": before["id"],
-                    }
-                    for before in previous
-                ],
-            }
+    available_indexes = {scan["id"]: index for index, scan in enumerate(available)}
+    skipped = (
+        0
+        if args.force
+        else sum(
+            1
+            for before_id, after_id in saved_pairs
+            if before_id in available_indexes
+            and after_id in available_indexes
+            and available_indexes[before_id] < available_indexes[after_id]
         )
+    )
+    pair_count = len(available) * (len(available) - 1) // 2
+    page_end = min(args.offset + MATCHING_PAIR_PAGE_MAX, pair_count)
+    pair_index = min(args.offset, pair_count)
+    after_index = (1 + math.isqrt(1 + 8 * pair_index)) // 2
+    before_index = pair_index - after_index * (after_index - 1) // 2
+    pairs = []
+    while pair_index < page_end:
+        before = available[before_index]
+        after = available[after_index]
+        if args.force or (before["id"], after["id"]) not in saved_pairs:
+            pairs.append(
+                {
+                    "afterScanId": after["id"],
+                    "beforeScanId": before["id"],
+                }
+            )
+        pair_index += 1
+        before_index += 1
+        if before_index == after_index:
+            after_index += 1
+            before_index = 0
     return {
-        "batches": batches,
+        "nextOffset": page_end if page_end < pair_count else None,
+        "pairs": pairs,
         "repository": str(repository),
         "scanCount": len(selected),
         "skippedPairs": skipped,
         "unavailableScans": len(selected) - len(available),
     }
 
 
+def get_scan_matching_inputs(
+    connection: sqlite3.Connection,
+    args: argparse.Namespace,
+    *,
+    require_scan: Callable[[sqlite3.Connection, str], sqlite3.Row],
+    read_coverage: Callable[[sqlite3.Row], dict[str, Any]],
+    backfill_finding_details: Callable[[sqlite3.Connection, sqlite3.Row], None],
+) -> dict[str, Any]:
+    scan = require_scan(connection, args.scan_id)
+    if scan["status"] != "complete":
+        raise SystemExit("Only completed scans can be matched.")
+    read_coverage(scan)
+    backfill_finding_details(connection, scan)
+    if connection.execute(
+        """
+        SELECT 1
+        FROM finding_occurrences
+        WHERE scan_id = ? AND length(CAST(details_json AS BLOB)) > ?
+        LIMIT 1
+        """,
+        (scan["id"], MATCHING_INPUT_PAGE_BYTES),
+    ).fetchone():
+        raise SystemExit(
+            "A scan finding exceeds the 512 KiB automatic matching input limit."
+        )
+    total_findings = connection.execute(
+        "SELECT COUNT(*) FROM finding_occurrences WHERE scan_id = ?",
+        (scan["id"],),
+    ).fetchone()[0]
+    rows = _scan_finding_page(
+        connection,
+        scan["id"],
+        offset=args.offset,
+        limit=MATCHING_FINDING_PAGE_MAX,
+    )
+    findings = []
+    encoded_bytes = 2
+    next_offset = min(args.offset, total_findings)
+    for row in rows:
+        finding = _matching_input(row)
+        item_bytes = len(
+            json.dumps(
+                finding,
+                allow_nan=False,
+                ensure_ascii=False,
+                separators=(",", ":"),
+            ).encode("utf-8")
+        )
+        if item_bytes > MATCHING_INPUT_PAGE_BYTES:
+            raise SystemExit(
+                "A scan finding exceeds the 512 KiB automatic matching input limit."
+            )
+        separator_bytes = 1 if findings else 0
+        if encoded_bytes + separator_bytes + item_bytes > MATCHING_INPUT_PAGE_BYTES:
+            if not findings:
+                raise SystemExit(
+                    "A scan finding exceeds the 512 KiB automatic matching input limit."
+                )
+            break
+        findings.append(finding)
+        encoded_bytes += separator_bytes + item_bytes
+        next_offset += 1
+    return {
+        "findings": findings,
+        "nextOffset": next_offset if next_offset < total_findings else None,
+        "scanId": scan["id"],
+        "totalFindings": total_findings,
+    }
+
+
 def compare_scans(
     connection: sqlite3.Connection,
     args: argparse.Namespace,
@@ -781,12 +856,44 @@ def _scan_findings(connection: sqlite3.Connection, scan_id: str) -> dict[str, sq
         FROM finding_occurrences AS occurrences
         LEFT JOIN finding_triage AS triage ON triage.occurrence_id = occurrences.id
         WHERE occurrences.scan_id = ?
+        ORDER BY occurrences.id
         """,
         (scan_id,),
     )
     return {row["finding_id"]: row for row in rows}
 
 
+def _scan_finding_page(
+    connection: sqlite3.Connection,
+    scan_id: str,
+    *,
+    offset: int,
+    limit: int,
+) -> list[sqlite3.Row]:
+    return list(
+        connection.execute(
+            """
+            SELECT occurrences.*,
+                COALESCE(triage.status, 'open') AS triage_status, triage.close_reason,
+                (
+                    SELECT locations.relative_path
+                    FROM finding_locations AS locations
+                    WHERE locations.occurrence_id = occurrences.id
+                    ORDER BY CASE WHEN locations.role = 'root_control' THEN 0 ELSE 1 END,
+                        locations.sort_order
+                    LIMIT 1
+                ) AS relative_path
+            FROM finding_occurrences AS occurrences
+            LEFT JOIN finding_triage AS triage ON triage.occurrence_id = occurrences.id
+            WHERE occurrences.scan_id = ?
+            ORDER BY occurrences.id
+            LIMIT ? OFFSET ?
+            """,
+            (scan_id, limit, offset),
+        )
+    )
+
+
 def scan_covers_path(
     scan: sqlite3.Row,
     *,