feat(passthrough): multi-provider (OpenAI + Gemini) capture to request_logs

.env.example

@@ -138,6 +138,27 @@
 # ENABLE_REQUEST_LOGGING=false
 
 
+# === PASSTHROUGH =================================================
+
+# Enable the /openai/* and /gemini/* passthrough capture routes. Default off: these routes forward client-supplied upstream credentials, so an always-on deployment would relay traffic upstream and write request_logs for anyone with network reach.
+# PASSTHROUGH_ROUTES_ENABLED=false
+
+# Max bytes of a streamed passthrough response buffered in memory for request_logs capture. The stream forwarded to the client is unaffected; only the recorded body is truncated beyond this.
+# PASSTHROUGH_STREAM_CAPTURE_MAX_BYTES=10485760
+
+# Enable asynchronous materialization of eligible passthrough request_logs into conversation history.
+# PASSTHROUGH_MATERIALIZE_ENABLED=false
+
+# Enable one-shot passthrough materialization backfill for historical request_logs.
+# PASSTHROUGH_MATERIALIZE_BACKFILL_ENABLED=false
+
+# Seconds between dashboard reconciliation scans for unmaterialized passthrough request_logs.
+# PASSTHROUGH_MATERIALIZE_RECONCILE_INTERVAL_SECONDS=300
+
+# Maximum passthrough request_logs processed per materialization batch.
+# PASSTHROUGH_MATERIALIZE_BATCH_SIZE=200
+
+
 # === TELEMETRY ===================================================
 
 # Anonymous usage telemetry (None defers to DB config, True/False overrides)

changelog.d/passthrough-materialization.md

@@ -0,0 +1,25 @@
+---
+category: Features
+---
+
+**Passthrough OpenAI/Gemini calls now materialize into conversation history, search, and export**
+  - Previously the `/openai/*` and `/gemini/*` passthrough routes only wrote
+    raw `request_logs`; those calls never appeared in `/api/history`,
+    `/api/debug/calls`, session summaries, FTS, or the JSONL export, and were
+    unreadable by downstream tooling.
+  - Adds a `passthrough_materialize` package that normalizes captured OpenAI
+    (chat + Responses, buffered + streamed) and Gemini (generateContent +
+    streamGenerateContent) payloads into the canonical Anthropic-shaped
+    conversation-event contract while preserving the exact provider-native
+    request/response verbatim for faithful reprobe.
+  - Materialization is idempotent (advisory lock + request-event existence
+    guard, single transaction, two session-summary updates) and driven two
+    ways: a live post-commit callback on `RequestLogRecorder` (gated by
+    `PASSTHROUGH_MATERIALIZE_ENABLED`) and a dashboard-only reconcile worker +
+    one-shot backfill CLI (gated by `PASSTHROUGH_MATERIALIZE_BACKFILL_ENABLED`)
+    for historical rows.
+  - Passthrough requests now carry user attribution via the same trusted
+    `X-Luthien-User-Id` / Bearer-JWT policy as the Anthropic path. Malformed or
+    unsupported eligible payloads fail loudly and stay retryable; no partial
+    rows are ever written. Existing read paths light up with zero new
+    migrations and SQLite/Postgres parity.

changelog.d/passthrough-multiprovider-capture.md

@@ -0,0 +1,10 @@
+---
+category: Features
+pr: 796
+---
+
+**Multi-provider passthrough capture**: capture OpenAI (`/openai/*`) and Gemini (`/gemini/*`) passthrough traffic into `request_logs`, mirroring the existing Anthropic `/v1/*` capture.
+  - Streaming and non-streaming responses are recorded; payloads are sanitized before persistence.
+  - Cross-provider session grouping via the existing header/metadata contract, so a single logical conversation is retrievable regardless of provider.
+  - Disabled by default: the routes forward client-supplied upstream credentials, so they only mount when `PASSTHROUGH_ROUTES_ENABLED=true` (otherwise they 404). This prevents an always-on deployment from acting as an open relay.
+  - Streamed-response capture is bounded by `PASSTHROUGH_STREAM_CAPTURE_MAX_BYTES` (default 10 MiB); the client stream is never affected, and the recorded body is flagged `capture_truncated` beyond the limit.

pyproject.toml

