feat: add trace cleanup tool

reports/trace_cleanup/trace_fix_log.csv

@@ -0,0 +1 @@
+entry_id,report_id,verify,rule_id,status,field_path,trace_index,file,raw_line,boundary_role,boundary_file,boundary_raw_line,relation,before,after,kept_index,reason

reports/trace_cleanup/trace_summary.json

@@ -0,0 +1,19 @@
+{
+  "entries_scanned": 408,
+  "trace_nodes_scanned": 2073,
+  "duplicate_groups": 7,
+  "duplicate_extra_nodes": 7,
+  "before_entry_findings": 139,
+  "after_critical_findings": 237,
+  "overlap_findings": 41,
+  "cross_file_skipped": 382,
+  "auto_fixed": 0,
+  "manual_review": 5,
+  "protected_verified": 415,
+  "changed_entries": 0,
+  "input_sha256": "2158b6bfef0be1812e7a6a77b32ad32b65964c2546c83018ff20a9a6f706c7b1",
+  "output_sha256": "2158b6bfef0be1812e7a6a77b32ad32b65964c2546c83018ff20a9a6f706c7b1",
+  "mode": "fix",
+  "duplicate_action": "remove-safe",
+  "order_action": "report"
+}

tests/trace_cleanup/__init__.py

@@ -0,0 +1 @@
+"""Tests for the trace cleanup tool."""

tests/trace_cleanup/fixtures.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from copy import deepcopy
+
+
+def node(
+    line: object,
+    *,
+    file: str = "src/example.py",
+    code: str | None = None,
+    desc: str | None = "step",
+    include_desc: bool = True,
+) -> dict:
+    value = {
+        "file": file,
+        "line": line,
+        "code": code if code is not None else f"code at {line}",
+    }
+    if include_desc:
+        value["desc"] = desc
+    return value
+
+
+def entry(
+    *,
+    entry_id: str = "entry-00001",
+    verify: int = 0,
+    entry_point: dict | None = None,
+    critical_operation: dict | None = None,
+    trace: list[dict] | None = None,
+) -> dict:
+    return {
+        "commit": "a" * 40,
+        "critical_operation": deepcopy(
+            critical_operation or node(20, code="critical")
+        ),
+        "entry_id": entry_id,
+        "entry_point": deepcopy(entry_point or node(10, code="entry")),
+        "origin": "GitHub Advisory Database (reviewed)",
+        "project": "fixture",
+        "repo_url": "https://github.com/example/fixture",
+        "report_id": "GHSA-AAAA-BBBB-CCCC",
+        "source_link": "https://github.com/advisories/GHSA-aaaa-bbbb-cccc",
+        "trace": deepcopy(trace or [node(10, code="entry"), node(20, code="critical")]),
+        "verify": verify,
+        "vuln_category_l1": "test",
+        "vuln_category_l2": "test",
+        "vuln_ids": ["GHSA-AAAA-BBBB-CCCC"],
+        "vuln_title": "fixture",
+    }

