From 2d270410808d30b5551c1295061c6716b65fe2fe Mon Sep 17 00:00:00 2001 From: seonghobae Date: Mon, 24 Aug 2026 17:40:21 +0900 Subject: [PATCH] fix(operability): structured orchestrator diagnostics behind generic 503s Issue #361: Global Ask's broad exception boundary captured unexpected programming defects but never carried them to server telemetry, and the 503 detail interpolated the raw provider exception message back to the caller. - new backend/app/orchestrator_boundary.py classifies known provider/transport/schema failures (warning-level orchestrator_provider_unavailable) from unexpected defects (error-level orchestrator_internal_fault with stack trace), keyed by a bounded operation-code vocabulary and per-request correlation id; - /api/ask now returns one static generic 503 detail on every failure path (no exception text leaves the service) while preserving exception chaining and emitting the two distinct alert events; - counters are process-local and exposed via boundary_counters() so operators get the availability-vs-defect metric split without adding a metrics dependency. --- backend/app/main.py | 21 +++- backend/app/orchestrator_boundary.py | 157 +++++++++++++++++++++++++ tests/test_orchestrator_boundary.py | 164 +++++++++++++++++++++++++++ 3 files changed, 336 insertions(+), 6 deletions(-) create mode 100644 backend/app/orchestrator_boundary.py create mode 100644 tests/test_orchestrator_boundary.py diff --git a/backend/app/main.py b/backend/app/main.py index 0e6fb68f2..48795b58f 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -21,6 +21,8 @@ import asyncio import json +import logging +import uuid from contextlib import asynccontextmanager from dataclasses import asdict from datetime import datetime @@ -52,6 +54,10 @@ from lineageweave.image_content import orchestrator_vision_client from lineageweave.embedding_client import orchestrator_embedding_client from lineageweave.llm_context import build_post_llm_metadata, use_llm_metadata +from .orchestrator_boundary import ( + new_correlation_id, + orchestrator_boundary, +) from lineageweave.corporate_hierarchy_inference import ( ContextualOrchestratorHierarchyInferenceClient, NullCorporateHierarchyInferenceClient, @@ -238,6 +244,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="LineageWeave API", lifespan=lifespan) +_logger = logging.getLogger(__name__) app.add_middleware( CORSMiddleware, allow_origins=load_settings().frontend_origins, @@ -2657,13 +2664,15 @@ async def ask_agent( "cited_post_images": [], "next_action": "No authorized source posts are available for this question.", } + correlation_id = new_correlation_id(uuid.uuid4().hex) try: - answer = await asyncio.to_thread(client.answer, question, sources) - except (HttpClientError, KeyError, OSError, ValueError) as exc: - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, - f"Ask Agent is unavailable: {exc}", - ) from exc + with orchestrator_boundary( + "global_ask", + "Ask Agent is unavailable: the orchestrator returned no complete evidence object", + ): + answer = await asyncio.to_thread(client.answer, question, sources) + except HTTPException: + raise cited_ids = list(answer.cited_post_ids) async with pool.acquire() as conn: lineage_graph = await lineage_graphs_for_posts( diff --git a/backend/app/orchestrator_boundary.py b/backend/app/orchestrator_boundary.py new file mode 100644 index 000000000..5256cc29d --- /dev/null +++ b/backend/app/orchestrator_boundary.py @@ -0,0 +1,157 @@ +"""Structured diagnostics at the contextual-orchestrator failure boundary. + +Customer-facing orchestrator-backed endpoints deliberately return one stable, +generic ``503`` regardless of what went wrong behind it, so provider traces, +exception text, prompts, responses, credentials, and tenant data never reach +the caller. The cost of that honest boundary is diagnosability: without extra +structure, an unexpected programming defect looks identical to a provider +outage and becomes an opaque availability incident (issue #361). + +This module restores operator-side diagnosability without weakening the +customer-facing contract: + +- Known provider/transport/schema failures (:class:`HttpClientError`, + ``KeyError``, ``OSError``, ``ValueError``) log a bounded + ``orchestrator_provider_unavailable`` event, bump the matching in-process + counter, and raise the generic ``503``. +- Any *unexpected* exception logs ``orchestrator_internal_fault`` with the + operation code, correlation id, exception class, and full stack trace + (via :meth:`logging.Logger.exception`), bumps the distinct + internal-fault counter, and raises the same generic ``503``. + +Forbidden content -- prompt text, model output, bearer tokens, provider keys, +source-post bodies -- is never passed to this module, so it can never leak +into logs or traces through it. Exception chaining (``raise ... from exc``) +is always preserved. + +Counters are plain process-local integers keyed by ``(event, operation_code)``; +``operation_code`` values come from a fixed call-site vocabulary, so metric +cardinality stays bounded. Alerting keys on the two event names: a rising +``orchestrator_provider_unavailable`` rate is an upstream availability signal, +while any ``orchestrator_internal_fault`` is a page-the-owner programming +defect. + +References (APA 7th): + +- OpenTelemetry Community. (2024). *A semantic approach to error handling in + telemetry*. https://opentelemetry.io/docs/specs/semconv/ -- error-type + attributes must stay low-cardinality and class-based, which is why only the + exception class name (never its message) is recorded as a field. +- Python Software Foundation. (2025). *The logging module: Logger.exception*. + https://docs.python.org/3/library/logging.html#logging.Logger.exception -- + ``Logger.exception`` inside an except block records the stack trace while + preserving normal exception propagation. +""" + +from __future__ import annotations + +import contextvars +import logging +import threading +from collections import defaultdict +from contextlib import contextmanager +from typing import Iterator + +from fastapi import HTTPException +from lineageweave.http_client import HttpClientError + +_logger = logging.getLogger(__name__) + +#: Exceptions that mean "the orchestrator/provider transport or its payload +#: failed". Anything else reaching this boundary is an unexpected programming +#: defect and gets the louder internal-fault treatment. +KNOWN_ORCHESTRATOR_EXCEPTIONS = (HttpClientError, KeyError, OSError, ValueError) + +#: Event names emitted on every boundary trip. Alerting should treat these +#: differently: provider-unavailable is upstream capacity, internal-fault is +#: a defect in this service. +EVENT_PROVIDER_UNAVAILABLE = "orchestrator_provider_unavailable" +EVENT_INTERNAL_FAULT = "orchestrator_internal_fault" + +_request_correlation_id: contextvars.ContextVar[str] = contextvars.ContextVar( + "orchestrator_boundary_correlation_id", default="" +) + +_counters_lock = threading.Lock() +_counters: defaultdict[tuple[str, str], int] = defaultdict(int) + + +def new_correlation_id(correlation_id: str) -> str: + """Install and return a fresh correlation id for this request/task. + + Callers generate the value (a short ``uuid4`` hex works well) so request + handlers stay the single owner of identity; the boundary only carries it + into log fields. An empty string means "not set" and simply omits the + field rather than emitting noise. + """ + token_value = str(correlation_id).strip() + _request_correlation_id.set(token_value) + return token_value + + +def current_correlation_id() -> str: + """Return the correlation id installed for this context, if any.""" + return _request_correlation_id.get() + + +def boundary_counters() -> dict[tuple[str, str], int]: + """Return a snapshot of ``(event, operation_code)`` counters. + + Tests assert against this snapshot; operators can expose it through a + health/metrics surface without adding a metrics dependency to the + backend install set. + """ + with _counters_lock: + return {key: value for key, value in _counters.items()} + + +def _bump(event: str, operation_code: str) -> None: + with _counters_lock: + _counters[(event, operation_code)] += 1 + + +@contextmanager +def orchestrator_boundary(operation_code: str, generic_detail: str) -> Iterator[None]: + """Translate orchestrator failures into one generic customer-facing 503. + + Wrap the narrowest span that calls into an orchestrator-backed client:: + + with orchestrator_boundary("global_ask", "Ask Agent is unavailable"): + answer = await asyncio.to_thread(client.answer, question, sources) + + Known provider/transport/schema exceptions become a warning-level + ``orchestrator_provider_unavailable`` event. Any other exception becomes + an error-level ``orchestrator_internal_fault`` event *with stack trace* + (``Logger.exception``). Both paths raise :class:`~fastapi.HTTPException` + with status 503 and exactly ``generic_detail`` -- never the underlying + exception message -- and both preserve exception chaining via + ``raise ... from exc``. + """ + try: + yield + except KNOWN_ORCHESTRATOR_EXCEPTIONS as exc: + _bump(EVENT_PROVIDER_UNAVAILABLE, operation_code) + _logger.warning( + "%s", + EVENT_PROVIDER_UNAVAILABLE, + extra={ + "event": EVENT_PROVIDER_UNAVAILABLE, + "operation_code": operation_code, + "correlation_id": current_correlation_id(), + "exception_class": type(exc).__name__, + }, + ) + raise HTTPException(503, generic_detail) from exc + except Exception as exc: + _bump(EVENT_INTERNAL_FAULT, operation_code) + _logger.exception( + "%s", + EVENT_INTERNAL_FAULT, + extra={ + "event": EVENT_INTERNAL_FAULT, + "operation_code": operation_code, + "correlation_id": current_correlation_id(), + "exception_class": type(exc).__name__, + }, + ) + raise HTTPException(503, generic_detail) from exc diff --git a/tests/test_orchestrator_boundary.py b/tests/test_orchestrator_boundary.py new file mode 100644 index 000000000..cb42f1cfa --- /dev/null +++ b/tests/test_orchestrator_boundary.py @@ -0,0 +1,164 @@ +"""Unit tests for the orchestrator failure boundary (issue #361). + +The boundary is the seam between honest customer-facing 503s and operator +diagnosability. These tests prove, without a live orchestrator or database: + +- known provider/transport/schema exceptions classify as + ``orchestrator_provider_unavailable`` (warning level, no stack trace + requirement) and raise one generic 503; +- unexpected exceptions (an injected ``AttributeError`` stands in for any + programming defect) classify as ``orchestrator_internal_fault`` at error + level *with* a stack trace and raise the same generic 503; +- the raw exception message never reaches the HTTP caller; +- exception chaining survives (``__cause__`` keeps the original error); +- log records carry operation code, correlation id, event name, and the + exception class only -- never prompt text, response text, tokens, or keys; +- counters separate provider-unavailable from internal-fault events per + operation, and stay bounded to the call sites actually used. +""" + +from __future__ import annotations + +import logging +from unittest.mock import MagicMock + +import pytest +from fastapi import HTTPException + +from backend.app.orchestrator_boundary import ( + EVENT_INTERNAL_FAULT, + EVENT_PROVIDER_UNAVAILABLE, + boundary_counters, + new_correlation_id, + orchestrator_boundary, +) +from lineageweave.http_client import HttpClientError + + +@pytest.fixture() +def _fresh_boundary_state(): + """Isolate correlation context and counters per test.""" + new_correlation_id("") + from backend.app import orchestrator_boundary as module + + module._counters.clear() + yield + module._counters.clear() + + +def _raise(exc: Exception) -> None: + raise exc + + +def test_known_provider_error_classifies_as_provider_unavailable( + caplog: pytest.LogCaptureFixture, _fresh_boundary_state +) -> None: + """An HttpClientError becomes a warning-level provider-unavailable 503.""" + with caplog.at_level(logging.WARNING): + with pytest.raises(HTTPException) as excinfo: + with orchestrator_boundary("global_ask", "Ask Agent is unavailable"): + _raise(HttpClientError("provider socket reset mid-answer")) + + assert excinfo.value.status_code == 503 + # The raw exception message must not reach the caller. + assert "socket reset" not in str(excinfo.value.detail) + record = next(r for r in caplog.records if r.__dict__.get("event") == EVENT_PROVIDER_UNAVAILABLE) + assert record.levelno == logging.WARNING + assert record.operation_code == "global_ask" + assert record.exception_class == "HttpClientError" + + +def test_unexpected_defect_classifies_as_internal_fault_with_stack( + caplog: pytest.LogCaptureFixture, _fresh_boundary_state +) -> None: + """An AttributeError-style defect gets an error-level trace + same 503.""" + with caplog.at_level(logging.ERROR): + with pytest.raises(HTTPException) as excinfo: + with orchestrator_boundary("global_ask", "Ask Agent is unavailable"): + _raise(AttributeError("'NoneType' object has no attribute 'answer'")) + + assert excinfo.value.status_code == 503 + record = next(r for r in caplog.records if r.__dict__.get("event") == EVENT_INTERNAL_FAULT) + assert record.levelno == logging.ERROR + assert record.exc_info is not None, "internal faults must carry a stack trace" + assert record.exception_class == "AttributeError" + + +def test_success_path_raises_nothing_and_logs_nothing( + caplog: pytest.LogCaptureFixture, _fresh_boundary_state +) -> None: + """A clean client call passes through untouched, without boundary events.""" + sentinel = MagicMock(return_value="answer") + with caplog.at_level(logging.DEBUG): + with orchestrator_boundary("global_ask", "Ask Agent is unavailable"): + result = sentinel("q", []) + assert result == "answer" + sentinel.assert_called_once_with("q", []) + assert not [r for r in caplog.records if hasattr(r, "event")] + + +def test_exception_chaining_is_preserved(_fresh_boundary_state) -> None: + """``raise ... from exc`` semantics survive both classification paths.""" + original = HttpClientError("transport down") + with pytest.raises(HTTPException) as excinfo: + with orchestrator_boundary("global_ask", "unavailable"): + _raise(original) + assert excinfo.value.__cause__ is original + + +def test_log_records_exclude_sensitive_fields( + caplog: pytest.LogCaptureFixture, _fresh_boundary_state +) -> None: + """Prompt/response/token/key strings may never appear in boundary logs.""" + secret_prompt = "confidential question about tenant payroll" + secret_token = "Bearer eyJhbGciOi-secret-value" + with caplog.at_level(logging.DEBUG): + with pytest.raises(HTTPException): + with orchestrator_boundary("global_ask", "Ask Agent is unavailable"): + _raise(HttpClientError(f"request rejected carrying {secret_token}")) + for record in caplog.records: + rendered = record.getMessage() + " ".join( + str(value) for key, value in record.__dict__.items() if key != "message" + ) + assert secret_prompt not in rendered + assert secret_token not in rendered + assert "rejected carrying" not in rendered + + +def test_correlation_id_flows_into_records( + caplog: pytest.LogCaptureFixture, _fresh_boundary_state +) -> None: + """The installed correlation id lands on every boundary record.""" + new_correlation_id("corr-123") + with caplog.at_level(logging.WARNING): + with pytest.raises(HTTPException): + with orchestrator_boundary("global_ask", "unavailable"): + _raise(HttpClientError("down")) + record = next(r for r in caplog.records if hasattr(r, "correlation_id")) + assert record.correlation_id == "corr-123" + + +def test_counters_separate_events_per_operation(_fresh_boundary_state) -> None: + """Provider-unavailable and internal-fault count independently.""" + for _ in range(3): + with pytest.raises(HTTPException): + with orchestrator_boundary("global_ask", "unavailable"): + _raise(HttpClientError("down")) + with pytest.raises(HTTPException): + with orchestrator_boundary("global_ask", "unavailable"): + _raise(AttributeError("unexpected defect")) + + snapshot = boundary_counters() + assert snapshot[(EVENT_PROVIDER_UNAVAILABLE, "global_ask")] == 3 + assert snapshot[(EVENT_INTERNAL_FAULT, "global_ask")] == 1 + + +def test_operation_code_vocabulary_stays_bounded(_fresh_boundary_state) -> None: + """Only fixed call-site operation codes keep metric cardinality bounded.""" + allowed = {"global_ask", "post_chat"} + for operation in sorted(allowed): + with pytest.raises(HTTPException): + with orchestrator_boundary(operation, "unavailable"): + _raise(HttpClientError("down")) + snapshot = boundary_counters() + assert {operation for (_, operation) in snapshot} <= allowed