@@ -31,7 +31,9 @@ dependencies = [
     "click>=8.1.0",
     "cryptography>=44.0.0",
     "httpx>=0.28.1",
+    "google-genai>=1.33.0",
     "jsonschema>=4.17.0",
+    "openai>=2.11.0",
     "opentelemetry-api>=1.20.0",
     "opentelemetry-sdk>=1.20.0",
     "opentelemetry-exporter-otlp-proto-grpc>=1.20.0",
@@ -48,6 +50,7 @@ dependencies = [
     "anthropic>=0.84.0",
     "aiohttp>=3.9.0",
     "sentry-sdk[fastapi]>=2.54.0",
+    "fastapi>=0.115.0",
 ]
 
 [tool.hatch.version]
@@ -141,6 +144,7 @@ dev = [
     "luthien-cli",
     "pytest-httpx>=0.35.0",
     "pytest-xdist>=3.6.0",
+    "basedpyright>=1.39.9",
 ]
 
 [tool.uv.sources]

scripts/backfill_passthrough_materialization.py

@@ -0,0 +1,48 @@
+"""Drain the passthrough materialization backfill from configured storage."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+import click
+
+from luthien_proxy.passthrough_materialize.backfill import drain_passthrough_backfill
+from luthien_proxy.settings import get_settings
+from luthien_proxy.telemetry import configure_logging
+from luthien_proxy.utils.db import DatabasePool
+
+logger = logging.getLogger(__name__)
+
+
+async def _run_backfill() -> None:
+    settings = get_settings()
+    if not settings.database_url:
+        raise click.ClickException("DATABASE_URL must be configured")
+    db_pool = DatabasePool(settings.database_url)
+    try:
+        await db_pool.get_pool()
+        totals = await drain_passthrough_backfill(
+            db_pool,
+            limit=settings.passthrough_materialize_batch_size,
+        )
+    finally:
+        await db_pool.close()
+    logger.info(
+        "Passthrough backfill complete: materialized=%d already_materialized=%d skipped_ineligible=%d failed=%d",
+        totals.materialized,
+        totals.already_materialized,
+        totals.skipped_ineligible,
+        totals.failed,
+    )
+
+
+@click.command()
+def main() -> None:
+    """Drain configured passthrough materialization backfill."""
+    asyncio.run(_run_backfill())
+
+
+if __name__ == "__main__":
+    configure_logging()
+    main()

src/luthien_proxy/config_fields.py

@@ -242,6 +242,44 @@ class ConfigFieldMeta:
         category="observability",
     ),
 
+    # ── passthrough ───────────────────────────────────────────────────────
+    ConfigFieldMeta(
+        "passthrough_routes_enabled", "PASSTHROUGH_ROUTES_ENABLED", bool, False,
+        "Enable the /openai/* and /gemini/* passthrough capture routes. Default off: these "
+        "routes forward client-supplied upstream credentials, so an always-on deployment would "
+        "relay traffic upstream and write request_logs for anyone with network reach.",
+        category="passthrough",
+    ),
+    ConfigFieldMeta(
+        "passthrough_stream_capture_max_bytes", "PASSTHROUGH_STREAM_CAPTURE_MAX_BYTES", int, 10485760,
+        "Max bytes of a streamed passthrough response buffered in memory for request_logs capture. "
+        "The stream forwarded to the client is unaffected; only the recorded body is truncated beyond this.",
+        category="passthrough",
+    ),
+    ConfigFieldMeta(
+        "passthrough_materialize_enabled", "PASSTHROUGH_MATERIALIZE_ENABLED", bool, False,
+        "Enable asynchronous materialization of eligible passthrough request_logs into conversation history.",
+        category="passthrough",
+    ),
+    ConfigFieldMeta(
+        "passthrough_materialize_backfill_enabled", "PASSTHROUGH_MATERIALIZE_BACKFILL_ENABLED", bool, False,
+        "Enable one-shot passthrough materialization backfill for historical request_logs.",
+        category="passthrough",
+    ),
+    ConfigFieldMeta(
+        "passthrough_materialize_reconcile_interval_seconds",
+        "PASSTHROUGH_MATERIALIZE_RECONCILE_INTERVAL_SECONDS",
+        int,
+        300,
+        "Seconds between dashboard reconciliation scans for unmaterialized passthrough request_logs.",
+        category="passthrough",
+    ),
+    ConfigFieldMeta(
+        "passthrough_materialize_batch_size", "PASSTHROUGH_MATERIALIZE_BATCH_SIZE", int, 200,
+        "Maximum passthrough request_logs processed per materialization batch.",
+        category="passthrough",
+    ),
+
     # ── telemetry ─────────────────────────────────────────────────────────
     ConfigFieldMeta(
         "usage_telemetry", "USAGE_TELEMETRY", bool, None,
@@ -374,6 +412,7 @@ class ConfigFieldMeta:
     "llm",
     "security",
     "observability",
+    "passthrough",
     "telemetry",
     "retention",
     "webhook",

src/luthien_proxy/dependencies.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass, field
 from typing import Any
 
+import httpx
 from fastapi import Depends, HTTPException, Request
 from redis.asyncio import Redis
 
@@ -49,6 +50,8 @@ class Dependencies:
     rate_limiter: TokenBucketRateLimiter | None = field(default=None)
     last_credential_info: dict[str, Any] = field(default_factory=dict)
     webhook_sender: WebhookSender | None = field(default=None)
+    passthrough_streaming_client: httpx.AsyncClient | None = field(default=None)
+    passthrough_buffered_client: httpx.AsyncClient | None = field(default=None)
 
     def get_anthropic_policy(self) -> AnthropicExecutionInterface:
         """Get the current Anthropic policy.

src/luthien_proxy/main.py

@@ -11,6 +11,7 @@
 from collections.abc import MutableMapping
 from contextlib import asynccontextmanager
 
+import httpx
 import uvicorn
 from fastapi import FastAPI, Request
 from fastapi.exceptions import HTTPException as FastAPIHTTPException
@@ -40,6 +41,8 @@
 )
 from luthien_proxy.observability.redis_event_publisher import RedisEventPublisher
 from luthien_proxy.observability.sentry import init_sentry
