diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index a4d311d63..1eb108d6f 100644 --- a/backend/app/global_ask_queue.py +++ b/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) diff --git a/backend/app/operability.py b/backend/app/operability.py new file mode 100644 index 000000000..f51e93ddb --- /dev/null +++ b/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 diff --git a/scripts/migrate_legacy_namespace.py b/scripts/migrate_legacy_namespace.py new file mode 100644 index 000000000..758997970 --- /dev/null +++ b/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()) diff --git a/tests/test_migrate_legacy_namespace.py b/tests/test_migrate_legacy_namespace.py new file mode 100644 index 000000000..e49093a2e --- /dev/null +++ b/tests/test_migrate_legacy_namespace.py @@ -0,0 +1,171 @@ +"""Tests for scripts/migrate_legacy_namespace.py (ADR 0157 tooling). + +The migration must be deterministic, dry-run by default, refuse unknown +namespaces, and never touch provenance columns. These tests exercise the +pure ``canonicalize`` mapping and the async scan/rewrite flow against an +in-memory fake connection -- no live PostgreSQL required. +""" + +from __future__ import annotations + +import asyncio +import importlib.util +from pathlib import Path + +import pytest + +_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "migrate_legacy_namespace.py" +_spec = importlib.util.spec_from_file_location("migrate_legacy_namespace", _SCRIPT) +migrate_legacy_namespace = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(migrate_legacy_namespace) + +CANONICAL = migrate_legacy_namespace.CANONICAL_NAMESPACE +LEGACY = migrate_legacy_namespace.LEGACY_NAMESPACE + + +class TestCanonicalize: + def test_maps_legacy_to_canonical(self) -> None: + assert migrate_legacy_namespace.canonicalize(f"{LEGACY}Project") == f"{CANONICAL}Project" + + def test_canonical_rows_are_left_alone(self) -> None: + iri = f"{CANONICAL}Person" + assert migrate_legacy_namespace.canonicalize(iri) is None + + def test_unknown_namespaces_return_none(self) -> None: + assert migrate_legacy_namespace.canonicalize("https://example.com/other#Thing") is None + + def test_fragment_is_preserved_exactly(self) -> None: + term = "CorporateEntity" + mapped = migrate_legacy_namespace.canonicalize(f"{LEGACY}{term}") + assert mapped == f"{CANONICAL}{term}" + assert mapped.endswith(term) + + +class FakeRecord: + def __init__(self, post_id: str, project_name: str, ontology_iri: str): + self._data = { + "post_id": post_id, + "project_name": project_name, + "ontology_iri": ontology_iri, + } + + def __getitem__(self, key: str): + return self._data[key] + + +@pytest.fixture() +def _patch_connect(monkeypatch: pytest.MonkeyPatch): + """Route asyncpg.connect to a factory over a caller-supplied connection.""" + holder: dict = {} + + def _factory(conn): + def _connect(dsn): + assert "postgresql://" in dsn + return _AsyncReturn(conn) + holder["conn"] = conn + return _connect + + holder["factory"] = _factory + yield holder + + +class _AsyncReturn: + """Awaitable that resolves immediately.""" + + def __init__(self, value): + self._value = value + + def __await__(self): + if False: + yield + return self._value + + +class FakeConnection: + """Minimal asyncpg surface: one select, transactional updates.""" + + def __init__(self, rows: list[FakeRecord]): + self.rows = rows + self.updates: list[tuple] = [] + self.transaction_entered = False + + async def fetch(self, query: str): + assert "post_project_mention" in query + return self.rows + + def transaction(self): + return self + + async def __aenter__(self): + self.transaction_entered = True + return self + + async def __aexit__(self, *exc_info): + return False + + async def execute(self, query: str, *args): + assert "update post_project_mention" in query + self.updates.append(args) + return "UPDATE 1" + + async def close(self): + pass + + +def test_dry_run_reports_without_writing(capsys: pytest.CaptureFixture[str], _patch_connect, monkeypatch: pytest.MonkeyPatch) -> None: + rows = [ + FakeRecord("p1", "Alpha", f"{LEGACY}Project"), + FakeRecord("p2", "Beta", f"{CANONICAL}Team"), + ] + conn = FakeConnection(rows) + monkeypatch.setattr(migrate_legacy_namespace.asyncpg, "connect", _patch_connect["factory"](conn)) + rc = asyncio.run(migrate_legacy_namespace.migrate("postgresql://unused", apply=False)) + + assert rc == 0 + out = capsys.readouterr().out + assert "dry run" in out + assert f"{LEGACY}Project -> {CANONICAL}Project" in out + assert conn.updates == [] + assert not conn.transaction_entered + + +def test_apply_rewrites_only_legacy_rows(_patch_connect, monkeypatch: pytest.MonkeyPatch) -> None: + rows = [ + FakeRecord("p1", "Alpha", f"{LEGACY}Project"), + FakeRecord("p2", "Beta", f"{CANONICAL}Team"), + ] + conn = FakeConnection(rows) + monkeypatch.setattr(migrate_legacy_namespace.asyncpg, "connect", _patch_connect["factory"](conn)) + rc = asyncio.run(migrate_legacy_namespace.migrate("postgresql://unused", apply=True)) + + assert rc == 0 + assert len(conn.updates) == 1 + post_id, project_name, new_iri, old_iri = conn.updates[0] + assert (post_id, project_name) == ("p1", "Alpha") + assert new_iri == f"{CANONICAL}Project" + assert old_iri == f"{LEGACY}Project" + + +def test_unknown_namespace_fails_closed(capsys: pytest.CaptureFixture[str], _patch_connect, monkeypatch: pytest.MonkeyPatch) -> None: + rows = [FakeRecord("p3", "Gamma", "https://example.com/weird#X")] + conn = FakeConnection(rows) + monkeypatch.setattr(migrate_legacy_namespace.asyncpg, "connect", _patch_connect["factory"](conn)) + rc = asyncio.run(migrate_legacy_namespace.migrate("postgresql://unused", apply=False)) + + assert rc == 1 + out = capsys.readouterr().out + assert "UNEXPECTED" in out + assert "nothing written" in out + assert conn.updates == [] + + +def test_clean_database_is_a_no_op(capsys: pytest.CaptureFixture[str], _patch_connect, monkeypatch: pytest.MonkeyPatch) -> None: + rows = [FakeRecord("p4", "Delta", f"{CANONICAL}Post")] + conn = FakeConnection(rows) + monkeypatch.setattr(migrate_legacy_namespace.asyncpg, "connect", _patch_connect["factory"](conn)) + rc = asyncio.run(migrate_legacy_namespace.migrate("postgresql://unused", apply=True)) + + assert rc == 0 + out = capsys.readouterr().out + assert "no legacy namespace rows remain" in out + assert conn.updates == [] diff --git a/tests/test_operability.py b/tests/test_operability.py new file mode 100644 index 000000000..547aa46dd --- /dev/null +++ b/tests/test_operability.py @@ -0,0 +1,124 @@ +"""Unit tests for the structured operability logger (issue #361). + +These run without any live stack: the module under test is pure logging, +so ``caplog`` verifies both what IS recorded (event type, operation code, +correlation id, exception class, stack trace for internal faults) and +what must never be (exception messages that could embed provider URLs, +bearer tokens, or prompt/response text). +""" + +from __future__ import annotations + +import logging + +import pytest + +from backend.app.operability import ( + INTERNAL_FAULT_EVENT, + PROVIDER_UNAVAILABLE_EVENT, + log_internal_fault, + log_provider_unavailable, +) + + +@pytest.fixture() +def _operability_level(caplog: pytest.LogCaptureFixture): + """Capture the operability logger at warning-and-below verbosity.""" + caplog.set_level(logging.WARNING, logger="lineageweave.operability") + return caplog + + +def test_provider_unavailable_records_operation_and_class(_operability_level) -> None: + """A known transport failure logs the provider-unavailable event type.""" + + class FakeHttpError(RuntimeError): + pass + + correlation_id = log_provider_unavailable("global_ask", FakeHttpError("connect to provider failed")) + + record = _operability_level.records[-1] + assert record.levelno == logging.WARNING + assert record.event_type == PROVIDER_UNAVAILABLE_EVENT + assert record.operation == "global_ask" + assert record.correlation_id == correlation_id + assert record.exception_class == "FakeHttpError" + + +def test_provider_unavailable_never_logs_the_exception_message(_operability_level) -> None: + """The message may embed provider URLs or payload fragments; it stays out.""" + secret = "bearer eyJhbGciOi-secret-token" + log_provider_unavailable("global_ask", RuntimeError(f"POST https://orchestrator failed with {secret}")) + + rendered = _operability_level.records[-1].getMessage() + assert secret not in rendered + assert "orchestrator" not in _operability_level.records[-1].__dict__.get("exception_class", "") + + +def test_internal_fault_carries_frames_and_class(_operability_level) -> None: + """An unexpected defect logs error-level with raise-site frames attached.""" + try: + raise AttributeError("'NoneType' object has no attribute 'answer'") + except AttributeError as exc: + original_traceback = exc.__traceback__ + correlation_id = log_internal_fault("global_ask", exc) + + record = _operability_level.records[-1] + assert record.levelno == logging.ERROR + assert record.event_type == INTERNAL_FAULT_EVENT + assert record.operation == "global_ask" + assert record.correlation_id == correlation_id + assert record.exception_class == "AttributeError" + # exc_info is attached so the stack frames reach structured telemetry, + # and the original traceback object is preserved for the raise site. + assert record.exc_info is not None + assert record.exc_info[2] is original_traceback + + +def test_internal_fault_redacts_the_exception_message(_operability_level) -> None: + """Tracebacks end with 'Class: message' -- the message must be redacted. + + A parsing exception's str() can embed provider payload or prompt + fragments; the emitted traceback must end in the fixed redaction + notice instead of that text (devin SEC thread on PR #577). + """ + secret = 'provider payload {"korean_summary": "민감한 내용"} leaked' + try: + raise ValueError(f"chat response did not match the required format: {secret}") + except ValueError as exc: + log_internal_fault("global_ask", exc) + + import traceback + + record = _operability_level.records[-1] + rendered = "".join(traceback.format_exception(*record.exc_info)) + assert secret not in rendered + assert "redacted" in rendered + # Frames are still present: the raise site stays diagnosable. + assert __file__.split("/")[-1] not in rendered or "test_" in rendered + + +def test_correlation_ids_are_unique_per_event(_operability_level) -> None: + """Two faults produce distinct correlation ids so reports stay separable.""" + first = log_provider_unavailable("global_ask", RuntimeError("first")) + second = log_internal_fault("global_ask", ValueError("second")) + + records = _operability_level.records[-2:] + assert {r.event_type for r in records} == { + PROVIDER_UNAVAILABLE_EVENT, + INTERNAL_FAULT_EVENT, + } + assert first != second + + +def test_no_prompt_or_response_text_is_emitted(_operability_level) -> None: + """The forbidden-field contract: prompt/response content never lands.""" + prompt = "What happened between these linked events in post body ..." + response_text = "The answer text a model produced." + exc = RuntimeError(prompt + response_text) + log_provider_unavailable("post_chat", exc) + log_internal_fault("post_chat", exc) + + for record in _operability_level.records[-2:]: + rendered = record.getMessage() + assert prompt not in rendered + assert response_text not in rendered