feat: let a project plugin refine the findings of a run

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Plugins
 
 - Every plugin module is registered in `sys.modules` before it runs, so a single-file plugin may define a dataclass under `from __future__ import annotations` (#105).
+- `ProjectPlugin.refine(ctx, findings)` lets a project plugin reclassify the findings of a run: severity, confidence and added metadata only, with `refined_by` and the original values recorded; anything else raises `RefinementError` (#106).
 
 ## 0.3.0 (2026-09-22)
 

docs/plugins.md

@@ -162,6 +162,38 @@ symbols, as the shipped `sample-advisories` plugin does. Requirements matching t
 reported, calls to the affected symbols become reachable vulnerabilities and tainted
 calls become exploitable ones.
 
+### Reclassifying findings: `ProjectPlugin.refine`
+
+A project plugin may also implement `refine(ctx, findings)`, called once with every
+finding of the run after all plugins have reported. It returns the same findings, in
+the same order, and may only change their `severity` and `confidence` or add metadata:
+the rule, the location, the message and the evidence a finding already carries are not
+the refiner's to touch, and the engine rejects a refinement that changes or drops any
+of them with a `RefinementError`. A reclassified finding records `refined_by` and the
+`original_severity` or `original_confidence` it had, so a report always shows what the
+detector said and who changed it.
+
+```python
+from dataclasses import replace
+
+from coretrace_python.findings import Confidence
+from coretrace_python.plugins import ProjectContext, ProjectPlugin
+
+
+class InternalOnly(ProjectPlugin):
+    name = "internal-only"
+
+    def analyze_project(self, ctx: ProjectContext):
+        return ()
+
+    def refine(self, ctx: ProjectContext, findings):
+        return tuple(
+            replace(f, confidence=Confidence.LOW, metadata={**f.metadata, "exposure": "internal"})
+            if f.function in INTERNAL else f
+            for f in findings
+        )
+```
+
 ### Secret scanners: `SecretDetector`
 
 Judges every string literal of the module, or every value of a configuration file,

src/coretrace_python/engine.py

@@ -89,6 +89,7 @@
     PluginRegistry,
     ProjectContext,
     ProjectPlugin,
+    apply_refinement,
     discover_plugins,
     run_plugins,
 )
@@ -460,6 +461,9 @@ def analyze_project(
     for plugin in all_plugins:
         if isinstance(plugin, ProjectPlugin):
             findings.extend(plugin.analyze_project(context))
+    for plugin in all_plugins:
+        if isinstance(plugin, ProjectPlugin):
+            findings = list(apply_refinement(plugin, findings, plugin.refine(context, tuple(findings))))
     kept, suppressed = partition(apply_policy(policy, findings), _text_of(sources))
     return ProjectAnalysis(
         graph,

src/coretrace_python/plugins/__init__.py

@@ -7,6 +7,8 @@
     PluginContext,
     ProjectContext,
     ProjectPlugin,
+    RefinementError,
+    apply_refinement,
     run_plugins,
 )
 from coretrace_python.plugins.detectors import SymbolCallDetector, TaintDetector
@@ -44,11 +46,13 @@
     "PluginRegistry",
     "ProjectContext",
     "ProjectPlugin",
+    "RefinementError",
     "SecretDetector",
     "SecretPattern",
     "SymbolCallDetector",
     "TaintDetector",
     "VersionRange",
+    "apply_refinement",
     "discover_plugins",
     "literals",
     "load_manifest",

src/coretrace_python/plugins/api.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Mapping, Sequence
 from pathlib import Path
@@ -99,6 +100,50 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]:
     def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]:
         raise NotImplementedError
 