+from luthien_proxy.passthrough_materialize.worker import PassthroughReconcileWorker
+from luthien_proxy.passthrough_routes import router as passthrough_router
 from luthien_proxy.pipeline.upstream_headers import validate_upstream_headers_at_startup
 from luthien_proxy.policy_manager import PolicyManager
 from luthien_proxy.rate_limit import TokenBucketRateLimiter
@@ -278,6 +281,11 @@ async def lifespan(app: FastAPI):
         if _enable_request_logging:
             logger.info("Request/response logging ENABLED")
 
+        _passthrough_streaming_client = httpx.AsyncClient(
+            timeout=httpx.Timeout(connect=10.0, read=300.0, write=10.0, pool=30.0)
+        )
+        _passthrough_buffered_client = httpx.AsyncClient(timeout=120.0)
+
         # Initialize usage telemetry
         settings = get_settings()
         _telemetry_config = await resolve_telemetry_config(
@@ -349,6 +357,15 @@ async def lifespan(app: FastAPI):
                 )
             logger.info("Conversation retention disabled (CONVERSATION_RETENTION_DAYS not set)")
 
+        _passthrough_reconcile_worker: PassthroughReconcileWorker | None = None
+        if settings.passthrough_materialize_backfill_enabled:
+            _passthrough_reconcile_worker = PassthroughReconcileWorker(
+                db_pool=db_pool,
+                limit=settings.passthrough_materialize_batch_size,
+                interval_seconds=settings.passthrough_materialize_reconcile_interval_seconds,
+            )
+            _passthrough_reconcile_worker.start()
+
         # Initialize webhook sender
         _webhook_url = settings.webhook_url or None
         _webhook_sender = WebhookSender(
@@ -381,6 +398,8 @@ async def lifespan(app: FastAPI):
             config_registry=_config_registry,
             rate_limiter=_rate_limiter,
             webhook_sender=_webhook_sender,
+            passthrough_streaming_client=_passthrough_streaming_client,
+            passthrough_buffered_client=_passthrough_buffered_client,
         )
 
         # Store dependencies container in app state
@@ -399,10 +418,14 @@ async def lifespan(app: FastAPI):
         # before request handling has fully drained, fire_and_forget calls
         # could land against an already-closed httpx client.
         await _webhook_sender.stop()
+        if _passthrough_reconcile_worker is not None:
+            await _passthrough_reconcile_worker.stop()
         if _purger is not None:
             await _purger.stop()
         if _telemetry_sender is not None:
             await _telemetry_sender.stop()
+        await _passthrough_streaming_client.aclose()
+        await _passthrough_buffered_client.aclose()
         await _inference_provider_registry.close()
         await _credential_manager.close()
         await anthropic_client_cache.close_all()
@@ -464,6 +487,7 @@ async def dispatch(self, request: Request, call_next):
 
     # Include routers
     app.include_router(gateway_router)  # /v1/messages
+    app.include_router(passthrough_router)  # /openai/* and /gemini/*
     app.include_router(debug_router)  # /api/debug/*
     app.include_router(ui_router)  # /activity/*, /policy-config, /diffs
     app.include_router(admin_router)  # /api/admin/* (policy management)