fix(operability): structured orchestrator diagnostics behind generic 503s

backend/app/global_ask_queue.py

@@ -36,6 +36,7 @@
 
 from .config import GLOBAL_ASK_JOB_DEADLINE_SECONDS
 from .lineage_ingestion import lineage_graphs_for_posts
+from .orchestrator_boundary import new_correlation_id, orchestrator_boundary
 from .post_chat_ingestion import _seoul_today, cited_post_images, gather_global_chat_sources
 
 GLOBAL_ASK_STREAM_KEY = "global_ask_request_stream"
@@ -179,9 +180,13 @@ def can_see(row: asyncpg.Record) -> bool:
             "cited_post_images": [],
             "next_action": "No authorized source posts are available for this question.",
         }
-    answer = await asyncio.to_thread(
-        chat_client.answer, _temporally_grounded_question(question_text, today=today), sources
-    )
+    with orchestrator_boundary(
+        "global_ask",
+        "Ask Agent is unavailable: the orchestrator returned no complete evidence object",
+    ):
+        answer = await asyncio.to_thread(
+            chat_client.answer, _temporally_grounded_question(question_text, today=today), sources
+        )
     cited_ids = list(answer.cited_post_ids)
     async with pool.acquire() as conn:
         lineage_graph = await lineage_graphs_for_posts(conn, can_see, cited_ids)
@@ -250,6 +255,7 @@ async def process_global_ask_job(
         )
     if row is None:
         return
+    new_correlation_id(job_id)
     try:
         async with pool.acquire() as conn:
             entity_ids, has_post_read = await load_account_visibility(

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

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