From 49b4c0e09044538671b52f8efc9496e0fef2ac1d Mon Sep 17 00:00:00 2001 From: Charles Blackmon-Luca <20627856+charlesbluca@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:23:33 +0000 Subject: [PATCH 1/2] fix(service): normalize public ingestion errors --- .../common/error_normalization.py | 217 ++++++++++++++++++ .../common/schemas/responses.py | 11 + .../nemo_retriever/ingestor/graph_ingestor.py | 8 +- .../nemo_retriever/service/routers/ingest.py | 16 +- .../service/services/job_tracker.py | 6 + .../service/services/metrics.py | 31 +++ .../service/services/pipeline_pool.py | 20 +- .../tests/test_error_normalization.py | 97 ++++++++ 8 files changed, 397 insertions(+), 9 deletions(-) create mode 100644 nemo_retriever/src/nemo_retriever/common/error_normalization.py create mode 100644 nemo_retriever/tests/test_error_normalization.py diff --git a/nemo_retriever/src/nemo_retriever/common/error_normalization.py b/nemo_retriever/src/nemo_retriever/common/error_normalization.py new file mode 100644 index 0000000000..80ae35971b --- /dev/null +++ b/nemo_retriever/src/nemo_retriever/common/error_normalization.py @@ -0,0 +1,217 @@ +"""Safe, consistent error rendering for public service surfaces.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +PUBLIC_ERROR_MESSAGE_LIMIT = 2048 +_LIST_LIMIT = 5 +_SENSITIVE_VALUE = re.compile( + r"(?i)\b(?:api[_-]?key|token|authorization|password|secret|credential)\s*[:=]\s*(?:bearer\s+)?[^\s,;]+" +) +_BEARER_VALUE = re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]+") +_URL = re.compile(r"https?://[^\s\]\[},;]+") + + +@dataclass(frozen=True) +class NormalizedError: + """Bounded, redacted error data safe for APIs, logs, and metrics.""" + + type: str + stage: str | None + endpoint: str | None + message: str + + @property + def summary(self) -> str: + context = [] + if self.stage: + context.append(f"stage={self.stage}") + if self.endpoint: + context.append(f"endpoint={self.endpoint}") + prefix = self.type + if context: + prefix += " [" + " ".join(context) + "]" + return f"{prefix}: {self.message}" + + def as_dict(self) -> dict[str, str | None]: + return {"type": self.type, "stage": self.stage, "endpoint": self.endpoint, "message": self.message} + + +def normalize_error(value: Any, *, limit: int = PUBLIC_ERROR_MESSAGE_LIMIT) -> NormalizedError: + """Normalize an exception or error-shaped value without exposing secrets. + + Strings are intentionally handled before all iterable types: a string is one + message, never a sequence of individual error items. + """ + normalized = _normalize(value, seen=set()) + return NormalizedError( + type=_safe_text(normalized.type, limit=128) or "Error", + stage=_safe_text(normalized.stage, limit=128), + endpoint=_safe_endpoint(normalized.endpoint), + message=_safe_text(normalized.message, limit=limit) or "Unknown error", + ) + + +def _normalize(value: Any, *, seen: set[int]) -> NormalizedError: + if isinstance(value, str): + return NormalizedError("Error", None, None, value) + if value is None: + return NormalizedError("Error", None, None, "Unknown error") + if isinstance(value, (list, tuple)): + return _normalize_sequence(value, seen=seen) + if isinstance(value, dict): + return _normalize_mapping(value, seen=seen) + if isinstance(value, BaseException): + return _normalize_exception(value, seen=seen) + return NormalizedError(type(value).__name__, None, None, str(value)) + + +def _normalize_sequence(values: list[Any] | tuple[Any, ...], *, seen: set[int]) -> NormalizedError: + items = [_normalize(item, seen=seen) for item in values[:_LIST_LIMIT]] + if not items: + return NormalizedError("Error", None, None, "Unknown error") + first = items[0] + message = "; ".join(item.summary for item in items) + if len(values) > _LIST_LIMIT: + message += f" ({len(values) - _LIST_LIMIT} more)" + return NormalizedError(first.type if len(items) == 1 else "MultipleErrors", first.stage, first.endpoint, message) + + +def _normalize_mapping(value: dict[Any, Any], *, seen: set[int]) -> NormalizedError: + error_type = value.get("type") or value.get("error_type") or "Error" + stage = value.get("stage") + endpoint = value.get("endpoint") or value.get("url") or value.get("invoke_url") + message = value.get("message") or value.get("detail") + status = _status_code(value) + nested = value.get("error") or value.get("exception") or value.get("errors") + nested_error = _normalize(nested, seen=seen) if nested is not None and nested is not value else None + if message is None and nested_error is not None: + message = nested_error.message + if stage is None and nested_error is not None: + stage = nested_error.stage + if endpoint is None and nested_error is not None: + endpoint = nested_error.endpoint + if error_type == "Error" and nested_error is not None: + error_type = nested_error.type + if status is not None: + prefix = f"HTTP {status}" + message = f"{prefix}: {message}" if message else prefix + if message is None: + message = str(error_type) + return NormalizedError(str(error_type), _as_text(stage), _as_text(endpoint), _as_text(message) or "Unknown error") + + +def _normalize_exception(exc: BaseException, *, seen: set[int]) -> NormalizedError: + ident = id(exc) + if ident in seen: + return NormalizedError(type(exc).__name__, None, None, "Exception chain cycle") + seen.add(ident) + try: + if type(exc).__name__ == "GraphIngestionError" and hasattr(exc, "records"): + return _normalize_graph_ingestion_error(exc, seen=seen) + response = getattr(exc, "response", None) + status = getattr(response, "status_code", None) if response is not None else None + endpoint = getattr(response, "url", None) if response is not None else None + message = str(exc) or type(exc).__name__ + if response is not None: + try: + response_body = response.json() + except (AttributeError, TypeError, ValueError): + response_body = None + if isinstance(response_body, dict): + response_message = _normalize_mapping(response_body, seen=seen).message + if response_message and response_message != "Error": + message = response_message + if status is not None: + message = f"HTTP {status}: {message}" + cause = exc.__cause__ or (None if exc.__suppress_context__ else exc.__context__) + if cause is not None: + causal = _normalize(cause, seen=seen) + if not endpoint: + endpoint = causal.endpoint + message = f"{message}; caused by {causal.summary}" + return NormalizedError(type(exc).__name__, None, _as_text(endpoint), message) + finally: + seen.discard(ident) + + +def _normalize_graph_ingestion_error(exc: BaseException, *, seen: set[int]) -> NormalizedError: + records = getattr(exc, "records", []) + # Old callers may have passed a string despite the historical list annotation. + if isinstance(records, str): + records = [records] + if not isinstance(records, (list, tuple)): + records = [records] + first = records[0] if records else str(exc) + record = first if isinstance(first, dict) else {"error": first} + nested = _normalize(record.get("error"), seen=seen) + diagnostics = getattr(exc, "stage_diagnostics", {}) or {} + column = record.get("column") + diagnostic = diagnostics.get(column) if isinstance(diagnostics, dict) and isinstance(column, str) else None + role = getattr(diagnostic, "role", None) + display_name = getattr(diagnostic, "display_name", None) + stage = _as_text(role).upper() if _as_text(role) else _as_text(display_name) + endpoint = getattr(diagnostic, "invoke_url", None) or nested.endpoint + return NormalizedError("GraphIngestionError", stage or nested.stage, _as_text(endpoint), nested.message) + + +def _status_code(value: dict[Any, Any]) -> int | None: + for key in ("status_code", "http_status", "status", "code"): + candidate = value.get(key) + if isinstance(candidate, int) and not isinstance(candidate, bool) and 100 <= candidate < 1000: + return candidate + if isinstance(candidate, str) and candidate.strip().isdigit(): + parsed = int(candidate.strip()) + if 100 <= parsed < 1000: + return parsed + return None + + +def _safe_endpoint(value: str | None) -> str | None: + text = " ".join(str(value).split()) if value is not None else "" + if len(text) > 1024: + text = text[:1024].rstrip() + "..." + if not text: + return None + try: + parsed = urlsplit(text) + if parsed.scheme and parsed.netloc: + host = parsed.hostname or "" + if parsed.port: + host = f"{host}:{parsed.port}" + return urlunsplit((parsed.scheme, host, parsed.path, "", "")) + except ValueError: + pass + return _BEARER_VALUE.sub("Bearer ", _SENSITIVE_VALUE.sub("", text)) + + +def _safe_text(value: Any, *, limit: int) -> str | None: + if value is None: + return None + text = _redact_text(str(value)) + text = "".join(ch if ch.isprintable() else " " for ch in text).strip() + text = " ".join(text.split()) + if not text: + return None + if len(text) > limit: + return text[:limit].rstrip() + "..." + return text + + +def _redact_text(text: str) -> str: + text = _SENSITIVE_VALUE.sub("", text) + text = _BEARER_VALUE.sub("Bearer ", text) + + def replace_url(match: re.Match[str]) -> str: + return _safe_endpoint(match.group(0)) or "" + + # `_safe_endpoint` only calls `_redact_text` for non-URLs, preventing recursion. + return _URL.sub(replace_url, text) + + +def _as_text(value: Any) -> str | None: + return None if value is None else str(value) diff --git a/nemo_retriever/src/nemo_retriever/common/schemas/responses.py b/nemo_retriever/src/nemo_retriever/common/schemas/responses.py index c5988e8b7a..cc879ff230 100644 --- a/nemo_retriever/src/nemo_retriever/common/schemas/responses.py +++ b/nemo_retriever/src/nemo_retriever/common/schemas/responses.py @@ -12,6 +12,15 @@ from nemo_retriever.common.schemas.collections import IngestOperation +class ErrorDetails(RichModel): + """Redacted structured diagnostics accompanying a compatible error string.""" + + type: str + stage: str | None = None + endpoint: str | None = None + message: str + + class IngestAccepted(RichModel): """Response for the general ``POST /v1/ingest`` endpoint.""" @@ -58,6 +67,7 @@ class JobStatusResponse(RichModel): result_rows: int | None = None result_data: list[dict[str, Any]] | None = None error: str | None = None + error_details: ErrorDetails | None = None class SidecarUploadResponse(RichModel): @@ -130,6 +140,7 @@ class DocumentStatusResponse(RichModel): result_rows: int | None = None result_data: list[dict[str, Any]] | None = None error: str | None = None + error_details: ErrorDetails | None = None collection_name: str | None = None content_sha256: str | None = None diff --git a/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py b/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py index e25286b99f..ef9c6a3c41 100644 --- a/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py +++ b/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py @@ -134,12 +134,14 @@ class GraphIngestionError(RuntimeError): def __init__( self, - records: list[Any], + records: Any, stage_diagnostics: dict[str, _StageDiagnostic] | None = None, ) -> None: - self.records = records + self.records = ( + [records] if isinstance(records, str) else list(records) if isinstance(records, (list, tuple)) else [records] + ) self.stage_diagnostics = dict(stage_diagnostics) if stage_diagnostics else {} - super().__init__(_format_stage_error_message(records, self.stage_diagnostics)) + super().__init__(_format_stage_error_message(self.records, self.stage_diagnostics)) def _normalize_stage_error_record(record: Any) -> dict[str, Any] | None: diff --git a/nemo_retriever/src/nemo_retriever/service/routers/ingest.py b/nemo_retriever/src/nemo_retriever/service/routers/ingest.py index fa5130c415..bf403219d9 100644 --- a/nemo_retriever/src/nemo_retriever/service/routers/ingest.py +++ b/nemo_retriever/src/nemo_retriever/service/routers/ingest.py @@ -32,6 +32,7 @@ from pydantic import BaseModel, Field, model_validator from starlette.responses import StreamingResponse +from nemo_retriever.common.error_normalization import normalize_error from nemo_retriever.common.schemas.collections import IngestOperation from nemo_retriever.common.schemas.pipeline_spec import PipelineSpec from nemo_retriever.common.schemas.requests import IngestRequest, JobCreateRequest @@ -972,6 +973,7 @@ def _document_to_response(rec, *, result_data=None) -> DocumentStatusResponse: result_rows=rec.result_rows, result_data=result_data, error=rec.error, + error_details=rec.error_details, collection_name=rec.collection_name, content_sha256=rec.content_sha256, ) @@ -1426,6 +1428,7 @@ async def _status_response(request: Request, item_id: str) -> JSONResponse: result_rows=rec.result_rows, result_data=result_data, error=rec.error, + error_details=rec.error_details, ).model_dump() if is_terminal: @@ -1967,12 +1970,18 @@ async def job_callback(request: Request) -> JSONResponse: job_id_for_log = pre_rec.job_id if pre_rec is not None else None status = body.get("status", "completed") + error_for_log: str | None = None if status == "failed": + normalized_error = normalize_error(body.get("error_details") or body.get("error", "unknown error")) + error_for_log = normalized_error.summary outcome = tracker.mark_failed( item_id, - body.get("error", "unknown error"), + normalized_error.summary, + error_details=normalized_error.as_dict(), elapsed_s=body.get("elapsed_s"), ) + if outcome == MarkOutcome.TRANSITIONED and (m := get_metrics()) is not None: + m.record_terminal_failure(item_id, normalized_error.summary, normalized_error.as_dict()) else: result_rows = body.get("result_rows", 0) if pre_rec is None and result_rows and body.get("result_worker_ip"): @@ -2021,13 +2030,14 @@ async def job_callback(request: Request) -> JSONResponse: log_fn = logger.warning if outcome == MarkOutcome.UNKNOWN_DOCUMENT else logger.info log_fn( - "Gateway callback: id=%s job_id=%s status=%s outcome=%s rows=%s subscribers=%d", + "Gateway callback: id=%s job_id=%s status=%s outcome=%s rows=%s subscribers=%d error=%s", item_id, job_id_for_log or "?", status, outcome.value, body.get("result_rows", 0), sub_count, + error_for_log, ) if broker is not None and lease_record is not None: try: @@ -2232,6 +2242,7 @@ def _snapshot_terminal_jobs(tracker: Any, *, job_id: str | None = None) -> list[ "result_rows": rec.result_rows, "elapsed_s": rec.elapsed_s, "error": rec.error, + "error_details": rec.error_details, } for rec in recs if rec.status in terminal @@ -2283,6 +2294,7 @@ async def ingest_status_batch(request: Request) -> JSONResponse: "result_rows": rec.result_rows, "elapsed_s": rec.elapsed_s, "error": rec.error, + "error_details": rec.error_details, } terminal_count = sum( diff --git a/nemo_retriever/src/nemo_retriever/service/services/job_tracker.py b/nemo_retriever/src/nemo_retriever/service/services/job_tracker.py index a1aed49c62..6fd7690f4b 100644 --- a/nemo_retriever/src/nemo_retriever/service/services/job_tracker.py +++ b/nemo_retriever/src/nemo_retriever/service/services/job_tracker.py @@ -147,6 +147,7 @@ class DocumentRecord(RichModel): result_rows: int | None = None result_data: list[dict[str, Any]] | None = None error: str | None = None + error_details: dict[str, str | None] | None = None filename: str | None = Field(default=None, description="Original upload filename surfaced in the dashboard") collection_name: str | None = None content_sha256: str | None = None @@ -564,6 +565,7 @@ def mark_failed( document_id: str, error: str, *, + error_details: dict[str, str | None] | None = None, elapsed_s: float | None = None, ) -> MarkOutcome: """Transition a document to ``failed``; maybe finalize the job. @@ -574,6 +576,7 @@ def mark_failed( document_id, new_status=DocumentStatus.FAILED, error=error, + error_details=error_details, elapsed_s=elapsed_s, ) @@ -585,6 +588,7 @@ def _mark_terminal( result_rows: int = 0, result_data: list[dict[str, Any]] | None = None, error: str | None = None, + error_details: dict[str, str | None] | None = None, elapsed_s: float | None = None, ) -> MarkOutcome: # Phase 1: under lock, mutate state and gather snapshots. @@ -617,6 +621,7 @@ def _mark_terminal( retain_results = bool(agg_for_retain.retain_results) if agg_for_retain is not None else False rec.result_data = copy.deepcopy(result_data) if retain_results else None rec.error = error + rec.error_details = copy.deepcopy(error_details) if error_details is not None else None if elapsed_s is not None: rec.elapsed_s = elapsed_s else: @@ -804,6 +809,7 @@ def _publish_document_event(self, rec: DocumentRecord) -> None: "result_rows": rec.result_rows, "elapsed_s": rec.elapsed_s, "error": rec.error, + "error_details": rec.error_details, "filename": rec.filename, } self._event_bus.publish_sync(event, job_id=rec.job_id) diff --git a/nemo_retriever/src/nemo_retriever/service/services/metrics.py b/nemo_retriever/src/nemo_retriever/service/services/metrics.py index 6e7d76a67a..662601624c 100644 --- a/nemo_retriever/src/nemo_retriever/service/services/metrics.py +++ b/nemo_retriever/src/nemo_retriever/service/services/metrics.py @@ -36,6 +36,7 @@ from pydantic import ConfigDict, Field from nemo_retriever.common.schemas.base import RichModel +from nemo_retriever.common.schemas.responses import ErrorDetails logger = logging.getLogger(__name__) @@ -63,6 +64,7 @@ class PageMetric(RichModel): completed_at: str | None = None processing_duration_s: float | None = None error: str | None = None + error_details: ErrorDetails | None = None class DocumentMetric(RichModel): @@ -82,6 +84,7 @@ class DocumentMetric(RichModel): completed_at: str | None = None processing_duration_s: float | None = None error: str | None = None + error_details: ErrorDetails | None = None class JobMetric(RichModel): @@ -298,6 +301,34 @@ def record_page_completed(self, page_id: str) -> None: ) break + def record_terminal_failure(self, item_id: str, error: str, error_details: dict[str, str | None]) -> None: + """Record a normalized failure for whichever metric record owns *item_id*.""" + now = datetime.now(timezone.utc).isoformat() + details = ErrorDetails(**error_details) + with self._lock: + changed = False + if item_id in self._documents: + doc = self._documents[item_id] + self._documents[item_id] = doc.model_copy( + update={"status": "failed", "completed_at": now, "error": error, "error_details": details} + ) + if doc.job_id and doc.job_id in self._jobs: + job = self._jobs[doc.job_id] + self._jobs[doc.job_id] = job.model_copy( + update={"documents_failed": job.documents_failed + 1} + ) + changed = True + for i, page in enumerate(self._pages): + if page.page_id == item_id: + self._pages[i] = page.model_copy( + update={"status": "failed", "completed_at": now, "error": error, "error_details": details} + ) + changed = True + break + if changed: + self._total_errors += 1 + self._errors_by_type[details.type] = self._errors_by_type.get(details.type, 0) + 1 + # ── single-record lookups ──────────────────────────────────────── def get_job(self, job_id: str) -> JobMetric | None: diff --git a/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py b/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py index 6d539d975a..306c0e2142 100644 --- a/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py +++ b/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py @@ -160,6 +160,7 @@ async def _fire_gateway_callback( *, result_rows: int = 0, error: str | None = None, + error_details: dict[str, str | None] | None = None, result_worker_ip: str | None = None, callback_headers: Mapping[str, str] | None = None, retry_after_cap_s: float = _CALLBACK_RETRY_DELAYS_S[-1], @@ -181,6 +182,8 @@ async def _fire_gateway_callback( } if error: payload["error"] = error + if error_details: + payload["error_details"] = error_details if result_worker_ip: payload["result_worker_ip"] = result_worker_ip if lease_id is not None: @@ -365,6 +368,7 @@ async def _schedule_gateway_callback_retry( status: str, result_rows: int = 0, error: str | None = None, + error_details: dict[str, str | None] | None = None, result_worker_ip: str | None = None, callback_headers: Mapping[str, str] | None = None, retain_results: bool = False, @@ -389,6 +393,7 @@ async def _schedule_gateway_callback_retry( status=status, result_rows=result_rows, error=error, + error_details=error_details, result_worker_ip=result_worker_ip, callback_headers=callback_headers, retain_results=retain_results, @@ -431,6 +436,7 @@ async def _retry_gateway_callback_until_expired( status: str, result_rows: int, error: str | None, + error_details: dict[str, str | None] | None, result_worker_ip: str | None, callback_headers: Mapping[str, str] | None, retain_results: bool, @@ -454,6 +460,7 @@ async def _retry_gateway_callback_until_expired( status, result_rows=result_rows, error=error, + error_details=error_details, result_worker_ip=result_worker_ip, callback_headers=callback_headers, retry_after_cap_s=_CALLBACK_DEFERRED_MAX_DELAY_S, @@ -616,13 +623,16 @@ async def _heartbeat_loop(claimed_item: WorkItem = item) -> None: self._processed += 1 except Exception as exc: outcome = "failed" + normalized_error = normalize_error(exc) + error = normalized_error.summary + error_details = normalized_error.as_dict() if item.callback_url: - error = f"{type(exc).__name__}: {exc}" callback_outcome = await _fire_gateway_callback( item.callback_url, item.id, "failed", error=error, + error_details=error_details, callback_headers=item.callback_headers, lease_id=item.lease_id, lease_generation=item.lease_generation, @@ -633,18 +643,20 @@ async def _heartbeat_loop(claimed_item: WorkItem = item) -> None: item_id=item.id, status="failed", error=error, + error_details=error_details, callback_headers=item.callback_headers, work_item=item, ) else: tracker = get_job_tracker() if tracker is not None: - tracker.mark_failed(item.id, f"{type(exc).__name__}: {exc}") - logger.exception( - "Pool '%s' worker %d failed on item %s", + tracker.mark_failed(item.id, error, error_details=error_details) + logger.error( + "Pool '%s' worker %d failed on item %s: %s", self._name, worker_id, item.id, + error, ) finally: # Always observe; cheaper to keep latency series complete diff --git a/nemo_retriever/tests/test_error_normalization.py b/nemo_retriever/tests/test_error_normalization.py new file mode 100644 index 0000000000..ba7f16ecb7 --- /dev/null +++ b/nemo_retriever/tests/test_error_normalization.py @@ -0,0 +1,97 @@ +"""Regression coverage for safe public error normalization.""" + +from __future__ import annotations + +from fastapi.testclient import TestClient +import requests + +from nemo_retriever.common.error_normalization import normalize_error +from nemo_retriever.ingestor.graph_ingestor import GraphIngestionError, _StageDiagnostic +from nemo_retriever.service.app import create_app +from nemo_retriever.service.auth import AuthConfig +from nemo_retriever.service.config import ServiceConfig +from nemo_retriever.service.services.job_tracker import get_job_tracker +from nemo_retriever.service.services.metrics import get_metrics + + +def _http_422() -> requests.HTTPError: + response = requests.Response() + response.status_code = 422 + response.url = "http://ocr.local/nemotron-ocr-v2?token=never-show-this" + response._content = b'{"detail":"invalid image data"}' + return requests.HTTPError("422 Client Error", response=response) + + +def _ocr_error() -> GraphIngestionError: + return GraphIngestionError( + [{ + "row_index": 0, + "column": "ocr", + "path": "error", + "error": _http_422(), + }], + stage_diagnostics={"ocr": _StageDiagnostic( + column="ocr", display_name="OCR NIM", + invoke_url="http://ocr.local/nemotron-ocr-v2?token=never-show-this", role="ocr", + )}, + ) + + +def test_string_is_atomic_and_lists_join_only_at_list_level() -> None: + assert normalize_error("GraphIngestionError").message == "GraphIngestionError" + assert ";" not in normalize_error("GraphIngestionError").summary + assert normalize_error(["first failure", "second failure"]).message == "Error: first failure; Error: second failure" + + +def test_generic_exception_is_normalized() -> None: + assert normalize_error(Exception("plain failure")).as_dict() == { + "type": "Exception", "stage": None, "endpoint": None, "message": "plain failure" + } + + +def test_graph_ingestion_string_constructor_never_renders_characters() -> None: + rendered = str(GraphIngestionError("GraphIngestionError")) + assert "G;r;a;p;h" not in rendered + assert "GraphIngestionError" in rendered + + +def test_nested_ocr_error_preserves_safe_structured_fields() -> None: + assert normalize_error(_ocr_error()).as_dict() == { + "type": "GraphIngestionError", "stage": "OCR", + "endpoint": "http://ocr.local/nemotron-ocr-v2", "message": "HTTP 422: invalid image data", + } + + +def test_secrets_and_request_payloads_are_not_rendered() -> None: + normalized = normalize_error({ + "type": "HTTPError", "endpoint": "https://user:pass@example.test/ocr?api_key=super-secret", + "message": "Authorization: Bearer top-secret-token invalid image data", + "request_body": {"image": "base64-private-payload"}, + }) + assert "top-secret-token" not in normalized.summary + assert "super-secret" not in normalized.summary + assert "base64-private-payload" not in normalized.summary + assert normalized.endpoint == "https://example.test/ocr" + + +def test_status_and_metrics_share_normalized_failure() -> None: + config = ServiceConfig(mode="standalone", auth=AuthConfig(allow_unscoped_dev=True)) + with TestClient(create_app(config)) as client: + tracker = get_job_tracker() + metrics = get_metrics() + assert tracker is not None and metrics is not None + tracker.register_job("normalization-job", expected_documents=1) + tracker.register_document("normalization-doc", job_id="normalization-job") + tracker.mark_processing("normalization-doc") + metrics.record_document_accepted(document_id="normalization-doc", job_id="normalization-job") + response = client.post("/v1/internal/job-callback", json={ + "id": "normalization-doc", "status": "failed", "error_details": normalize_error(_ocr_error()).as_dict(), + }) + assert response.status_code == 200, response.text + status = client.get("/v1/ingest/status/normalization-doc") + metric = client.get("/v1/ingest/metrics/document/normalization-doc") + assert status.status_code == 200 + assert metric.status_code == 200 + assert status.json()["error_details"] == metric.json()["error_details"] + assert status.json()["error"] == metric.json()["error"] + assert "G;r;a;p;h" not in status.json()["error"] From ba51b2b2b2306d727d252a75d960ff8a128862bb Mon Sep 17 00:00:00 2001 From: Charles Blackmon-Luca <20627856+charlesbluca@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:27:34 +0000 Subject: [PATCH 2/2] style: satisfy pre-commit checks --- .../nemo_retriever/ingestor/graph_ingestor.py | 4 +- .../service/services/metrics.py | 4 +- .../service/services/pipeline_pool.py | 1 + .../tests/test_error_normalization.py | 61 ++++++++++++------- 4 files changed, 45 insertions(+), 25 deletions(-) diff --git a/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py b/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py index ef9c6a3c41..40e7ae6785 100644 --- a/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py +++ b/nemo_retriever/src/nemo_retriever/ingestor/graph_ingestor.py @@ -138,7 +138,9 @@ def __init__( stage_diagnostics: dict[str, _StageDiagnostic] | None = None, ) -> None: self.records = ( - [records] if isinstance(records, str) else list(records) if isinstance(records, (list, tuple)) else [records] + [records] + if isinstance(records, str) + else list(records) if isinstance(records, (list, tuple)) else [records] ) self.stage_diagnostics = dict(stage_diagnostics) if stage_diagnostics else {} super().__init__(_format_stage_error_message(self.records, self.stage_diagnostics)) diff --git a/nemo_retriever/src/nemo_retriever/service/services/metrics.py b/nemo_retriever/src/nemo_retriever/service/services/metrics.py index 662601624c..15dea9a480 100644 --- a/nemo_retriever/src/nemo_retriever/service/services/metrics.py +++ b/nemo_retriever/src/nemo_retriever/service/services/metrics.py @@ -314,9 +314,7 @@ def record_terminal_failure(self, item_id: str, error: str, error_details: dict[ ) if doc.job_id and doc.job_id in self._jobs: job = self._jobs[doc.job_id] - self._jobs[doc.job_id] = job.model_copy( - update={"documents_failed": job.documents_failed + 1} - ) + self._jobs[doc.job_id] = job.model_copy(update={"documents_failed": job.documents_failed + 1}) changed = True for i, page in enumerate(self._pages): if page.page_id == item_id: diff --git a/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py b/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py index 306c0e2142..72dd4413d4 100644 --- a/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py +++ b/nemo_retriever/src/nemo_retriever/service/services/pipeline_pool.py @@ -33,6 +33,7 @@ from pydantic import ConfigDict, Field +from nemo_retriever.common.error_normalization import normalize_error from nemo_retriever.service.config import AuthConfig, PipelinePoolConfig, WorkQueueConfig from nemo_retriever.common.schemas.base import RichModel from nemo_retriever.common.schemas.collections import IngestOperation diff --git a/nemo_retriever/tests/test_error_normalization.py b/nemo_retriever/tests/test_error_normalization.py index ba7f16ecb7..9f97c4a655 100644 --- a/nemo_retriever/tests/test_error_normalization.py +++ b/nemo_retriever/tests/test_error_normalization.py @@ -24,16 +24,22 @@ def _http_422() -> requests.HTTPError: def _ocr_error() -> GraphIngestionError: return GraphIngestionError( - [{ - "row_index": 0, - "column": "ocr", - "path": "error", - "error": _http_422(), - }], - stage_diagnostics={"ocr": _StageDiagnostic( - column="ocr", display_name="OCR NIM", - invoke_url="http://ocr.local/nemotron-ocr-v2?token=never-show-this", role="ocr", - )}, + [ + { + "row_index": 0, + "column": "ocr", + "path": "error", + "error": _http_422(), + } + ], + stage_diagnostics={ + "ocr": _StageDiagnostic( + column="ocr", + display_name="OCR NIM", + invoke_url="http://ocr.local/nemotron-ocr-v2?token=never-show-this", + role="ocr", + ) + }, ) @@ -45,7 +51,10 @@ def test_string_is_atomic_and_lists_join_only_at_list_level() -> None: def test_generic_exception_is_normalized() -> None: assert normalize_error(Exception("plain failure")).as_dict() == { - "type": "Exception", "stage": None, "endpoint": None, "message": "plain failure" + "type": "Exception", + "stage": None, + "endpoint": None, + "message": "plain failure", } @@ -57,17 +66,22 @@ def test_graph_ingestion_string_constructor_never_renders_characters() -> None: def test_nested_ocr_error_preserves_safe_structured_fields() -> None: assert normalize_error(_ocr_error()).as_dict() == { - "type": "GraphIngestionError", "stage": "OCR", - "endpoint": "http://ocr.local/nemotron-ocr-v2", "message": "HTTP 422: invalid image data", + "type": "GraphIngestionError", + "stage": "OCR", + "endpoint": "http://ocr.local/nemotron-ocr-v2", + "message": "HTTP 422: invalid image data", } def test_secrets_and_request_payloads_are_not_rendered() -> None: - normalized = normalize_error({ - "type": "HTTPError", "endpoint": "https://user:pass@example.test/ocr?api_key=super-secret", - "message": "Authorization: Bearer top-secret-token invalid image data", - "request_body": {"image": "base64-private-payload"}, - }) + normalized = normalize_error( + { + "type": "HTTPError", + "endpoint": "https://user:pass@example.test/ocr?api_key=super-secret", + "message": "Authorization: Bearer top-secret-token invalid image data", + "request_body": {"image": "base64-private-payload"}, + } + ) assert "top-secret-token" not in normalized.summary assert "super-secret" not in normalized.summary assert "base64-private-payload" not in normalized.summary @@ -84,9 +98,14 @@ def test_status_and_metrics_share_normalized_failure() -> None: tracker.register_document("normalization-doc", job_id="normalization-job") tracker.mark_processing("normalization-doc") metrics.record_document_accepted(document_id="normalization-doc", job_id="normalization-job") - response = client.post("/v1/internal/job-callback", json={ - "id": "normalization-doc", "status": "failed", "error_details": normalize_error(_ocr_error()).as_dict(), - }) + response = client.post( + "/v1/internal/job-callback", + json={ + "id": "normalization-doc", + "status": "failed", + "error_details": normalize_error(_ocr_error()).as_dict(), + }, + ) assert response.status_code == 200, response.text status = client.get("/v1/ingest/status/normalization-doc") metric = client.get("/v1/ingest/metrics/document/normalization-doc")