From 840909864001f8cfea45c3f3b71e328a9976ec5e Mon Sep 17 00:00:00 2001 From: Akangsha Goel Date: Mon, 21 Sep 2026 11:56:34 -0700 Subject: [PATCH] feat(mcp_readability): decide what a run can reuse from its baseline Given a baseline and this run's fingerprints, work out which tools still hold and which have to be re-judged, merge the carried findings with the fresh ones, and record where each finding came from. Pure functions only: nothing calls them yet, so a run still judges every endpoint in full. --- .../scorers/mcp_readability/carry_forward.py | 587 +++++++++++++++ evalbench/test/mcp_carry_forward_test.py | 684 ++++++++++++++++++ 2 files changed, 1271 insertions(+) create mode 100644 evalbench/scorers/mcp_readability/carry_forward.py create mode 100644 evalbench/test/mcp_carry_forward_test.py diff --git a/evalbench/scorers/mcp_readability/carry_forward.py b/evalbench/scorers/mcp_readability/carry_forward.py new file mode 100644 index 00000000..77274657 --- /dev/null +++ b/evalbench/scorers/mcp_readability/carry_forward.py @@ -0,0 +1,587 @@ +"""Deciding what to re-judge, and merging carried findings with fresh ones. + +The invariant: same endpoint, same per-tool fingerprints and same judge +fingerprint means byte-identical feedback and zero model calls. Every departure +is reported with a named reason (see CHANGE_REASONS). + +Lives under scorers/ because evaluator.mcp_readability imports the orchestrator, +which imports the scorers, so Baseline is imported under TYPE_CHECKING only. +""" + +from collections.abc import Mapping, Sequence +import copy +from dataclasses import dataclass, field +import hashlib +import re +from typing import TYPE_CHECKING, Any + +from scorers.mcp_readability.fingerprint import component_diff + +if TYPE_CHECKING: + from evaluator.mcp_readability.baseline import Baseline + + +# How this run's feedback was produced. +MODE_FULL_JUDGE = "full_judge" # every tool judged +MODE_PARTIAL = "partial" # some tools judged, the rest carried +MODE_CARRIED = "carried" # nothing judged, no model call + + +# Why this run's feedback may differ from the previous one. Exactly one is +# reported per endpoint, in mcp_readability_change_reason. +UNCHANGED = "unchanged" +TOOLS_CHANGED = "tools_changed" +STYLE_GUIDE_CHANGED = "style_guide_changed" +MODEL_CHANGED = "model_changed" +WAIVERS_CHANGED = "waivers_changed" +PROMPT_CHANGED = "prompt_changed" +BASELINE_EXPIRED = "baseline_expired" +FORCED_REFRESH = "forced_refresh" +ENDPOINT_IDENTITY_CHANGED = "endpoint_identity_changed" +BASELINE_UNAVAILABLE = "baseline_unavailable" +NO_BASELINE = "no_baseline" + +CHANGE_REASONS = ( + UNCHANGED, + TOOLS_CHANGED, + STYLE_GUIDE_CHANGED, + MODEL_CHANGED, + WAIVERS_CHANGED, + PROMPT_CHANGED, + BASELINE_EXPIRED, + FORCED_REFRESH, + ENDPOINT_IDENTITY_CHANGED, + BASELINE_UNAVAILABLE, + NO_BASELINE, +) + + +# Which judge component maps to which reported reason. Ordered: when several +# components change at once the earliest match wins, so the most consequential +# (and most likely to explain a count swing) is the one reported. +_COMPONENT_REASONS = ( + ("judge_model", MODEL_CHANGED), + ("style_guide_sha", STYLE_GUIDE_CHANGED), + ("prompt_version", PROMPT_CHANGED), + ("prompt_sha", PROMPT_CHANGED), + ("scorer_name", PROMPT_CHANGED), + ("exceptions", WAIVERS_CHANGED), + ("product_name", ENDPOINT_IDENTITY_CHANGED), +) + +# The judge's entry for issues that belong to no individual tool. +GENERAL = "general" + +# Severity order within a tool's entry, as the judge is asked to emit it. +_SEVERITY_ORDER = ("P0", "P1", "P2") + + +@dataclass +class BaselineContext: + """What the orchestrator hands the scorer so it can skip work. + + override_reason is set when the baseline must not be used (expiry, forced + refresh, unreadable store); the scorer then re-judges in full and reports + that reason. + """ + + endpoint_key: str = "" + job_id: str = "" + tool_fingerprints: dict[str, str] = field(default_factory=dict) + baseline: "Baseline | None" = None + override_reason: str = "" + + +@dataclass +class Decision: + """What to judge for one endpoint, and how to explain the outcome.""" + + mode: str = MODE_FULL_JUDGE + change_reason: str = NO_BASELINE + rejudged_tools: list[str] = field(default_factory=list) + carried_tools: list[str] = field(default_factory=list) + added_tools: list[str] = field(default_factory=list) + removed_tools: list[str] = field(default_factory=list) + component_diff: list[str] = field(default_factory=list) + judge_components: dict[str, Any] = field(default_factory=dict) + baseline: "Baseline | None" = None + + @property + def needs_model_call(self) -> bool: + """Whether this run still has to call the judge.""" + return self.mode != MODE_CARRIED + + @property + def names_changed(self) -> bool: + """Whether the set of tool names changed, not just their contents.""" + return bool(self.added_tools or self.removed_tools) + + @property + def accepted_tools(self) -> set[str]: + """Per-tool entries the judge's output is trusted for. + + Excludes general, which _merge_partial decides on separately. + """ + return set(self.rejudged_tools) | set(self.added_tools) + + @property + def tools_to_report(self) -> list[str]: + """What the judge is asked to report on in a partial run. + + Includes general when the tool-name set changed, since a rename or + removal is what creates a cross-tool finding. + """ + names = self.accepted_tools + if self.names_changed: + names.add(GENERAL) + return sorted(names) + + +def _full_judge(context: BaselineContext, change_reason: str) -> Decision: + """Re-judge every tool, keeping the baseline as the comparison point.""" + return Decision( + mode=MODE_FULL_JUDGE, + change_reason=change_reason, + rejudged_tools=list(context.tool_fingerprints or {}), + baseline=context.baseline, + ) + + +def decide(context: BaselineContext | None, judge_fingerprint: str) -> Decision: + """Choose full / partial / carried for one endpoint.""" + if context is None: + return Decision(mode=MODE_FULL_JUDGE, change_reason=NO_BASELINE) + + baseline = context.baseline + if context.override_reason: + # Findings are not reused, but the baseline is still the reference for + # what changed since last time. + return _full_judge(context, context.override_reason) + if baseline is None: + return _full_judge(context, NO_BASELINE) + if judge_fingerprint != baseline.judge_fingerprint: + return _full_judge(context, PROMPT_CHANGED) + + current_fingerprints = context.tool_fingerprints or {} + previous_fingerprints = baseline.tool_fingerprints or {} + added = [ + tool + for tool in current_fingerprints + if tool not in previous_fingerprints + ] + removed = [ + tool + for tool in previous_fingerprints + if tool not in current_fingerprints + ] + changed = [ + tool + for tool in current_fingerprints + if tool in previous_fingerprints + and current_fingerprints[tool] != previous_fingerprints[tool] + ] + unchanged = [ + tool + for tool in current_fingerprints + if tool in previous_fingerprints + and current_fingerprints[tool] == previous_fingerprints[tool] + ] + + if not added and not removed and not changed: + return Decision( + mode=MODE_CARRIED, + change_reason=UNCHANGED, + carried_tools=unchanged, + baseline=baseline, + ) + + return Decision( + mode=MODE_PARTIAL, + change_reason=TOOLS_CHANGED, + rejudged_tools=changed, + carried_tools=unchanged, + added_tools=added, + removed_tools=removed, + baseline=baseline, + ) + + +def decide_with_components( + context: BaselineContext | None, + judge_fingerprint: str, + judge_components: Mapping[str, Any], +) -> Decision: + """decide(), with the component diff computed against context.""" + decision = decide(context, judge_fingerprint) + decision.judge_components = dict(judge_components or {}) + baseline = context.baseline if context else None + if ( + baseline is not None + and decision.mode == MODE_FULL_JUDGE + and not context.override_reason + ): + if not baseline.judge_components: + # Diffing against {} would report every key as changed, so say + # plainly that it could not be compared. + decision.component_diff = [] + decision.change_reason = BASELINE_UNAVAILABLE + else: + decision.component_diff = component_diff( + baseline.judge_components, judge_components + ) + decision.change_reason = _reason_for_components( + decision.component_diff + ) + return decision + + +def _reason_for_components(diff: Sequence[str]) -> str: + """Map a component diff to the single reason reported for the run.""" + for component, reason in _COMPONENT_REASONS: + if component in diff: + return reason + if diff: + # A component that predates this mapping; the superset is accurate. + return PROMPT_CHANGED + # The fingerprint differs but no individual component does. A baseline + # exists, so this is not a first run -- it simply cannot be compared. + return BASELINE_UNAVAILABLE + + +def unreported_findings( + decision: Decision, judged: Mapping[str, Any] | None +) -> list[dict[str, Any]]: + """Baseline findings on re-judged tools that this run's judge did not repeat. + + Either the change resolved the issue or the judge did not mention it, so + they go to a second, narrower pass. Returned in findings_by_tool shape to + feed straight back into merge_feedback. + """ + baseline = decision.baseline + if decision.mode != MODE_PARTIAL or baseline is None: + return [] + + scope = set(decision.rejudged_tools) + if decision.names_changed: + scope.add(GENERAL) + reported = set(_by_id(_entries(judged or {}))) + + unreported = [] + for entry in _entries(baseline.feedback or {}): + if entry["tool"] not in scope: + continue + missing = [ + finding + for finding in entry["findings"] + if finding.get("finding_id") not in reported + ] + if missing: + unreported.append({"tool": entry["tool"], "findings": missing}) + return unreported + + +def merge_feedback( + decision: Decision, + judged: Mapping[str, Any] | None, + tool_order: Sequence[str], + restored: Sequence[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Combine carried baseline findings with the judge's fresh ones. + + judged is None in carried mode. restored holds findings the reconciliation + pass ruled still apply; they keep their original wording. Entries follow + man-page order with general first. The caller recomputes the counts. + """ + baseline = decision.baseline + baseline_feedback = (baseline.feedback if baseline else {}) or {} + + if decision.mode == MODE_CARRIED: + merged = { + "findings_by_tool": _entries(baseline_feedback), + "waived": baseline_feedback.get("waived") or [], + "summary": baseline_feedback.get("summary", ""), + } + elif decision.mode == MODE_PARTIAL: + merged = _merge_partial(decision, judged or {}, baseline_feedback) + else: + merged = { + "findings_by_tool": _entries(judged or {}), + "waived": (judged or {}).get("waived") or [], + "summary": (judged or {}).get("summary", ""), + } + + if restored: + _restore(merged["findings_by_tool"], restored) + merged["findings_by_tool"] = _ordered(merged["findings_by_tool"], tool_order) + return merged + + +def _restore( + entries: list[dict[str, Any]], restored: Sequence[dict[str, Any]] +) -> None: + """Put reconciled findings back under their tool, in place. + + A tool whose every finding went unreported has no entry to rejoin, so one + is created. Touched entries are re-sorted so a restored P0 does not + trail a P2. + """ + by_tool = {entry["tool"]: entry for entry in entries} + for entry in restored: + findings = [ + finding + for finding in entry.get("findings", []) + if isinstance(finding, dict) + ] + if not findings: + continue + target = by_tool.get(entry["tool"]) + if target is None: + target = {"tool": entry["tool"], "findings": []} + by_tool[entry["tool"]] = target + entries.append(target) + known = { + finding.get("finding_id") for finding in target["findings"] + } + target["findings"].extend( + finding + for finding in findings + if finding.get("finding_id") not in known + ) + target["findings"].sort(key=_severity_rank) + + +def _severity_rank(finding: Mapping[str, Any]) -> int: + """Sort key placing P0 first and unknown severities last.""" + severity = str(finding.get("severity", "")).upper() + if severity in _SEVERITY_ORDER: + return _SEVERITY_ORDER.index(severity) + return len(_SEVERITY_ORDER) + + +def _merge_partial( + decision: Decision, + judged: Mapping[str, Any], + baseline_feedback: Mapping[str, Any], +) -> dict[str, Any]: + """Accept judge entries only for changed or added tools; carry the rest. + + Filtering here, not the prompt, is what provides the guarantee; the prompt + clause only trims output tokens. + """ + accept = decision.accepted_tools + carry = set(decision.carried_tools) + baseline_entries = _entries(baseline_feedback) + judged_entries = _entries(judged) + + merged_entries = [] + for entry in judged_entries: + if entry["tool"] in accept: + merged_entries.append(entry) + _stabilize(merged_entries, baseline_entries) + for entry in baseline_entries: + if entry["tool"] in carry: + merged_entries.append(entry) + + # A fresh "general" entry is only trustworthy when the set of tool names + # changed, since a rename, addition or removal is what creates a cross-tool + # inconsistency. When names did change it is re-judged, so its absence is + # ambiguous and goes to the second pass rather than being read as fixed. + general = [ + entry + for entry in ( + judged_entries if decision.names_changed else baseline_entries + ) + if entry["tool"] == GENERAL + ] + _stabilize(general, baseline_entries) + merged_entries.extend(general) + + return { + "findings_by_tool": merged_entries, + "waived": judged.get("waived") or baseline_feedback.get("waived") or [], + "summary": judged.get("summary", "") or baseline_feedback.get( + "summary", "" + ), + } + + +def _stabilize( + entries: list[dict[str, Any]], baseline_entries: Sequence[dict[str, Any]] +) -> None: + """Give surviving findings back their previous wording, in place. + + A re-judged tool is described from scratch, so an untouched issue comes + back reworded and the report moves for no reason. A matching finding_id + means the same rule on the same locator, so the baseline's text is kept. + """ + previous_by_id = _by_id(baseline_entries) + if not previous_by_id: + return + for entry in entries: + entry["findings"] = [ + previous_by_id.get(finding.get("finding_id"), finding) + for finding in entry["findings"] + ] + + +def _by_id(entries: Sequence[dict[str, Any]]) -> dict[str, dict[str, Any]]: + """Index findings by finding_id. Ids embed the tool, so one map covers all.""" + return { + finding["finding_id"]: finding + for entry in entries + for finding in entry.get("findings", []) + if isinstance(finding, dict) and finding.get("finding_id") + } + + +def _entries(feedback: Mapping[str, Any] | None) -> list[dict[str, Any]]: + """Return the usable per-tool entries, deep-copied. + + A Baseline is shared by every caller and the merged findings are mutated + later, so returning the store's own dicts would let one endpoint write into + another's baseline. Entries with no findings are dropped. + """ + raw = (feedback or {}).get("findings_by_tool") + if not isinstance(raw, list): + return [] + entries = [] + for entry in raw: + if not isinstance(entry, dict): + continue + tool = str(entry.get("tool", "")).strip() + findings = entry.get("findings") + if not tool or not isinstance(findings, list): + continue + findings = [ + finding for finding in findings if isinstance(finding, dict) + ] + if findings: + entries.append({"tool": tool, "findings": copy.deepcopy(findings)}) + return entries + + +def _ordered( + entries: Sequence[dict[str, Any]], tool_order: Sequence[str] +) -> list[dict[str, Any]]: + """Order entries: general first, then man-page order, then anything else.""" + rank = {name: i for i, name in enumerate(tool_order)} + rank[GENERAL] = -1 + fallback = len(tool_order) + return sorted(entries, key=lambda entry: rank.get(entry["tool"], fallback)) + + +def mint_finding_ids(entries: list[dict[str, Any]]) -> None: + """Assign a stable finding_id to every finding, in place. + + (tool, rule_id) is not unique, so the optional locator and then the title + disambiguate, with an ordinal as the last resort. Carried findings keep the + id they arrived with, so identity only holds from the first carried run. + """ + seen: dict[str, int] = {} + for entry in entries: + tool = entry.get("tool", "") + for finding in entry.get("findings", []): + if not isinstance(finding, dict) or finding.get("finding_id"): + continue + base = _finding_id(tool, finding) + count = seen.get(base, 0) + seen[base] = count + 1 + finding["finding_id"] = base if count == 0 else f"{base}-{count}" + + +def _finding_id(tool: str, finding: Mapping[str, Any]) -> str: + """Hash a finding's tool, rule and discriminator into a short id.""" + discriminator = finding.get("locator") or finding.get("title") or "" + raw = "|".join( + [ + str(tool), + str(finding.get("rule_id", "")), + _normalize(str(discriminator)), + ] + ) + return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:12] + + +def _normalize(text: str) -> str: + """Collapse whitespace and case so wording churn keeps the same id.""" + return re.sub(r"\s+", " ", text).strip().lower() + + +def finding_ids(entries: Sequence[dict[str, Any]]) -> set[str]: + """Return every finding_id present in a list of per-tool entries.""" + return { + finding["finding_id"] + for entry in entries + for finding in entry.get("findings", []) + if isinstance(finding, dict) and finding.get("finding_id") + } + + +def build_provenance( + decision: Decision, merged_entries: Sequence[dict[str, Any]] +) -> dict[str, Any]: + """Build the provenance block recorded for this run and shown in HTML. + + Records where each tool's findings came from and which are new versus + resolved. Persisted to mcp_readability_feedback_provenance_json only. + """ + baseline = decision.baseline + previous_entries = _entries(baseline.feedback if baseline else {}) + previous_ids = finding_ids(previous_entries) + current_ids = finding_ids(merged_entries) + + # Markers only mean something when some tools were spared; in a full judge + # the banner already says why everything was re-judged. + tool_provenance = {} + if decision.mode != MODE_FULL_JUDGE: + for tool in decision.carried_tools: + tool_provenance[tool] = "carried" + for tool in decision.rejudged_tools: + tool_provenance[tool] = "rejudged" + for tool in decision.added_tools: + tool_provenance[tool] = "new" + + return { + "component_changes": _component_changes(decision), + "mode": decision.mode, + "change_reason": decision.change_reason, + "baseline_job_id": baseline.job_id if baseline else "", + "baseline_timestamp": baseline.check_timestamp if baseline else "", + "rejudged_tools": sorted(decision.rejudged_tools), + "carried_tools": sorted(decision.carried_tools), + "added_tools": sorted(decision.added_tools), + "removed_tools": sorted(decision.removed_tools), + "component_diff": decision.component_diff, + "tool_provenance": tool_provenance, + "new_finding_ids": sorted(current_ids - previous_ids), + "resolved_finding_ids": sorted(previous_ids - current_ids), + "previous_finding_count": _count(previous_entries), + "finding_count": _count(merged_entries), + } + + +def _component_changes(decision: Decision) -> dict[str, dict[str, Any]]: + """Return before/after values for the scalar judge components that changed. + + Non-scalar components, such as the waiver list, are reported as changed + without their contents. + """ + baseline = decision.baseline + previous_components = (baseline.judge_components if baseline else {}) or {} + current_components = decision.judge_components or {} + changes = {} + for component in decision.component_diff: + before = previous_components.get(component) + after = current_components.get(component) + if isinstance(before, (str, int, float, type(None))) and isinstance( + after, (str, int, float, type(None)) + ): + changes[component] = {"from": before, "to": after} + else: + changes[component] = {} + return changes + + +def _count(entries: Sequence[dict[str, Any]]) -> int: + """Total findings across every entry.""" + return sum(len(entry.get("findings", [])) for entry in entries) diff --git a/evalbench/test/mcp_carry_forward_test.py b/evalbench/test/mcp_carry_forward_test.py new file mode 100644 index 00000000..4306187e --- /dev/null +++ b/evalbench/test/mcp_carry_forward_test.py @@ -0,0 +1,684 @@ +"""Unit tests for the readability carry-forward decision and merge. + +Two failures matter here and they fail in opposite directions. Carrying a +finding that should have been re-judged reports a stale surface as current; +re-judging a tool nobody touched puts the counts back at the mercy of the +model's non-determinism, which is the whole reason any of this exists. The +tests below pin each decision to the input that is supposed to drive it. +""" + +import unittest + +from evaluator.mcp_readability.baseline import Baseline +from scorers.mcp_readability import carry_forward as cf + + +_JUDGE = "judge-fp" + +# Distinguishes "no baseline" from "default baseline" in _context below. +_DEFAULT = object() + + +def _baseline(tool_fingerprints=None, findings=None, **overrides): + values = { + "endpoint_key": "AlloyDB|http://x|PROD", + "job_id": "job-old", + "check_timestamp": "2026-09-01T00:00:00+00:00", + "judge_fingerprint": _JUDGE, + "judge_components": {"judge_model": "m", "style_guide_sha": "s"}, + "tool_fingerprints": tool_fingerprints or {"a": "fp-a", "b": "fp-b"}, + "feedback": { + "findings_by_tool": findings + if findings is not None + else [ + {"tool": "a", "findings": [_finding("R1", "A finding")]}, + {"tool": "b", "findings": [_finding("R2", "B finding")]}, + ], + "waived": [{"rule_id": "w", "reason": "why"}], + "summary": "previous summary", + }, + "readability_score": 70, + } + values.update(overrides) + # Ids are minted when a finding is generated, so a real baseline always + # arrives carrying them; without that, every carried finding would look new. + cf.mint_finding_ids(values["feedback"].get("findings_by_tool") or []) + return Baseline(**values) + + +def _finding(rule_id, title, **extra): + finding = { + "severity": "P1", + "rule_id": rule_id, + "title": title, + "message": "m", + "suggestion": "s", + } + finding.update(extra) + return finding + + +def _context(current=None, baseline=_DEFAULT, **overrides): + return cf.BaselineContext( + endpoint_key="AlloyDB|http://x|PROD", + job_id="job-new", + tool_fingerprints=( + current if current is not None else {"a": "fp-a", "b": "fp-b"} + ), + baseline=_baseline() if baseline is _DEFAULT else baseline, + **overrides, + ) + + +class DecideTest(unittest.TestCase): + + def test_no_context_is_a_full_judge(self): + decision = cf.decide(None, _JUDGE) + self.assertEqual(decision.mode, cf.MODE_FULL_JUDGE) + self.assertEqual(decision.change_reason, cf.NO_BASELINE) + self.assertTrue(decision.needs_model_call) + + def test_no_baseline_judges_every_current_tool(self): + decision = cf.decide(_context(baseline=None), _JUDGE) + self.assertEqual(decision.mode, cf.MODE_FULL_JUDGE) + self.assertEqual(decision.change_reason, cf.NO_BASELINE) + self.assertEqual(sorted(decision.rejudged_tools), ["a", "b"]) + + def test_identical_surface_makes_no_model_call(self): + decision = cf.decide(_context(), _JUDGE) + self.assertEqual(decision.mode, cf.MODE_CARRIED) + self.assertEqual(decision.change_reason, cf.UNCHANGED) + self.assertEqual(sorted(decision.carried_tools), ["a", "b"]) + self.assertFalse(decision.needs_model_call) + + def test_only_the_changed_tool_is_rejudged(self): + decision = cf.decide( + _context({"a": "fp-a", "b": "EDITED"}), _JUDGE + ) + self.assertEqual(decision.mode, cf.MODE_PARTIAL) + self.assertEqual(decision.change_reason, cf.TOOLS_CHANGED) + self.assertEqual(decision.rejudged_tools, ["b"]) + self.assertEqual(decision.carried_tools, ["a"]) + + def test_added_and_removed_tools_are_reported_separately(self): + decision = cf.decide(_context({"a": "fp-a", "c": "fp-c"}), _JUDGE) + self.assertEqual(decision.added_tools, ["c"]) + self.assertEqual(decision.removed_tools, ["b"]) + self.assertTrue(decision.names_changed) + # A removal alone still leaves the surviving tools carried. + self.assertEqual(decision.carried_tools, ["a"]) + + def test_a_changed_judge_invalidates_every_tool(self): + """A new model or guide can judge an untouched tool differently.""" + decision = cf.decide(_context(), "different-judge-fp") + self.assertEqual(decision.mode, cf.MODE_FULL_JUDGE) + self.assertEqual(sorted(decision.rejudged_tools), ["a", "b"]) + self.assertEqual(decision.carried_tools, []) + + def test_override_reason_forces_a_full_judge_but_keeps_the_baseline(self): + """Expiry and forced refresh still need "what changed since" to report.""" + decision = cf.decide( + _context(override_reason=cf.BASELINE_EXPIRED), _JUDGE + ) + self.assertEqual(decision.mode, cf.MODE_FULL_JUDGE) + self.assertEqual(decision.change_reason, cf.BASELINE_EXPIRED) + self.assertIsNotNone(decision.baseline) + + def test_reported_tools_include_general_only_when_names_changed(self): + renamed = cf.decide(_context({"a": "fp-a", "c": "fp-c"}), _JUDGE) + self.assertIn(cf.GENERAL, renamed.tools_to_report) + edited = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + self.assertNotIn(cf.GENERAL, edited.tools_to_report) + # general is never trusted as a per-tool entry. + self.assertNotIn(cf.GENERAL, renamed.accepted_tools) + + +class DecideWithComponentsTest(unittest.TestCase): + + def _components(self, **overrides): + components = {"judge_model": "m", "style_guide_sha": "s"} + components.update(overrides) + return components + + def test_unchanged_components_leave_the_carried_decision_alone(self): + decision = cf.decide_with_components( + _context(), _JUDGE, self._components() + ) + self.assertEqual(decision.mode, cf.MODE_CARRIED) + self.assertEqual(decision.change_reason, cf.UNCHANGED) + + def test_the_changed_component_names_the_reason(self): + for override, reason in [ + ({"judge_model": "other"}, cf.MODEL_CHANGED), + ({"style_guide_sha": "other"}, cf.STYLE_GUIDE_CHANGED), + ({"exceptions": [{"rule_id": "r"}]}, cf.WAIVERS_CHANGED), + ({"prompt_version": "2"}, cf.PROMPT_CHANGED), + ]: + with self.subTest(component=sorted(override)[0]): + decision = cf.decide_with_components( + _context(), "new-fp", self._components(**override) + ) + self.assertEqual(decision.change_reason, reason) + + def test_the_most_consequential_component_wins(self): + """Several inputs can move at once; one reason is reported.""" + decision = cf.decide_with_components( + _context(), + "new-fp", + self._components(judge_model="other", style_guide_sha="other"), + ) + self.assertEqual(decision.change_reason, cf.MODEL_CHANGED) + + def test_a_baseline_without_components_says_so(self): + """Diffing against {} would blame whichever key sorts first.""" + decision = cf.decide_with_components( + _context(baseline=_baseline(judge_components={})), + "new-fp", + self._components(), + ) + self.assertEqual(decision.change_reason, cf.BASELINE_UNAVAILABLE) + self.assertEqual(decision.component_diff, []) + + def test_an_unmapped_component_falls_back_to_prompt_changed(self): + self.assertEqual( + cf._reason_for_components(["something_new"]), cf.PROMPT_CHANGED + ) + + def test_a_fingerprint_mismatch_with_no_component_diff_is_unavailable(self): + self.assertEqual( + cf._reason_for_components([]), cf.BASELINE_UNAVAILABLE + ) + + +class MergeFeedbackTest(unittest.TestCase): + + def _judged(self, entries, summary="fresh summary"): + # The scorer mints ids as it parses the judge's response, so the merge + # never sees an unidentified fresh finding. Matching a survivor against + # the baseline depends on that, so the fixture has to do it too. + cf.mint_finding_ids(entries) + return { + "findings_by_tool": entries, + "waived": [{"rule_id": "w2", "reason": "fresh"}], + "summary": summary, + } + + def test_carried_mode_returns_the_baseline_verbatim(self): + decision = cf.decide(_context(), _JUDGE) + merged = cf.merge_feedback(decision, None, ["a", "b"]) + self.assertEqual( + [entry["tool"] for entry in merged["findings_by_tool"]], ["a", "b"] + ) + self.assertEqual(merged["summary"], "previous summary") + + def test_a_partial_run_keeps_fresh_findings_only_for_changed_tools(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + judged = self._judged( + [ + {"tool": "a", "findings": [_finding("R9", "should be ignored")]}, + {"tool": "b", "findings": [_finding("R3", "fresh b")]}, + ] + ) + merged = cf.merge_feedback(decision, judged, ["a", "b"]) + by_tool = { + entry["tool"]: entry["findings"] + for entry in merged["findings_by_tool"] + } + self.assertEqual(by_tool["a"][0]["title"], "A finding") + self.assertEqual(by_tool["b"][0]["title"], "fresh b") + + def test_a_surviving_finding_keeps_its_previous_wording(self): + """A re-judged tool is described afresh; the report must not move for it.""" + baseline = _baseline( + findings=[ + { + "tool": "b", + "findings": [ + _finding("R2", "B finding", locator="timeout_ms") + ], + } + ] + ) + decision = cf.decide( + _context({"a": "fp-a", "b": "EDITED"}, baseline=baseline), _JUDGE + ) + merged = cf.merge_feedback( + decision, + self._judged( + [{ + "tool": "b", + "findings": [ + _finding( + "R2", + "the judge reworded this", + locator="timeout_ms", + message="and this", + ) + ], + }] + ), + ["a", "b"], + ) + finding = merged["findings_by_tool"][0]["findings"][0] + self.assertEqual(finding["title"], "B finding") + self.assertEqual(finding["message"], "m") + + def test_a_genuinely_new_finding_is_taken_from_the_judge(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + merged = cf.merge_feedback( + decision, + self._judged( + [{"tool": "b", "findings": [_finding("R7", "brand new")]}] + ), + ["a", "b"], + ) + by_tool = { + entry["tool"]: entry["findings"] + for entry in merged["findings_by_tool"] + } + self.assertEqual(by_tool["b"][0]["title"], "brand new") + + def test_the_filter_not_the_prompt_is_what_guarantees_carrying(self): + """The judge is shown every tool, so it can volunteer extra entries.""" + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + judged = self._judged( + [{"tool": "a", "findings": [_finding("R9", "unsolicited")]}] + ) + merged = cf.merge_feedback(decision, judged, ["a", "b"]) + titles = [ + finding["title"] + for entry in merged["findings_by_tool"] + for finding in entry["findings"] + ] + self.assertNotIn("unsolicited", titles) + + def test_a_removed_tool_drops_out_of_the_merged_findings(self): + decision = cf.decide(_context({"a": "fp-a"}), _JUDGE) + merged = cf.merge_feedback(decision, self._judged([]), ["a"]) + self.assertEqual( + [entry["tool"] for entry in merged["findings_by_tool"]], ["a"] + ) + + def test_general_is_carried_unless_the_tool_names_changed(self): + baseline = _baseline( + findings=[ + {"tool": "a", "findings": [_finding("R1", "A finding")]}, + { + "tool": cf.GENERAL, + "findings": [_finding("R0", "old cross-tool")], + }, + ] + ) + edited = cf.decide( + _context({"a": "EDITED", "b": "fp-b"}, baseline=baseline), _JUDGE + ) + merged = cf.merge_feedback(edited, self._judged([]), ["a", "b"]) + self.assertEqual( + merged["findings_by_tool"][0]["findings"][0]["title"], + "old cross-tool", + ) + + renamed = cf.decide( + _context({"a": "fp-a", "c": "fp-c"}, baseline=baseline), _JUDGE + ) + fresh = cf.merge_feedback( + renamed, + self._judged( + [{ + "tool": cf.GENERAL, + "findings": [_finding("R0", "new cross-tool")], + }] + ), + ["a", "c"], + ) + self.assertEqual( + fresh["findings_by_tool"][0]["findings"][0]["title"], + "new cross-tool", + ) + + def test_entries_follow_man_page_order_with_general_first(self): + baseline = _baseline( + tool_fingerprints={"a": "fp-a", "b": "fp-b"}, + findings=[ + {"tool": "b", "findings": [_finding("R2", "B")]}, + {"tool": "a", "findings": [_finding("R1", "A")]}, + {"tool": cf.GENERAL, "findings": [_finding("R0", "G")]}, + ], + ) + decision = cf.decide(_context(baseline=baseline), _JUDGE) + merged = cf.merge_feedback(decision, None, ["a", "b"]) + self.assertEqual( + [entry["tool"] for entry in merged["findings_by_tool"]], + [cf.GENERAL, "a", "b"], + ) + + def test_merging_does_not_mutate_the_baseline(self): + """One store serves every endpoint; a merge must not write into it.""" + baseline = _baseline() + decision = cf.decide(_context(baseline=baseline), _JUDGE) + merged = cf.merge_feedback(decision, None, ["a", "b"]) + merged["findings_by_tool"][0]["findings"][0]["title"] = "rewritten" + original = baseline.feedback["findings_by_tool"][0]["findings"][0] + self.assertEqual(original["title"], "A finding") + + def test_entries_without_findings_are_dropped(self): + decision = cf.decide(_context(baseline=None), _JUDGE) + merged = cf.merge_feedback( + decision, + self._judged([{"tool": "a", "findings": []}]), + ["a"], + ) + self.assertEqual(merged["findings_by_tool"], []) + + +class UnreportedFindingsTest(unittest.TestCase): + """What the second pass is asked to rule on. + + Over-reporting here costs a model call on findings code could have decided; + under-reporting silently resolves an issue that still exists, which is the + failure the second pass exists to prevent. + """ + + def _judged(self, entries): + cf.mint_finding_ids(entries) + return {"findings_by_tool": entries} + + def test_a_dropped_finding_on_a_rejudged_tool_is_returned(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + unreported = cf.unreported_findings(decision, self._judged([])) + self.assertEqual( + [ + (entry["tool"], entry["findings"][0]["title"]) + for entry in unreported + ], + [("b", "B finding")], + ) + + def test_a_carried_tool_is_never_up_for_reconciliation(self): + """Tool a was not re-judged, so its silence means nothing.""" + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + unreported = cf.unreported_findings(decision, self._judged([])) + self.assertNotIn("a", [entry["tool"] for entry in unreported]) + + def test_nothing_is_returned_when_the_judge_repeated_everything(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + unreported = cf.unreported_findings( + decision, + self._judged( + [{"tool": "b", "findings": [_finding("R2", "B finding")]}] + ), + ) + self.assertEqual(unreported, []) + + def test_an_added_tool_has_nothing_to_reconcile(self): + decision = cf.decide(_context({"a": "fp-a", "b": "fp-b", "c": "fp-c"}), + _JUDGE) + self.assertEqual(cf.unreported_findings(decision, self._judged([])), []) + + def test_a_carried_run_never_reaches_the_second_pass(self): + decision = cf.decide(_context(), _JUDGE) + self.assertEqual(decision.mode, cf.MODE_CARRIED) + self.assertEqual(cf.unreported_findings(decision, None), []) + + def test_general_is_in_scope_only_when_the_tool_names_changed(self): + baseline = _baseline( + findings=[ + {"tool": cf.GENERAL, "findings": [_finding("R0", "cross-tool")]} + ] + ) + edited = cf.decide( + _context({"a": "EDITED", "b": "fp-b"}, baseline=baseline), _JUDGE + ) + self.assertEqual(cf.unreported_findings(edited, self._judged([])), []) + + renamed = cf.decide( + _context({"a": "fp-a", "c": "fp-c"}, baseline=baseline), _JUDGE + ) + self.assertEqual( + [ + entry["tool"] + for entry in cf.unreported_findings( + renamed, self._judged([]) + ) + ], + [cf.GENERAL], + ) + + +class RestoreTest(unittest.TestCase): + """Findings the second pass ruled still apply, put back into the report.""" + + def _judged(self, entries): + cf.mint_finding_ids(entries) + return {"findings_by_tool": entries, "waived": [], "summary": "s"} + + def test_a_restored_finding_returns_with_its_original_wording(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + judged = self._judged([]) + restored = cf.unreported_findings(decision, judged) + merged = cf.merge_feedback( + decision, judged, ["a", "b"], restored=restored + ) + by_tool = { + entry["tool"]: entry["findings"] + for entry in merged["findings_by_tool"] + } + self.assertEqual(by_tool["b"][0]["title"], "B finding") + + def test_a_tool_that_lost_every_finding_gets_its_entry_back(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + merged = cf.merge_feedback( + decision, + self._judged([]), + ["a", "b"], + restored=[ + {"tool": "b", "findings": [_finding("R2", "B finding")]} + ], + ) + self.assertIn( + "b", [entry["tool"] for entry in merged["findings_by_tool"]] + ) + + def test_a_restored_p0_leads_its_entry(self): + baseline = _baseline( + findings=[ + { + "tool": "b", + "findings": [_finding("R2", "blocker", severity="P0")], + } + ] + ) + decision = cf.decide( + _context({"a": "fp-a", "b": "EDITED"}, baseline=baseline), _JUDGE + ) + judged = self._judged( + [{"tool": "b", "findings": [_finding("R5", "minor")]}] + ) + merged = cf.merge_feedback( + decision, + judged, + ["a", "b"], + restored=cf.unreported_findings(decision, judged), + ) + by_tool = { + entry["tool"]: entry["findings"] + for entry in merged["findings_by_tool"] + } + self.assertEqual( + [finding["title"] for finding in by_tool["b"]], + ["blocker", "minor"], + ) + + def test_restoring_never_duplicates_a_finding_the_judge_repeated(self): + # Restored findings come off the baseline, so they always carry ids. + restored = [{"tool": "b", "findings": [_finding("R2", "B finding")]}] + cf.mint_finding_ids(restored) + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + merged = cf.merge_feedback( + decision, + self._judged( + [{"tool": "b", "findings": [_finding("R2", "B finding")]}] + ), + ["a", "b"], + restored=restored, + ) + by_tool = { + entry["tool"]: entry["findings"] + for entry in merged["findings_by_tool"] + } + self.assertEqual(len(by_tool["b"]), 1) + + def test_a_resolved_finding_stays_out(self): + """The second pass ruling "fixed" must actually drop the finding.""" + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + merged = cf.merge_feedback( + decision, self._judged([]), ["a", "b"], restored=[] + ) + self.assertEqual( + [entry["tool"] for entry in merged["findings_by_tool"]], ["a"] + ) + + +class FindingIdTest(unittest.TestCase): + + def test_the_same_finding_gets_the_same_id(self): + first = [{"tool": "a", "findings": [_finding("R1", "Title")]}] + second = [{"tool": "a", "findings": [_finding("R1", "Title")]}] + cf.mint_finding_ids(first) + cf.mint_finding_ids(second) + self.assertEqual( + first[0]["findings"][0]["finding_id"], + second[0]["findings"][0]["finding_id"], + ) + + def test_the_same_rule_twice_on_one_tool_stays_distinguishable(self): + entries = [ + { + "tool": "a", + "findings": [ + _finding("R1", "first", locator="param_x"), + _finding("R1", "second", locator="param_y"), + ], + } + ] + cf.mint_finding_ids(entries) + ids = {finding["finding_id"] for finding in entries[0]["findings"]} + self.assertEqual(len(ids), 2) + + def test_identical_findings_are_separated_by_an_ordinal(self): + entries = [ + { + "tool": "a", + "findings": [_finding("R1", "same"), _finding("R1", "same")], + } + ] + cf.mint_finding_ids(entries) + ids = [finding["finding_id"] for finding in entries[0]["findings"]] + self.assertEqual(len(set(ids)), 2) + self.assertTrue(ids[1].endswith("-1")) + + def test_an_existing_id_is_never_reminted(self): + entries = [{"tool": "a", "findings": [_finding("R1", "t")]}] + entries[0]["findings"][0]["finding_id"] = "carried-id" + cf.mint_finding_ids(entries) + self.assertEqual(entries[0]["findings"][0]["finding_id"], "carried-id") + + def test_the_title_is_normalized_before_hashing(self): + """Whitespace and case churn in the title must not mint a new id.""" + one = [{"tool": "a", "findings": [_finding("R1", "A Title")]}] + two = [{"tool": "a", "findings": [_finding("R1", "a title")]}] + cf.mint_finding_ids(one) + cf.mint_finding_ids(two) + self.assertEqual( + one[0]["findings"][0]["finding_id"], + two[0]["findings"][0]["finding_id"], + ) + + +class BuildProvenanceTest(unittest.TestCase): + + def _provenance(self, decision, judged=None, order=("a", "b")): + merged = cf.merge_feedback(decision, judged, list(order)) + entries = merged["findings_by_tool"] + cf.mint_finding_ids(entries) + return cf.build_provenance(decision, entries), entries + + def test_a_carried_run_reports_no_new_or_resolved_findings(self): + decision = cf.decide(_context(), _JUDGE) + provenance, _ = self._provenance(decision) + self.assertEqual(provenance["mode"], cf.MODE_CARRIED) + self.assertEqual(provenance["new_finding_ids"], []) + self.assertEqual(provenance["resolved_finding_ids"], []) + self.assertEqual( + provenance["finding_count"], provenance["previous_finding_count"] + ) + + def test_a_partial_run_labels_every_tool_by_where_it_came_from(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + judged = { + "findings_by_tool": [ + {"tool": "b", "findings": [_finding("R3", "fresh b")]} + ], + "waived": [], + "summary": "", + } + provenance, _ = self._provenance(decision, judged) + self.assertEqual( + provenance["tool_provenance"], {"a": "carried", "b": "rejudged"} + ) + self.assertEqual(provenance["rejudged_tools"], ["b"]) + + def test_a_full_judge_labels_nothing_as_carried(self): + """Every tool was re-judged, so a per-tool marker would be a lie.""" + decision = cf.decide(_context(), "new-judge-fp") + provenance, _ = self._provenance( + decision, + {"findings_by_tool": [], "waived": [], "summary": ""}, + ) + self.assertEqual(provenance["tool_provenance"], {}) + + def test_a_dropped_finding_is_reported_as_resolved(self): + decision = cf.decide(_context({"a": "fp-a", "b": "EDITED"}), _JUDGE) + judged = { + "findings_by_tool": [], # b's finding is gone + "waived": [], + "summary": "", + } + provenance, _ = self._provenance(decision, judged) + self.assertEqual(len(provenance["resolved_finding_ids"]), 1) + self.assertEqual(provenance["finding_count"], 1) + + def test_component_changes_carry_the_before_and_after_values(self): + decision = cf.decide_with_components( + _context(), + "new-fp", + {"judge_model": "new-model", "style_guide_sha": "s"}, + ) + provenance, _ = self._provenance( + decision, + {"findings_by_tool": [], "waived": [], "summary": ""}, + ) + self.assertEqual( + provenance["component_changes"]["judge_model"], + {"from": "m", "to": "new-model"}, + ) + + def test_a_non_scalar_component_change_is_reported_without_values(self): + decision = cf.decide_with_components( + _context(), + "new-fp", + { + "judge_model": "m", + "style_guide_sha": "s", + "exceptions": [{"rule_id": "r"}], + }, + ) + provenance, _ = self._provenance( + decision, + {"findings_by_tool": [], "waived": [], "summary": ""}, + ) + self.assertEqual(provenance["component_changes"]["exceptions"], {}) + + +if __name__ == "__main__": + unittest.main()