feat(operability): structured server diagnostics behind the generic 503

backend/app/global_ask_queue.py

@@ -25,7 +25,9 @@
 
 import asyncpg
 import redis.asyncio as redis
+from fastapi import HTTPException, status
 
+from lineageweave.http_client import HttpClientError
 from lineageweave.post_chat import (
     PostChatClient,
     cited_post_evidence,
@@ -36,6 +38,7 @@
 
 from .config import GLOBAL_ASK_JOB_DEADLINE_SECONDS
 from .lineage_ingestion import lineage_graphs_for_posts
+from .operability import log_internal_fault, log_provider_unavailable
 from .post_chat_ingestion import _seoul_today, cited_post_images, gather_global_chat_sources
 
 GLOBAL_ASK_STREAM_KEY = "global_ask_request_stream"
@@ -179,9 +182,35 @@ 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
-    )
+    try:
+        answer = await asyncio.to_thread(
+            chat_client.answer, _temporally_grounded_question(question_text, today=today), sources
+        )
+    except (HttpClientError, OSError) as exc:
+        # Known transport/provider failure. Same generic 503 text on every
+        # failure path so callers cannot probe which internal classifier
+        # fired; the event_type distinction lives only in server logs.
+        log_provider_unavailable("global_ask", exc)
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "Ask Agent is unavailable: contextual-orchestrator could not complete the answer",
+        ) from exc
+    except (KeyError, ValueError) as exc:
+        # Contract/schema fault: the orchestrator responded but its payload
+        # did not match the evidence-object contract.
+        log_internal_fault("global_ask", exc)
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "Ask Agent is unavailable: contextual-orchestrator could not complete the answer",
+        ) from exc
+    except Exception as exc:
+        # Unexpected defect. Keep the customer boundary and emit a full
+        # structured internal-fault diagnostic (message-redacted).
+        log_internal_fault("global_ask", exc)
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "Ask Agent is unavailable: contextual-orchestrator could not complete the answer",
+        ) from exc
     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)

backend/app/operability.py

