perf(cpg): one taint row per evidence key, with the path count

Ten real pmpro pairs at callDepth 3 returned 3,838 flow rows. Every field the driver reads is a property
of the sink or of (source kind, source, sanitized); only `length` varies per path, and the driver ranks
by the shortest. Grouped on that key with the minimum length and a `paths` count: 1,155 rows carrying
3,838 paths, same time. The specs are corrected too -- the first write-up called the depth-3 rows
"mostly the same evidence repeated", and 3.3x is what the measurement says; the other 30x is distinct
sources.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

.kiro/specs/contributor-scan/tasks.md

@@ -905,12 +905,13 @@
     | 1 | 124.9 s | 12.5 s | 38 |
     | 3 | 150.5 s | 15.0 s | **3,838** |
 
-    Depth is a 1.4× time term, not the 5× the sharding-era numbers implied -- but it is a **100× payload
-    term**: at depth 3 the same ten pairs return 3,838 flow rows, most of them the same (sink, source kind,
-    sanitized) evidence repeated once per path through the reachable set. Across 339 real requests that is
-    on the order of 130,000 rows to ship and parse, and the reporter collapses them only after they have
-    been paid for. The next cost to take out is in the query's row emission, not its graph work: one row
-    per (sink, source kind, sanitized), with a path count, would be the same evidence at 1% of the bytes.
+    Depth is a 1.4× time term, not the 5× the sharding-era numbers implied -- and a 100× payload term,
+    3,838 rows against 38. **How much of that was duplication, measured:** the query now emits one row per
+    (sink, source kind, source, sanitized) with the shortest `length` and a `paths` count, and the same ten
+    requests at depth 3 return **1,155 rows carrying 3,838 paths** -- a 3.3× collapse, exact by
+    accounting, at the same time (15.9 s/request). The other 30× is depth 3 finding more distinct sources
+    per sink, which is evidence, not repetition; the first write-up of this table called it "mostly the
+    same evidence repeated" and that was a guess the measurement did not support.
   - **The whole-repository scan completes, 2026-09-11.** pmpro, 637 files, the product's own build path
     (overlays at build, seeds scoped, size guard), no judge, `OPENULTRASAST_CPG_QUERY_TIMEOUT=7200`:
     build 97 s, evidence 76 s, **taint 4,105 s for 348 real requests = 11.8 s each**, arbitrate 99 s,

.kiro/specs/flow-aware-ranking/brief.md

@@ -110,8 +110,9 @@ A phase that cannot show that is not an improvement, it is a story.
   found at `class.memberorder.php:936`**, among 250 entailed findings across 40 files. The 250 is the next
   fact to reckon with -- it is what "500 regions when 50 would do" looks like as output -- and it is phase
   2's material. `callDepth`, measured on the same graph, is a 1.4× time term and a 100× payload term
-  (3,838 flow rows for ten requests at depth 3 against 38 at depth 1); collapsing rows per evidence key
-  is the next cost out.
+  (3,838 flow rows for ten requests at depth 3 against 38 at depth 1). Collapsing rows per evidence key
+  takes 3.3× of that out (1,155 rows carrying 3,838 paths, exact); the rest is depth 3 naming more
+  distinct sources per sink, which is evidence the ranker gets to use.
 
   Phase 0 reported 1,148 ms per request after memoisation. That was an average over a 250-request set of
   which 88% were tier 0 and returned instantly; the ~30 real requests in it cost about 9.5 s each. Pruning

src/openultrasast/cpg/queries/taint.sc

@@ -619,28 +619,41 @@
           boundShape.matcher(c).find() && argNames.exists(a => mentionsToken(c, List(a)))
         )
 
-    flows.map { flow =>
-      val elements = flow.elements.map(_.code).l
-      val sanitized = sanitizerNames.nonEmpty && flow.elements.l.exists(node => sanitizesHere(node))
-      val sourceKind = sourceKindOf(flow.elements.l.headOption)
-      ujson.Obj(
-        "sink"            -> sink.code.take(200),
-        "sourceKind"      -> sourceKind,
-        "sinkLine"        -> sink.lineNumber.getOrElse(-1).toString,
-        "sinkMethod"      -> sink.method.name,
-        // Where the sink actually is. With callDepth > 0 that need not be the region's own file, and a
-        // finding reported against the handler's file when the bug is in another module is unactionable.
-        "sinkFile"        -> sink.method.filename,
-        "source"          -> elements.headOption.getOrElse("").take(200),
-        "sanitized"       -> sanitized,
-        "length"          -> elements.size,
-        "sinkArity"       -> arity,
-        "sinkArg0Literal" -> arg0Literal,
-        "bounded"         -> bounds.nonEmpty,
-        "bound"            -> bounds.headOption.getOrElse("").take(120),
-        "inLabeledScope"  -> (function.isEmpty || nestedInLabeled(sink.method) || reachableMethods.contains(sink.method.fullName))
-      )
-    }
+    // ONE ROW PER EVIDENCE KEY, not per path. Every field the driver reads is a property of the sink or of
+    // (source kind, source, sanitized); only `length` varies per path, and the driver ranks by the shortest.
+    // Emitting every path was a 100x payload term: ten real requests at callDepth 3 returned 3,838 rows
+    // for 38 distinct keys, and a whole-repository scan would ship on the order of 130,000 rows to have
+    // the reporter collapse them after they were paid for. `paths` keeps the count. Task 5.13.
+    flows
+      .map { flow =>
+        val elements   = flow.elements.map(_.code).l
+        val sanitized  = sanitizerNames.nonEmpty && flow.elements.l.exists(node => sanitizesHere(node))
+        val sourceKind = sourceKindOf(flow.elements.l.headOption)
+        ((sourceKind, elements.headOption.getOrElse("").take(200), sanitized), elements.size)
+      }
+      .groupBy(_._1)
+      .toList
+      .sortBy(_._1)
+      .map { case ((sourceKind, source, sanitized), paths) =>
+        ujson.Obj(
+          "sink"            -> sink.code.take(200),
+          "sourceKind"      -> sourceKind,
+          "sinkLine"        -> sink.lineNumber.getOrElse(-1).toString,
+          "sinkMethod"      -> sink.method.name,
+          // Where the sink actually is. With callDepth > 0 that need not be the region's own file, and a
+          // finding reported against the handler's file when the bug is in another module is unactionable.
+          "sinkFile"        -> sink.method.filename,
+          "source"          -> source,
+          "sanitized"       -> sanitized,
+          "length"          -> paths.map(_._2).min,
+          "paths"           -> paths.size,
+          "sinkArity"       -> arity,
+          "sinkArg0Literal" -> arg0Literal,
+          "bounded"         -> bounds.nonEmpty,
+          "bound"            -> bounds.headOption.getOrElse("").take(120),
+          "inLabeledScope"  -> (function.isEmpty || nestedInLabeled(sink.method) || reachableMethods.contains(sink.method.fullName))
+        )
+      }
   }
 
     rows