+    def refine(self, ctx: ProjectContext, findings: tuple[Finding, ...]) -> Sequence[Finding]:
+        """Reclassify the findings of the run, once every plugin has reported.
+
+        A refinement returns the same findings in the same order and may change only
+        their severity, their confidence and add metadata: what a finding is, where it
+        is and the evidence behind it are not the refiner's to touch. The engine records
+        who reclassified a finding and what it was before, and rejects anything else."""
+
+        return findings
+
+
+class RefinementError(Exception):
+    """A refinement did more than reclassify a finding."""
+
+
+REFINABLE = ("severity", "confidence")
+
+
+def apply_refinement(plugin: Plugin, original: Sequence[Finding], refined: Sequence[Finding]) -> tuple[Finding, ...]:
+    """The refined findings, checked against the contract of ``ProjectPlugin.refine`` and
+    stamped with a trace of the reclassification."""
+
+    if len(refined) != len(original):
+        raise RefinementError(f"plugin {plugin.name!r} returned {len(refined)} findings for {len(original)}")
+    kept: list[Finding] = []
+    for before, after in zip(original, refined, strict=True):
+        for attribute in ("rule_id", "message", "span", "function"):
+            if getattr(before, attribute) != getattr(after, attribute):
+                raise RefinementError(f"plugin {plugin.name!r} changed the {attribute} of a finding")
+        lost = [k for k, v in before.metadata.items() if after.metadata.get(k) != v]
+        if lost:
+            raise RefinementError(f"plugin {plugin.name!r} changed the evidence {lost} of a finding")
+        changed = [a for a in REFINABLE if getattr(before, a) != getattr(after, a)]
+        if not changed:
+            kept.append(after)
+            continue
+        trace = dict(after.metadata)
+        for attribute in changed:
+            trace.setdefault(f"original_{attribute}", getattr(before, attribute).value)
+        refiners = trace.get("refined_by", "")
+        trace["refined_by"] = f"{refiners},{plugin.name}" if refiners else plugin.name
+        kept.append(dataclasses.replace(after, metadata=trace))
+    return tuple(kept)
+
 
 class PluginContext:
     """What one plugin may see while analysing one module."""

tests/test_project.py

@@ -299,3 +299,60 @@ def test_check_accepts_a_directory(tmp_path: Path, capsys) -> None:  # type: ign
 def test_emit_ir_rejects_directories(tmp_path: Path, capsys) -> None:  # type: ignore[no-untyped-def]
     assert main(["--emit-ir", str(tmp_path)]) == 2
     assert "directory" in capsys.readouterr().err
+
+
+# --------------------------------------------------------------------------- refinement
+
+
+def test_a_project_plugin_may_refine_the_findings_of_a_run(tmp_path: Path) -> None:
+    from dataclasses import replace
+
+    from coretrace_python.findings import Severity
+    from coretrace_python.plugins import ProjectContext, ProjectPlugin
+
+    class Exposure(ProjectPlugin):
+        name = "exposure"
+
+        def analyze_project(self, ctx: ProjectContext) -> tuple[Finding, ...]:
+            return ()
+
+        def refine(self, ctx: ProjectContext, findings: tuple[Finding, ...]) -> tuple[Finding, ...]:
+            return tuple(
+                replace(f, severity=Severity.LOW, metadata={**f.metadata, "exposure": "internal"}) for f in findings
+            )
+
+    root = project(tmp_path, {"app.py": "import os\n\ndef run():\n    os.system(input())\n"})
+
+    findings = engine.analyze_project(root, [PLUGINS], plugins=[Exposure()]).findings
+
+    found = next(f for f in findings if f.rule_id == "command-injection")
+    assert found.severity is Severity.LOW
+    assert found.metadata["exposure"] == "internal"
+    assert found.metadata["refined_by"] == "exposure"
+    assert found.metadata["original_severity"] == "high"
+    assert found.metadata["source"] == "python.builtins.input"
+
+
+@pytest.mark.parametrize("tamper", ["identity", "evidence", "drop"])
+def test_a_refinement_may_only_reclassify(tmp_path: Path, tamper: str) -> None:
+    from dataclasses import replace
+
+    from coretrace_python.plugins import ProjectContext, ProjectPlugin, RefinementError
+
+    class Tampering(ProjectPlugin):
+        name = "tampering"
+
+        def analyze_project(self, ctx: ProjectContext) -> tuple[Finding, ...]:
+            return ()
+
+        def refine(self, ctx: ProjectContext, findings: tuple[Finding, ...]) -> tuple[Finding, ...]:
+            if tamper == "identity":
+                return tuple(replace(f, rule_id="other") for f in findings)
+            if tamper == "evidence":
+                return tuple(replace(f, metadata={k: v for k, v in f.metadata.items() if k != "source"}) for f in findings)
+            return findings[1:]
+
+    root = project(tmp_path, {"app.py": "import os\n\ndef run():\n    os.system(input())\n"})
+
+    with pytest.raises(RefinementError, match="tampering"):
+        engine.analyze_project(root, [PLUGINS], plugins=[Tampering()])