fix: preserve confirmed ingestion failure outcomes

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

Author: placerda

.github/scripts/check-quality.py

@@ -164,6 +164,9 @@ def policy_findings(root: Path, base: str, current: list[dict], protected: list[
     for field in ("runtime_roots", "contracts", "toolchain", "required_checks", "dynamic_imports", "review"):
         if policy[field] != base_policy[field]:
             findings.append(q.finding("protected-policy-change", ".quality/policy.json", reason=field))
+    for field in ("coverage_stage", "planned_expansion", "review"):
+        if scope.get(field) != base_scope.get(field):
+            findings.append(q.finding("protected-policy-change", ".quality/typing-scope.json", reason=field))
     old_surfaces = {module["id"]: module for module in base_policy["modules"]}
     for surface in policy["modules"]:
         old = old_surfaces.get(surface["id"])
@@ -196,7 +199,8 @@ def policy_findings(root: Path, base: str, current: list[dict], protected: list[
     return findings
 
 
-def lint(root: Path, config: Path, sources: dict) -> tuple[list[dict], dict]:
+def lint(root: Path, config: Path, sources: dict, entries: list[dict],
+         modules: dict[str, str], review_stage: str) -> tuple[list[dict], dict]:
     result = q.run_tool([sys.executable, "-m", "ruff", "check", "--config", str(config),
                          "--output-format", "json", "--no-cache", *sources], root)
     try:
@@ -205,13 +209,31 @@ def lint(root: Path, config: Path, sources: dict) -> tuple[list[dict], dict]:
         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")
+    approved_sites = []
+    path_ids = {path: key for key, path in modules.items()}
+    for site in q.handler_sites(sources):
+        site["module_id"] = path_ids[site["file"]]
+        matching = [record for record in q.matching_exception_records(site, entries)
+                    if record["review"].get("status") == "active"
+                    and q.valid_exception_metadata(record, review_stage)]
+        if len(matching) == 1:
+            approved_sites.append((site, matching[0]["id"]))
     findings = []
+    used_ids = set()
     for item in diagnostics:
         if not all(key in item for key in ("code", "filename", "location", "message")):
             raise q.QualityError("Unsupported Ruff diagnostic")
         file = Path(item["filename"]).resolve().relative_to(root).as_posix()
+        if item["code"] == "BLE001":
+            approved = [record_id for site, record_id in approved_sites
+                        if site["file"] == file
+                        and site["line"] <= item["location"]["row"] <= site["header_end_line"]]
+            if len(approved) == 1:
+                # The separate required exceptions job proves same-run behavior evidence.
+                used_ids.update(approved)
+                continue
         findings.append(q.finding(item["code"], file, item["location"]["row"], item["message"]))
-    return findings, {}
+    return findings, {"exception_ids_used": sorted(used_ids)}
 
 
 def typing(root: Path, config: Path, sources: dict, modules: dict, covered: set,
@@ -357,6 +379,9 @@ def main() -> int:
         if any(modules.get(module_id) not in sources for module_id in covered):
             raise q.QualityError("Covered module is missing; coverage cannot disappear")
         selected = q.REQUIRED_CHECKS[:-1] if args.check == "all" else (args.check,)
+        entries = current[3]["entries"] if bootstrap else [
+            entry for entry in current[3]["entries"] if entry in protected[3]["entries"]
+        ]
         with tempfile.TemporaryDirectory(prefix="ingestion-quality-") as temporary:
             config = Path(temporary) / "pyproject.toml"
             config.write_text(config_text, encoding="utf-8")
@@ -365,7 +390,9 @@ def main() -> int:
                     findings = policy_findings(root, base, current, protected, sources, declared_modules, base_modules, bootstrap)
                     details = {"bootstrap": bootstrap}
                 elif check == "lint":
-                    findings, details = lint(root, config, sources)
+                    findings, details = lint(root, config, sources, entries, modules,
+                                            protected[1].get("coverage_stage", "initial"))
+                    report["exception_ids_used"].extend(details["exception_ids_used"])
                 elif check == "typing":
                     baseline = current[2] if bootstrap else {"entries": [
                         entry for entry in current[2]["entries"] if entry in protected[2]["entries"]
@@ -376,11 +403,10 @@ def main() -> int:
                 else:
                     evidence = load_evidence(args.test_evidence, base, head)
                     # Candidate additions cannot grant exemptions to base-policy code.
-                    entries = current[3]["entries"] if bootstrap else [
-                        entry for entry in current[3]["entries"] if entry in protected[3]["entries"]
-                    ]
                     findings = q.exception_findings(sources, entries, evidence,
-                                                     {path: key for key, path in modules.items()})
+                                                     {path: key for key, path in modules.items()},
+                                                     accepted_ids=report["exception_ids_used"],
+                                                     review_stage=protected[1].get("coverage_stage", "initial"))
                     for record in policy["dynamic_imports"]:
                         if (record.get("review", {}).get("status") != "active"
                                 or not record.get("evidence_tests")
@@ -394,6 +420,7 @@ def main() -> int:
     except (q.QualityError, OSError, UnicodeError, KeyError, TypeError, subprocess.TimeoutExpired) as exc:
         report["findings"].append(q.finding("execution-error", reason=str(exc)))
         LOG.error("Quality execution incomplete: %s", exc)
+    report["exception_ids_used"] = sorted(set(report["exception_ids_used"]))
     report["duration_seconds"] = round(time.monotonic() - started, 3)
     report["artifact_integrity"] = q.digest(report)
     args.report.parent.mkdir(parents=True, exist_ok=True)

.github/scripts/quality_policy.py

@@ -23,6 +23,7 @@
 REQUIRED_CHECKS = ("lint", "typing", "architecture", "exceptions", "policy", "unit-tests")
 NON_RUNTIME = (".github/", "tests/", "scripts/", "samples/", "frontend/", ".artifacts/")
 DIAGNOSTIC_KEY = ("module_id", "symbol", "source_fingerprint", "rule", "message_fingerprint")
+EXCEPTION_KEY = ("module_id", "symbol", "handler_fingerprint", "caught_types")
 
 
 class QualityError(ValueError):
@@ -338,6 +339,7 @@ def handler_sites(sources: dict[str, str]) -> list[dict]:
                     symbol, _ = source_context(tree, handler.lineno)
                     sites.append({
                         "module_id": module_name(path), "file": path, "line": handler.lineno,
+                        "header_end_line": handler.type.end_lineno if handler.type else handler.lineno,
                         "symbol": symbol, "caught_types": names,
                         "handler_fingerprint": digest({
                             "try": ast.dump(node, include_attributes=False),
@@ -347,18 +349,32 @@ def handler_sites(sources: dict[str, str]) -> list[dict]:
     return sites
 
 
+def matching_exception_records(site: dict, records: list[dict]) -> list[dict]:
+    return [record for record in records if all(record.get(key) == site[key] for key in EXCEPTION_KEY)]
+
+
+def valid_exception_metadata(record: dict, review_stage: str) -> bool:
+    required = ("boundary", "reason", "failure_outcome", "diagnostic_path")
+    return (all(record.get(key) for key in required)
+            and bool(record["review"].get("reference"))
+            and record.get("review_by_stage") == review_stage)
+
+
 def exception_findings(sources: dict[str, str], records: list[dict], evidence: dict,
-                       module_ids: dict[str, str] | None = None) -> list[dict]:
+                       module_ids: dict[str, str] | None = None,
+                       accepted_ids: list[str] | None = None,
+                       review_stage: str = "initial") -> list[dict]:
     findings = []
     used = set()
+    proposed = set()
     identities = Counter()
     for site in handler_sites(sources):
         site["module_id"] = (module_ids or {}).get(site["file"], site["module_id"])
         identity = tuple(str(site[key]) for key in ("module_id", "symbol", "handler_fingerprint", "caught_types"))
         identities[identity] += 1
-        matches = [record for record in records if all(record.get(key) == site[key]
-                   for key in ("module_id", "symbol", "handler_fingerprint", "caught_types"))
-                   and record.get("review", {}).get("status") == "active"]
+        matching = matching_exception_records(site, records)
+        proposed.update(record["id"] for record in matching if record.get("review", {}).get("status") == "proposed")
+        matches = [record for record in matching if record.get("review", {}).get("status") == "active"]
         if len(matches) != 1 or identities[identity] > 1:
             findings.append(finding("unapproved-handler", site["file"], site["line"],
                                     "Requires one exact reviewed boundary record", **{
@@ -367,14 +383,17 @@ def exception_findings(sources: dict[str, str], records: list[dict], evidence: d
             continue
         record = matches[0]
         used.add(record["id"])
-        required = ("boundary", "reason", "failure_outcome", "diagnostic_path", "review_by_stage")
-        if not all(record.get(key) for key in required) or not record["review"].get("reference"):
+        errors_before = len(findings)
+        if not valid_exception_metadata(record, review_stage):
             findings.append(finding("invalid-exception", site["file"], site["line"], record["id"]))
         if not record.get("evidence_tests") or any(evidence.get(test) != "passed" for test in record["evidence_tests"]):
             findings.append(finding("exception-evidence", site["file"], site["line"], record["id"]))
+        if len(findings) == errors_before and accepted_ids is not None:
+            accepted_ids.append(record["id"])
     for record in records:
         if record["id"] not in used:
-            findings.append(finding("stale-exception", reason=record["id"]))
+            rule = "exception-review-pending" if record["id"] in proposed else "stale-exception"
+            findings.append(finding(rule, reason=record["id"]))
     return findings
 
 

.quality/exceptions.json

@@ -1,9 +1,75 @@
 {
   "schema_version": 1,
-  "entries": [],
+  "entries": [
+    {
+      "id": "audit-unexpected-sanitizer-failure",
+      "module_id": "telemetry.audit",
+      "symbol": "_emit",
+      "handler_fingerprint": "62bd9ada079c3a129b34b1d3ef5cf20037e5b840ca44b73bd9b46e0d365deb51",
+      "caught_types": ["Exception"],
+      "boundary": "Optional audit sanitization before export",
+      "reason": "A sanitizer defect must drop an unsafe audit event, not fail an already-running ingestion operation. This record covers only the unexpected sanitizer catch, not source/configuration failures.",
+      "failure_outcome": "contractual-best-effort-side-effect",
+      "diagnostic_path": "gptrag.audit_warning: dropped event type and exception class, without exception payload",
+      "evidence_tests": ["tests/test_audit_emitter.py::test_audit_side_effect_failure_is_nonblocking_and_does_not_log_payload[sanitizer]"],
+      "review": {"status": "proposed", "reference": "https://github.com/Azure/gpt-rag-ingestion/pull/296"},
+      "review_by_stage": "initial"
+    },
+    {
+      "id": "audit-exporter-failure",
+      "module_id": "telemetry.audit",
+      "symbol": "_emit",
+      "handler_fingerprint": "0dbfe922bf286e9bdd01421b045fa7add8315b201f11a3c421425d7c0b83307f",
+      "caught_types": ["Exception"],
+      "boundary": "Optional audit logger/exporter invocation",
+      "reason": "A failed logger/exporter must not reverse a confirmed Search outcome or turn a successful worker into a failure. Only the audit side effect is dropped; the primary result is unchanged.",
+      "failure_outcome": "contractual-best-effort-side-effect",
+      "diagnostic_path": "gptrag.audit_warning: failed export stage, event type and exception class only",
+      "evidence_tests": [
+        "tests/test_audit_emitter.py::test_broken_exporter_does_not_raise_and_does_not_fail_the_run",
+        "tests/test_audit_emitter.py::test_audit_side_effect_failure_is_nonblocking_and_does_not_log_payload[exporter]"
+      ],
+      "review": {"status": "proposed", "reference": "https://github.com/Azure/gpt-rag-ingestion/pull/296"},
+      "review_by_stage": "initial"
+    },
+    {
+      "id": "audit-run-failure-propagation",
+      "module_id": "telemetry.audit",
+      "symbol": "audit_run",
+      "handler_fingerprint": "978bd4cd83a4cd6092d208c117da9f6be5eda81fe689a1df1b39335df0f064e4",
+      "caught_types": ["Exception"],
+      "boundary": "Run context manager observing the primary worker failure",
+      "reason": "Record exactly one failed terminal event and re-raise the same primary exception. The context-variable cleanup must still run; this catch never converts an indexing/purge failure to success.",
+      "failure_outcome": "propagation",
+      "diagnostic_path": "Correlated ingestion.run.failed event; original exception propagates to the worker caller",
+      "evidence_tests": [
+        "tests/test_audit_emitter.py::test_unhandled_exception_emits_failed_and_reraises",
+        "tests/test_audit_emitter.py::test_cancelled_error_is_preserved_and_emits_cancelled_event"
+      ],
+      "review": {"status": "proposed", "reference": "https://github.com/Azure/gpt-rag-ingestion/pull/296"},
+      "review_by_stage": "initial"
+    },
+    {
+      "id": "document-audit-construction-failure",
+      "module_id": "telemetry.audit",
+      "symbol": "record_search_batch_result",
+      "handler_fingerprint": "6a1ad3870993d274922017fa77170b067eb97be1e74e885c99c3397854acee1f",
+      "caught_types": ["Exception"],
+      "boundary": "Optional document audit projection of an existing Search response",
+      "reason": "Malformed audit projection must not alter the caller's primary SDK result. The write adapter separately validates confirmation; this catch does not approve or synthesize a successful upload/delete.",
+      "failure_outcome": "contractual-best-effort-side-effect",
+      "diagnostic_path": "gptrag.audit_warning: document emission failed, with exception class but no source or exception payload",
+      "evidence_tests": [
+        "tests/test_audit_emitter.py::test_document_audit_never_raises_even_with_a_malformed_result",
+        "tests/test_audit_emitter.py::test_audit_side_effect_failure_is_nonblocking_and_does_not_log_payload[document]"
+      ],
+      "review": {"status": "proposed", "reference": "https://github.com/Azure/gpt-rag-ingestion/pull/296"},
+      "review_by_stage": "initial"
+    }
+  ],
   "review": {
     "status": "proposed",
     "reference": "https://github.com/Azure/GPT-RAG/pull/689",
-    "rationale": "No blanket exemptions. Unreviewed inherited handlers remain blocking findings; audit best-effort behavior is not rewritten to satisfy a gate."
+    "rationale": "Four exact audit-boundary proposals with failure evidence, not active approvals. All unreviewed handlers remain blocking; no blanket exemptions."
   }
 }

.quality/policy.json

@@ -44,6 +44,7 @@
         "Depends",
         "Dict",
         "HTTPException",
+        "JobLookupError",
         "List",
         "Literal",
         "ManagedIdentityCredential",
@@ -1250,6 +1251,7 @@
         "AzureCliCredential",
         "AzureError",
         "ChainedTokenCredential",
+        "Counter",
         "Dict",
         "List",
         "ManagedIdentityCredential",

AGENTS.md

@@ -114,8 +114,9 @@ and handler checks include flat roots, late/type-only imports and package
 facades. `main` compatibility re-exports are not another owner of mutable state.
 Use the jobs-owned test seam when patching scheduling.
 
-The initial exception ledger is empty and inherited handler/lint findings
-remain blocking. This is not an active required-merge claim. Protected policy
+Four audit exception records are proposed with exact failure evidence; none
+is active. Other inherited handler/lint findings remain blocking. This is not
+an active required-merge claim. Protected policy
 review, a clean reference PR and administrative rule activation are separate
 acceptance requirements; no agent may change GitHub settings to bypass them.
 

CHANGELOG.md

@@ -19,7 +19,23 @@
   main re-exports, manual reservation behavior and scheduler lifecycle. Lazy
   worker exports keep state-only imports from initializing Azure configuration
   before startup authentication. Remove two overwritten, unused SharePoint
-  wrappers; keep the registered audit-wrapped functions unchanged.
+  wrappers while retaining the shared registry/lock and startup ordering.
+
+### Fixed
+
+- **Primary failures no longer masquerade as confirmed work.** Search deletion
+  uses the SDK delete operation and counts matching confirmed outcomes rather
+  than submissions; NL2SQL purge propagates failed scans and partial deletion.
+  Missing, malformed or duplicate Search responses cannot emit positive audit
+  outcomes. Cron/manual failures propagate through the existing audit context,
+  while startup jobs retain independent failure isolation.
+
+- **Configuration and diagnostic failures remain explicit and safe.** Governance
+  reads no longer silently disable governance after provider failure.
+  Configuration apply returns an error on failed scheduling, and writes with a
+  failed local refresh return the existing partial-failure shape. Audit
+  sanitization/export/projection remains non-blocking, with payload-free warning
+  diagnostics and four exact, unapproved boundary proposals for review.
 
 ## [v2.7.3] - 2026-09-03
 

README.md

@@ -27,8 +27,11 @@ Python 3.12 contributor commands, the jobs-owned scheduler interface,
 incremental typing scope, import/error policy and recovery guidance are in
 [Python quality gates](docs/python-quality.md). The bootstrap reports inherited
 violations without blanket exemptions; required-check activation and remaining
-failure-boundary review are explicitly pending. Runtime contracts and operator
-startup/configuration remain unchanged.
+failure-boundary review are explicitly pending. Existing successful response
+shapes, schema bytes and authentication remain unchanged. Search and NL2SQL
+deletion counts now require confirmed results; worker and governance failures
+propagate, and configuration apply/refresh failures are reported explicitly
+rather than as success. Startup jobs retain independent failure isolation.
 
 ## Governance and audit events