fix: reject malformed diagnostics and contradictory quality receipts

Add regression-first Q6 coverage with real aggregate CLI negative cases. Validate diagnostic fields before baselining; preserve protected evaluation and zero active exception approvals.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: placerda

.github/scripts/check-quality.py

@@ -355,11 +355,13 @@ def lint(root: Path, config: Path, sources: dict, entries: list[dict],
                          "--output-format", "json", "--no-cache",
                          *[str(root / path) for path in sources]], config.parent)
     try:
-        diagnostics = json.loads(result.stdout)
+        diagnostics = json.loads(result.stdout, object_pairs_hook=q.unique_keys)
     except json.JSONDecodeError as exc:
         raise q.QualityError("Ruff did not return JSON") from exc
     if not isinstance(diagnostics, list) or (result.returncode == 1 and not diagnostics):
         raise q.QualityError("Incomplete Ruff execution")
+    if result.returncode == 0 and diagnostics:
+        raise q.QualityError("Ruff exit status contradicts its diagnostics")
     approved_sites = []
     path_ids = {path: key for key, path in modules.items()}
     for site in q.handler_sites(sources):
@@ -373,9 +375,17 @@ def lint(root: Path, config: Path, sources: dict, entries: list[dict],
     findings = []
     used_ids = set()
     for item in diagnostics:
-        if not all(key in item for key in ("code", "filename", "location", "message")):
+        if (not isinstance(item, dict)
+                or not all(isinstance(item.get(key), str) and item[key]
+                           for key in ("code", "filename", "message"))
+                or not isinstance(item.get("location"), dict)
+                or not all(type(item["location"].get(key)) is int and item["location"][key] >= 1
+                           for key in ("row", "column"))):
             raise q.QualityError("Unsupported Ruff diagnostic")
-        file = Path(item["filename"]).resolve().relative_to(root).as_posix()
+        path = Path(item["filename"]).resolve()
+        if not path.is_relative_to(root):
+            raise q.QualityError("Ruff emitted an out-of-repository source error")
+        file = path.relative_to(root).as_posix()
         if item["code"] == "BLE001":
             approved = [record_id for site, record_id in approved_sites
                         if site["file"] == file
@@ -402,15 +412,22 @@ def typing(root: Path, config: Path, sources: dict, modules: dict, covered: set,
         if not line.strip():
             continue
         try:
-            item = json.loads(line)
+            item = json.loads(line, object_pairs_hook=q.unique_keys)
         except json.JSONDecodeError as exc:
             raise q.QualityError("Mypy returned an unrecognized diagnostic") from exc
-        if not all(key in item for key in ("file", "line", "column", "message", "code", "severity")):
+        if (not isinstance(item, dict)
+                or not all(isinstance(item.get(key), str) and item[key]
+                           for key in ("file", "message", "severity"))
+                or type(item.get("line")) is not int or item["line"] < 1
+                or type(item.get("column")) is not int or item["column"] < 0
+                or "code" not in item):
             raise q.QualityError("Incomplete mypy diagnostic")
         if item["severity"] not in {"error", "note"}:
             raise q.QualityError("Unsupported mypy severity")
         if item["severity"] != "error":
             continue
+        if not isinstance(item["code"], str) or not item["code"]:
+            raise q.QualityError("Mypy error lacks a diagnostic code")
         path = Path(item["file"])
         path = (config.parent / path).resolve() if not path.is_absolute() else path.resolve()
         if not path.is_relative_to(root):

.github/scripts/quality_policy.py

@@ -758,9 +758,11 @@ def aggregate(jobs: dict, reports: dict, base_sha: str, head_sha: str) -> list[d
             continue
         report = reports.get(name, {})
         if not all((
-            report.get("schema_version") == 1, report.get("check_name") == name,
+            type(report.get("schema_version")) is int and report["schema_version"] == 1,
+            report.get("check_name") == name,
             report.get("status") == "passed", report.get("base_sha") == base_sha,
             report.get("head_sha") == head_sha,
+            report.get("findings") == [],
             isinstance(report.get("policy_sha"), str) and len(report["policy_sha"]) == 64,
         )):
             result.append(finding("invalid-report", reason=name))

CHANGELOG.md

@@ -23,6 +23,12 @@
 
 ### Fixed
 
+- **Malformed quality evidence cannot become a passing receipt.** Validate
+  Ruff/mypy diagnostic types, positions, duplicate keys and exit consistency;
+  reject aggregate reports whose passed status contradicts their findings.
+  Real aggregate CLI fixtures cover missing/skipped jobs and stale or mismatched
+  artifacts without treating incomplete execution as successful validation.
+
 - **Direct uploads require authoritative Search confirmation.** Exercise the
   actual `/ingest-documents` route with the pinned SDK response model, reject
   missing, duplicate and unrelated results, and retain the per-record response

docs/python-quality.md

@@ -188,6 +188,14 @@ declared area are retained. Record parsing rejects missing/unknown fields,
 invalid nested types and unsupported lifecycle values before running checks.
 Boolean or floating-point schema versions are not version 1. The exact
 requirements manifest must agree with the policy toolchain.
+Tool diagnostics are validated before they can participate in the baseline:
+Ruff and mypy require typed, nonempty diagnostic fields and valid source
+positions, and duplicate JSON keys or contradictory exit status are execution
+errors. The aggregate also rejects a report claiming `passed` while carrying
+findings. Existing-runner fixtures execute the aggregate CLI with missing or
+skipped jobs, missing reports, wrong base/head/policy/source/run/attempt,
+skipped or stale test evidence, and duplicate evidence keys, alongside the
+clean positive control.
 
 ## Broad handlers and remaining acceptance
 

tests/test_quality_policy.py

@@ -482,7 +482,8 @@ def timeout(*args, **kwargs):
 
 def report(name):
     return {"schema_version": 1, "check_name": name, "status": "passed",
-            "base_sha": "b" * 40, "head_sha": "c" * 40, "policy_sha": "d" * 64}
+            "base_sha": "b" * 40, "head_sha": "c" * 40, "policy_sha": "d" * 64,
+            "findings": []}
 
 
 def test_aggregate_requires_fixed_actual_results_and_fresh_reports():
@@ -499,6 +500,114 @@ def test_aggregate_requires_fixed_actual_results_and_fresh_reports():
     assert quality.aggregate({}, reports, "b" * 40, "c" * 40)
 
 
+@pytest.fixture
+def checker():
+    spec = importlib.util.spec_from_file_location("check_quality_under_test", SCRIPT.with_name("check-quality.py"))
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.mark.parametrize(("field", "value"), [
+    ("line", True), ("line", "1"), ("line", 0), ("column", None),
+    ("column", False), ("file", ""), ("message", []), ("code", None),
+    ("severity", "warning"),
+])
+def test_malformed_mypy_diagnostics_are_execution_errors(checker, monkeypatch, tmp_path, field, value):
+    diagnostic = {
+        "file": str(tmp_path / "a.py"), "line": 1, "column": 0,
+        "message": "bad return type", "code": "return-value", "severity": "error",
+    }
+    diagnostic[field] = value
+    monkeypatch.setattr(quality, "run_tool", lambda *a, **k: subprocess.CompletedProcess(
+        [], 1, json.dumps(diagnostic), "",
+    ))
+    with pytest.raises(quality.QualityError):
+        checker.typing(tmp_path, tmp_path / "pyproject.toml", {"a.py": "value = 1"},
+                       {"a": "a.py"}, {"a"}, {"entries": []})
+
+
+@pytest.mark.parametrize(("returncode", "diagnostics"), [(1, ""), (0, "error")])
+def test_mypy_status_must_match_diagnostics(checker, monkeypatch, tmp_path, returncode, diagnostics):
+    stdout = "" if not diagnostics else json.dumps({
+        "file": str(tmp_path / "a.py"), "line": 1, "column": 0,
+        "message": "bad return type", "code": "return-value", "severity": "error",
+    })
+    monkeypatch.setattr(quality, "run_tool", lambda *a, **k: subprocess.CompletedProcess(
+        [], returncode, stdout, "",
+    ))
+    with pytest.raises(quality.QualityError):
+        checker.typing(tmp_path, tmp_path / "pyproject.toml", {"a.py": "value = 1"},
+                       {"a": "a.py"}, {"a"}, {"entries": []})
+
+
+@pytest.mark.parametrize(("field", "value"), [
+    ("code", None), ("filename", ""), ("message", []),
+    ("location", {"row": True, "column": 1}), ("location", {"row": 0, "column": 1}),
+])
+def test_malformed_ruff_diagnostics_are_execution_errors(checker, monkeypatch, tmp_path, field, value):
+    diagnostic = {
+        "filename": str(tmp_path / "a.py"), "location": {"row": 1, "column": 1},
+        "message": "undefined name", "code": "F821",
+    }
+    diagnostic[field] = value
+    monkeypatch.setattr(quality, "run_tool", lambda *a, **k: subprocess.CompletedProcess(
+        [], 1, json.dumps([diagnostic]), "",
+    ))
+    with pytest.raises(quality.QualityError):
+        checker.lint(tmp_path, tmp_path / "pyproject.toml", {"a.py": "missing"}, [], {"a": "a.py"}, "initial")
+
+
+@pytest.mark.parametrize("mutation", [
+    "none", "missing-job", "skipped-job", "error-job", "missing-report",
+    "wrong-head", "wrong-base", "wrong-policy", "wrong-source", "wrong-run", "wrong-attempt",
+    "status-contradiction", "skipped-test", "stale-test", "wrong-test-attempt", "duplicate-test-key",
+])
+def test_aggregate_cli_requires_actual_consistent_current_receipts(tmp_path, mutation):
+    jobs = {name: {"result": "success"} for name in quality.REQUIRED_CHECKS}
+    for name in quality.REQUIRED_CHECKS[:-1]:
+        data = dict(report(name), repository="Azure/gpt-rag-ingestion", source_sha="a" * 64,
+                    run_id="123", run_attempt="1")
+        if name == "lint":
+            changes = {
+                "wrong-head": ("head_sha", "e" * 40), "wrong-base": ("base_sha", "e" * 40),
+                "wrong-policy": ("policy_sha", "e" * 64), "wrong-source": ("source_sha", "e" * 64),
+                "wrong-run": ("run_id", "124"), "wrong-attempt": ("run_attempt", "2"),
+                "status-contradiction": ("findings", [{"rule": "execution-error", "reason": "crash"}]),
+            }
+            if mutation in changes:
+                key, value = changes[mutation]
+                data[key] = value
+            if mutation == "missing-report":
+                continue
+        data["artifact_integrity"] = quality.digest(data)
+        (tmp_path / f"{name}.json").write_text(json.dumps(data), encoding="utf-8")
+    if mutation == "missing-job":
+        del jobs["unit-tests"]
+    elif mutation in {"skipped-job", "error-job"}:
+        jobs["unit-tests"]["result"] = mutation.removesuffix("-job")
+    evidence = {
+        "schema_version": 1, "base_sha": "b" * 40,
+        "head_sha": "e" * 40 if mutation == "stale-test" else "c" * 40,
+        "run_id": "123", "run_attempt": "2" if mutation == "wrong-test-attempt" else "1",
+        "junit_sha": "f" * 64,
+        "tests": {"tests/test_failure.py::test_error": "skipped" if mutation == "skipped-test" else "passed"},
+    }
+    evidence["artifact_integrity"] = quality.digest(evidence)
+    encoded = json.dumps(evidence)
+    if mutation == "duplicate-test-key":
+        encoded = encoded.replace('"tests": {', '"tests": {}, "tests": {')
+    (tmp_path / "test-evidence.json").write_text(encoded, encoding="utf-8")
+    result = subprocess.run(
+        [sys.executable, "-I", str(SCRIPT.with_name("quality-gate.py")),
+         "--reports", str(tmp_path), "--base-sha", "b" * 40, "--head-sha", "c" * 40],
+        env=dict(os.environ, NEEDS_JSON=json.dumps(jobs), GITHUB_RUN_ID="123", GITHUB_RUN_ATTEMPT="1"),
+        capture_output=True, text=True, timeout=30,
+    )
+    assert (result.returncode == 0) == (mutation == "none"), result.stderr
+
+
 @pytest.fixture
 def policy_repository(tmp_path):
     """A committed minimum policy and a separate trusted evaluator copy."""