@@ -0,0 +1,127 @@
+"""Structured server-side operability logging for LLM-channel endpoints.
+
+Global Ask and every other orchestrator-backed endpoint deliberately hide
+provider failures behind a stable generic ``503`` so customer-facing
+responses never leak provider traces (ADR 0123 / CWE-209 discipline). The
+cost of that boundary is operator blindness: when the cause is an
+unexpected programming defect rather than provider unavailability, the
+generic response alone turns a regression into an opaque availability
+incident.
+
+This module restores operator diagnosability *without* weakening the
+customer boundary:
+
+- **Two event types.** ``orchestrator_provider_unavailable`` marks a known,
+  expected transport/provider failure (connection refused, HTTP error from
+  the orchestrator gateway). ``orchestrator_internal_fault`` marks an
+  unexpected exception -- a programming defect or contract break -- and
+  carries the full stack trace. Alerting keys on ``event_type`` so pager
+  load distinguishes "provider down" from "our bug".
+- **Correlation ids.** Each diagnostic carries a random correlation id so an
+  incident report can be matched to exactly one log line without exposing
+  anything account-scoped.
+- **Forbidden fields.** Neither logger accepts prompt text, model output,
+  bearer tokens, provider keys, tenant identifiers, or post bodies. Only
+  the operation code, correlation id, and exception *class name* are
+  logged for provider faults. Internal faults carry the stack *frames*
+  (raise site) but not the original exception message: Python tracebacks
+  end with ``ExceptionType: message``, and parsing/transport messages can
+  embed provider payloads or prompt fragments, so both helpers re-emit
+  through a placeholder exception whose text is a fixed redaction notice.
+  Callers cannot opt back into raw messages -- the redaction is the
+  contract, not a default.
+
+References: issue #361; ADR 0123 (non-disclosure boundary).
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+
+_LOGGER = logging.getLogger("lineageweave.operability")
+
+PROVIDER_UNAVAILABLE_EVENT = "orchestrator_provider_unavailable"
+INTERNAL_FAULT_EVENT = "orchestrator_internal_fault"
+
+_REDACTED_NOTICE = (
+    "[message redacted: operability records carry class and frames only]"
+)
+
+
+class _MessageRedacted(Exception):
+    """Traceback carrier whose text is a fixed redaction notice."""
+
+
+def _new_correlation_id() -> str:
+    """Return a fresh correlation id safe to expose in incident reports."""
+    return uuid.uuid4().hex
+
+
+def log_provider_unavailable(operation: str, exc: Exception) -> str:
+    """Record a known provider/transport failure at warning level.
+
+    Emits one structured record keyed on
+    :data:`PROVIDER_UNAVAILABLE_EVENT` with the operation code, a fresh
+    correlation id, and the exception class name -- deliberately *not* the
+    exception message, which may embed provider URLs or payload fragments.
+    Returns the correlation id so the caller could surface it in a
+    follow-up activity entry if a future increment wants request-scoped
+    references.
+
+    Args:
+        operation: Stable operation code, e.g. ``"global_ask"``.
+        exc: The caught transport/provider exception.
+
+    Returns:
+        The correlation id attached to the emitted record.
+    """
+    correlation_id = _new_correlation_id()
+    _LOGGER.warning(
+        "%s",
+        PROVIDER_UNAVAILABLE_EVENT,
+        extra={
+            "event_type": PROVIDER_UNAVAILABLE_EVENT,
+            "operation": operation,
+            "correlation_id": correlation_id,
+            "exception_class": type(exc).__name__,
+        },
+    )
+    return correlation_id
+
+
+def log_internal_fault(operation: str, exc: Exception) -> str:
+    """Record an unexpected programming/contract fault at error level.
+
+    Emits one structured record keyed on :data:`INTERNAL_FAULT_EVENT` with
+    the operation code, a fresh correlation id, the exception class name,
+    and the raise-site stack frames. The original exception *message* is
+    never emitted: Python renders a traceback's final line as
+    ``ExceptionType: message``, and parsing/transport messages can embed
+    provider payloads or prompt fragments, so the record re-raises through
+    :class:`_MessageRedacted` -- same ``__traceback__`` (the frames an
+    engineer needs), fixed redaction notice as text. Exception chaining is
+    preserved by the caller's ``raise ... from exc``.
+
+    Args:
+        operation: Stable operation code, e.g. ``"global_ask"``.
+        exc: The unexpected exception.
+
+    Returns:
+        The correlation id attached to the emitted record.
+    """
+    correlation_id = _new_correlation_id()
+    carrier = _MessageRedacted(_REDACTED_NOTICE)
+    carrier.__traceback__ = exc.__traceback__
+    _LOGGER.error(
+        "%s",
+        INTERNAL_FAULT_EVENT,
+        exc_info=(type(exc), carrier, exc.__traceback__),
+        extra={
+            "event_type": INTERNAL_FAULT_EVENT,
+            "operation": operation,
+            "correlation_id": correlation_id,
+            "exception_class": type(exc).__name__,
+        },
+    )
+    return correlation_id

scripts/migrate_legacy_namespace.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Migrate stored ``post_project_mention.ontology_iri`` values off the
+deprecated repository-case namespace onto the canonical lowercase one.
+
+ADR 0157 keeps ``https://contextualwisdomlab.github.io/lineageweave/ontology#``
+canonical and demotes ``https://contextualwisdomlab.github.io/LineageWeave/ontology#``
+to a deprecated compatibility namespace. New writes mint only canonical
+IRIs (``lineageweave.ontology`` loads the lowercase graph), but rows written
+before the decision can still carry repository-case IRIs. RDF consumers treat
+the two spellings as different resources, so leaving them split makes
+downstream joins miss mentions that are semantically identical.
+
+This tool is deliberately *not* silent:
+
+- default mode is **dry run**: it prints every row it would change and exits;
+- ``--apply`` performs exactly the printed rewrites inside one transaction;
+- the extraction provenance columns (``extraction_method``, confidence,
+  evidence text) are never touched -- only the IRI spelling moves, so the
+  evidence chain of who extracted what remains intact per ADR 0157's
+  "do not silently rewrite historical evidence" rule;
+- any IRI outside the two known namespaces is reported and left alone so an
+  unexpected third spelling cannot be bulk-mangled.
+
+Usage::
+
+    python scripts/migrate_legacy_namespace.py --dsn postgresql://...
+    python scripts/migrate_legacy_namespace.py --dsn postgresql://... --apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import asyncpg
+
+CANONICAL_NAMESPACE = "https://contextualwisdomlab.github.io/lineageweave/ontology#"
+LEGACY_NAMESPACE = "https://contextualwisdomlab.github.io/LineageWeave/ontology#"
+
+
+def canonicalize(iri: str) -> str | None:
+    """Return the canonical spelling of ``iri``, or None if not legacy."""
+    if iri.startswith(LEGACY_NAMESPACE):
+        return CANONICAL_NAMESPACE + iri[len(LEGACY_NAMESPACE):]
+    return None
+
+
+async def migrate(dsn: str, apply: bool) -> int:
+    """Scan, report, and optionally rewrite legacy namespace IRIs.
+
+    Args:
+        dsn: PostgreSQL DSN for the target database.
+        apply: False for dry-run reporting; True to execute the rewrite.
+
+    Returns:
+        Process exit code: 0 when clean or migrated, 1 on unexpected IRIs.
+    """
+    conn = await asyncpg.connect(dsn)
+    try:
+        rows = await conn.fetch(
+            """
+            select post_id, project_name, ontology_iri
+            from post_project_mention
+            where ontology_iri is not null
+            order by post_id, project_name
+            """
+        )
+        unexpected: list[tuple[str, str, str]] = []
+        planned: list[tuple[str, str, str]] = []
+        for row in rows:
+            iri = row["ontology_iri"]
+            canonical = canonicalize(iri)
+            if canonical is None:
+                if not iri.startswith(CANONICAL_NAMESPACE):
+                    unexpected.append((row["post_id"], row["project_name"], iri))
+                continue
+            planned.append((row["post_id"], row["project_name"], f"{iri} -> {canonical}"))
+
+        print(f"scanned {len(rows)} mention row(s) with a non-null ontology_iri")
+        for post_id, project_name, change in planned:
+            print(f"  {post_id} / {project_name}: {change}")
+        for post_id, project_name, iri in unexpected:
+            print(
+                f"  UNEXPECTED {post_id} / {project_name}: {iri} "
+                f"(neither namespace; left untouched)"
+            )
+        if unexpected:
+            print(f"{len(unexpected)} row(s) carry an unrecognized namespace; nothing written")
+            return 1
+        if not planned:
+            print("no legacy namespace rows remain")
+            return 0
+        if not apply:
+            print(f"dry run: {len(planned)} row(s) would be rewritten; pass --apply to write")
+            return 0
+
+        async with conn.transaction():
+            for post_id, project_name, change in planned:
+                _old, _, new = change.rpartition(" -> ")
+                updated = await conn.execute(
+                    """
+                    update post_project_mention
+                    set ontology_iri = $3
+                    where post_id = $1 and project_name = $2 and ontology_iri = $4
+                    """,
+                    post_id,
+                    project_name,
+                    new,
+                    new.replace(CANONICAL_NAMESPACE, LEGACY_NAMESPACE),
+                )
+                if updated != "UPDATE 1":
+                    raise RuntimeError(f"row changed during migration: {post_id}/{project_name}")
+        print(f"applied: {len(planned)} row(s) rewritten to the canonical namespace")
+        return 0
+    finally:
+        await conn.close()
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI entry point."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dsn", required=True, help="PostgreSQL DSN for the target database")
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="execute the rewrite; without this flag the tool only reports",
+    )
+    args = parser.parse_args(argv)
+    return __import__("asyncio").run(migrate(args.dsn, args.apply))
+
+
+if __name__ == "__main__":
+    sys.exit(main())