tests/trace_cleanup/test_cli.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from tools.trace_cleanup.cli import main
+from tools.trace_cleanup.io import load_jsonl, write_jsonl_atomic
+
+from .fixtures import entry, node
+
+
+def write_input(path: Path, rows: list[dict]) -> bytes:
+    data = "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows).encode()
+    path.write_bytes(data)
+    return data
+
+
+class CliTests(unittest.TestCase):
+    def test_check_does_not_create_fixed_jsonl(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            reports = root / "reports"
+            fixed = root / "entries.trace_fixed.jsonl"
+            write_input(source, [entry()])
+            self.assertEqual(
+                main(["check", "--input", str(source), "--reports-dir", str(reports)]),
+                0,
+            )
+            self.assertFalse(fixed.exists())
+            self.assertTrue((reports / "trace_findings.csv").exists())
+            self.assertFalse((reports / "trace_fix_log.csv").exists())
+
+    def test_fix_removes_safe_duplicate_but_not_order_by_default(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            output = root / "fixed.jsonl"
+            reports = root / "reports"
+            repeated = node(5, code="repeat", desc="same")
+            write_input(source, [entry(trace=[repeated, repeated, node(20)])])
+            main(
+                [
+                    "fix",
+                    "--input",
+                    str(source),
+                    "--output",
+                    str(output),
+                    "--reports-dir",
+                    str(reports),
+                ]
+            )
+            fixed = load_jsonl(output)
+            self.assertEqual([n["line"] for n in fixed[0]["trace"]], [5, 20])
+            self.assertTrue((reports / "trace_fix_log.csv").exists())
+
+    def test_explicit_remove_definite_and_verified_protection(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            output = root / "fixed.jsonl"
+            write_input(
+                source,
+                [
+                    entry(entry_id="entry-00001", trace=[node(5)]),
+                    entry(entry_id="entry-00002", verify=1, trace=[node(5)]),
+                ],
+            )
+            main(
+                [
+                    "fix",
+                    "--input",
+                    str(source),
+                    "--output",
+                    str(output),
+                    "--reports-dir",
+                    str(root / "reports"),
+                    "--order-action",
+                    "remove-definite",
+                ]
+            )
+            rows = load_jsonl(output)
+            self.assertEqual(rows[0]["trace"], [])
+            self.assertEqual(len(rows[1]["trace"]), 1)
+
+    def test_fix_refuses_to_overwrite_input(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            source = Path(tmp) / "entries.jsonl"
+            original = write_input(source, [entry()])
+            with self.assertRaises(SystemExit):
+                main(["fix", "--input", str(source), "--output", str(source)])
+            self.assertEqual(source.read_bytes(), original)
+
+    def test_entry_filter_preserves_unselected_rows(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            output = root / "fixed.jsonl"
+            repeated = node(12, code="repeat", desc="same")
+            rows = [
+                entry(entry_id="entry-00001", trace=[repeated, repeated]),
+                entry(entry_id="entry-00002", trace=[repeated, repeated]),
+            ]
+            write_input(source, rows)
+            main(
+                [
+                    "fix",
+                    "--input",
+                    str(source),
+                    "--output",
+                    str(output),
+                    "--reports-dir",
+                    str(root / "reports"),
+                    "--entry-id",
+                    "entry-00001",
+                ]
+            )
+            fixed = load_jsonl(output)
+            self.assertEqual(len(fixed[0]["trace"]), 1)
+            self.assertEqual(fixed[1], rows[1])
+
+    def test_output_is_deterministic_and_idempotent(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            first = root / "first.jsonl"
+            second = root / "second.jsonl"
+            third = root / "third.jsonl"
+            repeated = node(12, code="repeat", desc="same")
+            write_input(source, [entry(trace=[repeated, repeated])])
+            for output, reports in [
+                (first, root / "reports1"),
+                (second, root / "reports2"),
+            ]:
+                main(
+                    [
+                        "fix",
+                        "--input",
+                        str(source),
+                        "--output",
+                        str(output),
+                        "--reports-dir",
+                        str(reports),
+                    ]
+                )
+            self.assertEqual(first.read_bytes(), second.read_bytes())
+            self.assertEqual(
+                (root / "reports1" / "trace_summary.json").read_bytes(),
+                (root / "reports2" / "trace_summary.json").read_bytes(),
+            )
+            main(
+                [
+                    "fix",
+                    "--input",
+                    str(first),
+                    "--output",
+                    str(third),
+                    "--reports-dir",
+                    str(root / "reports3"),
+                ]
+            )
+            self.assertEqual(first.read_bytes(), third.read_bytes())
+            summary = json.loads((root / "reports3" / "trace_summary.json").read_text())
+            self.assertEqual(summary["changed_entries"], 0)
+
+    def test_atomic_write_failure_leaves_no_partial_file(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            output = root / "output.jsonl"
+            with mock.patch("tools.trace_cleanup.io.os.replace", side_effect=OSError("boom")):
+                with self.assertRaises(OSError):
+                    write_jsonl_atomic(output, [entry()])
+            self.assertFalse(output.exists())
+            self.assertEqual(list(root.glob("*.tmp")), [])
+
+    def test_input_hash_is_unchanged_after_fix(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            output = root / "fixed.jsonl"
+            original = write_input(source, [entry()])
+            original_hash = hashlib.sha256(original).hexdigest()
+            main(
+                [
+                    "fix",
+                    "--input",
+                    str(source),
+                    "--output",
+                    str(output),
+                    "--reports-dir",
+                    str(root / "reports"),
+                ]
+            )
+            self.assertEqual(hashlib.sha256(source.read_bytes()).hexdigest(), original_hash)
+
+    def test_zero_change_fix_is_byte_identical_including_crlf(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            source = root / "entries.jsonl"
+            output = root / "fixed.jsonl"
+            row = json.dumps(entry(), ensure_ascii=False).encode("utf-8")
+            source.write_bytes(row + b"\r\n")
+            main(
+                [
+                    "fix",
+                    "--input",
+                    str(source),
+                    "--output",
+                    str(output),
+                    "--reports-dir",
+                    str(root / "reports"),
+                ]
+            )
+            self.assertEqual(output.read_bytes(), source.read_bytes())

tests/trace_cleanup/test_dataset_regression.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+
+from tools.trace_cleanup.detect import scan_entries
+from tools.trace_cleanup.io import load_jsonl
+from tools.trace_cleanup.policy import decide_findings
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+ENTRIES = REPO_ROOT / "data" / "entries.jsonl"
+
+
+class DatasetRegressionTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.entries = load_jsonl(ENTRIES)
+        cls.result = scan_entries(cls.entries)
+
+    def test_current_duplicate_groups_are_protected_desc_conflicts(self) -> None:
+        duplicates = [
+            finding
+            for finding in self.result.findings
+            if finding.rule_id == "duplicate_trace_node"
+        ]
+        self.assertEqual(len(duplicates), 7)
+        self.assertEqual(
+            [finding.entry_id for finding in duplicates],
+            [
+                "entry-00115",
+                "entry-00182",
+                "entry-00199",
+                "entry-00224",
+                "entry-00484",
+                "entry-00487",
+                "entry-00507",
+            ],
+        )
+        self.assertTrue(all(finding.verify == 1 for finding in duplicates))
+        self.assertTrue(all(not finding.safe_duplicate for finding in duplicates))
+        decisions = decide_findings(
+            duplicates,
+            mode="fix",
+            duplicate_action="remove-safe",
+            order_action="report",
+        )
+        self.assertTrue(all(d.status == "protected_verified" for d in decisions))
+        self.assertTrue(all(not d.remove_indices for d in decisions))
+
+    def test_priority_samples_have_expected_classification(self) -> None:
+        actual = {
+            (finding.entry_id, finding.rule_id, finding.trace_index)
+            for finding in self.result.findings
+        }
+        expected = {
+            ("entry-00103", "before_entry_definite", 1),
+            ("entry-00103", "overlap_critical", 3),
+            ("entry-00185", "after_critical_definite", 2),
+            ("entry-00320", "before_entry_definite", 2),
+            ("entry-00511", "overlap_entry", 0),
+            ("entry-00511", "after_critical_definite", 4),
+            ("entry-00511", "overlap_critical", 5),
+        }
+        self.assertTrue(expected.issubset(actual))