Flamehaven Initiative commited on
Commit
7b91ab8
·
1 Parent(s): dd357f4

feat: compact repeated explain findings

Browse files
Files changed (1) hide show
  1. stem_ai/render.py +57 -2
stem_ai/render.py CHANGED
@@ -263,6 +263,7 @@ def render_explain(result: dict[str, Any]) -> str:
263
  f"Score : {score['final_score']} / 100 ({score['formal_tier']})",
264
  f"Replic : {result.get('replication_score', 0)} / 100"
265
  f" ({result.get('replication_tier', 'R0')})",
 
266
  _EXPLAIN_SEP, "",
267
  ]
268
  for detector, findings in grouped.items():
@@ -437,6 +438,7 @@ def _markdown_airi_section(airi: dict[str, Any]) -> list[str]:
437
  f"**Bundle Scope:** `{bundle_scope}`",
438
  f"**Upstream Snapshot:** `{snapshot}`",
439
  "**Interpretation:** This is detector-mapped AIRI coverage inside the current runtime bundle, not a claim that unmapped risks are absent.",
 
440
  ]
441
  covered_risks = airi.get("covered_risks", [])
442
  if covered_risks:
@@ -459,15 +461,68 @@ def _markdown_airi_section(airi: dict[str, Any]) -> list[str]:
459
 
460
 
461
  def _explain_detector_group(detector: str, findings: list[dict[str, Any]]) -> list[str]:
 
462
  label = _explain_status_label({f["status"] for f in findings})
463
  noun = "finding" if len(findings) == 1 else "findings"
464
- lines = [f"{detector} [{label}] ({len(findings)} {noun})"]
465
- for f in findings:
 
 
 
 
466
  lines += _explain_finding_lines(f)
467
  lines.append("")
468
  return lines
469
 
470
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
471
  def _explain_finding_lines(f: dict[str, Any]) -> list[str]:
472
  occ = f["finding_id"].rsplit(":", 1)[-1]
473
  file_str = "(repository)" if f["file"] == "." else (
 
263
  f"Score : {score['final_score']} / 100 ({score['formal_tier']})",
264
  f"Replic : {result.get('replication_score', 0)} / 100"
265
  f" ({result.get('replication_tier', 'R0')})",
266
+ "Surface : repeated same-file evidence may be compacted in narrative output; JSON remains canonical.",
267
  _EXPLAIN_SEP, "",
268
  ]
269
  for detector, findings in grouped.items():
 
438
  f"**Bundle Scope:** `{bundle_scope}`",
439
  f"**Upstream Snapshot:** `{snapshot}`",
440
  "**Interpretation:** This is detector-mapped AIRI coverage inside the current runtime bundle, not a claim that unmapped risks are absent.",
441
+ "**Surface Note:** repeated same-file evidence may be compacted in human-readable surfaces; canonical per-finding rows remain in JSON.",
442
  ]
443
  covered_risks = airi.get("covered_risks", [])
444
  if covered_risks:
 
461
 
462
 
463
  def _explain_detector_group(detector: str, findings: list[dict[str, Any]]) -> list[str]:
464
+ compact_findings = _compact_explain_findings(findings)
465
  label = _explain_status_label({f["status"] for f in findings})
466
  noun = "finding" if len(findings) == 1 else "findings"
467
+ compact_noun = "row" if len(compact_findings) == 1 else "rows"
468
+ heading = f"{detector} [{label}] ({len(findings)} {noun})"
469
+ if len(compact_findings) != len(findings):
470
+ heading += f" -> compacted to {len(compact_findings)} {compact_noun}"
471
+ lines = [heading]
472
+ for f in compact_findings:
473
  lines += _explain_finding_lines(f)
474
  lines.append("")
475
  return lines
476
 
477
 
478
+ def _compact_explain_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]:
479
+ compact: list[dict[str, Any]] = []
480
+ grouped: dict[tuple[str, str, str, str], dict[str, Any]] = {}
481
+ grouped_order: list[tuple[str, str, str, str]] = []
482
+
483
+ for finding in findings:
484
+ status = str(finding.get("status", "unknown"))
485
+ file_path = str(finding.get("file", ""))
486
+ line = int(finding.get("line", 0) or 0)
487
+ reason = str(finding.get("explanation") or finding.get("message") or "").strip()
488
+
489
+ if status in {"detected", "warn", "pass"} and file_path not in {"", "."} and reason:
490
+ key = (status, file_path, reason, str(finding.get("pattern_id", "")))
491
+ if key not in grouped:
492
+ clone = dict(finding)
493
+ meta = dict(clone.get("metadata") or {})
494
+ meta["aggregate_count"] = 1
495
+ meta["aggregate_lines"] = [line] if line else []
496
+ meta["aggregate_surface"] = "explain_same_file_reason"
497
+ clone["metadata"] = meta
498
+ grouped[key] = clone
499
+ grouped_order.append(key)
500
+ else:
501
+ grouped[key]["metadata"]["aggregate_count"] += 1
502
+ if line and line not in grouped[key]["metadata"]["aggregate_lines"]:
503
+ grouped[key]["metadata"]["aggregate_lines"].append(line)
504
+ continue
505
+
506
+ compact.append(finding)
507
+
508
+ for key in grouped_order:
509
+ group = grouped[key]
510
+ count = int(group.get("metadata", {}).get("aggregate_count", 1))
511
+ lines = sorted(group.get("metadata", {}).get("aggregate_lines", []))
512
+ if count > 1:
513
+ if lines:
514
+ preview = ", ".join(str(n) for n in lines[:6])
515
+ if len(lines) > 6:
516
+ preview += ", ..."
517
+ group["explanation"] = f"{group.get('explanation', '')} Aggregated {count} similar findings from one file (lines: {preview}).".strip()
518
+ else:
519
+ group["explanation"] = f"{group.get('explanation', '')} Aggregated {count} similar findings from one file.".strip()
520
+ group["snippet"] = ""
521
+ compact.append(group)
522
+
523
+ return compact
524
+
525
+
526
  def _explain_finding_lines(f: dict[str, Any]) -> list[str]:
527
  occ = f["finding_id"].rsplit(":", 1)[-1]
528
  file_str = "(repository)" if f["file"] == "." else (