infra(litellm): /gpd/log endpoint with virtual-key auth + GCS streaming

Ships a tiny Dockerfile on top of ghcr.io/berriai/litellm-database:
main-stable that registers POST /gpd/log via LITELLM_WORKER_STARTUP_HOOKS
(PR anomalyco#22931 in v1.82.3-stable).

The route:
  1. Content-Length gate in BaseHTTPMiddleware — rejects >64 MB
     or missing header before the body touches RAM.
  2. Depends(user_api_key_auth) — LiteLLM's native validator. Handles
     revocation, expiry, team/user/org budget via common_checks. Per-
     virtual-key USD budget deliberately skipped (non-LLM route) so
     log writes never consume the user's LLM spend cap.
  3. proxy_logging_obj.pre_call_hook(call_type="pass_through_endpoint") —
     reuses LiteLLM's RPM/TPM limiter descriptors (shared with /v1/*).
  4. Redis per-key daily byte quota (GPD_LOG_BYTES_PER_DAY, default 1GB)
     via gpd:log:bytes:<hashed_token>:<yyyymmdd> counter. Fail-closed
     on Redis outage so quotas can never bypass.
  5. Server-derived user_hash = sha256(user_api_key_dict.user_id)[:16]
     — clients cannot write under another user's prefix.
  6. Idempotent GCS upload via if_generation_match=0. Treats 412 as
     success so client-side spill retries don't duplicate data.

Per-flush object names use 26-char ULIDs (lexicographic == chronological)
to sidestep GCS's 1-write/sec-per-object ceiling during subagent fan-out.
compactor.py fuses parts/*.jsonl.gz → root.jsonl.gz nightly via
Objects.compose() (32-at-a-time, two-phase for >32 parts).

Deploy: point Railway service at infra/litellm/Dockerfile, add
GOOGLE_APPLICATION_CREDENTIALS_JSON + GPD_LOG_BUCKET env vars, Redeploy.
See infra/litellm/README.md for step-by-step.

SA key never leaves Railway. Desktop never gets GCS credentials.

Author: cmaloney111

infra/litellm/Dockerfile

@@ -0,0 +1,16 @@
+# GPD's LiteLLM image: stock upstream + our custom /gpd/log route.
+#
+# Rolling tag: each Railway redeploy rebuilds from the latest
+# ghcr.io/berriai/litellm-database:main-stable. That's how LiteLLM
+# upgrades flow through — you don't have to touch anything locally,
+# just click "Redeploy" in Railway.
+FROM ghcr.io/berriai/litellm-database:main-stable
+
+# Our hook module — see infra/litellm/gpd_log/README.md for architecture.
+COPY gpd_log /app/gpd_log
+
+# Activate the hook. LiteLLM iterates this comma-separated list during
+# worker startup (proxy_server.py:777-803, PR #22931) and calls each
+# entry as `module:function`.
+ENV LITELLM_WORKER_STARTUP_HOOKS=gpd_log.hook:register
+ENV PYTHONPATH=/app

infra/litellm/README.md

@@ -0,0 +1,116 @@
+# GPD's LiteLLM image
+
+Ships stock `ghcr.io/berriai/litellm-database:main-stable` + one custom
+route, `POST /gpd/log`, for session-log ingest. The SA key that writes
+to `gs://gpd-desktop-logs` lives only on Railway — desktop clients
+authenticate with their existing LiteLLM virtual key.
+
+## What's in here
+
+| File | What it does |
+|---|---|
+| `Dockerfile` | 3-line layer on top of stock LiteLLM |
+| `gpd_log/hook.py` | `register()` entry point for `LITELLM_WORKER_STARTUP_HOOKS` |
+| `gpd_log/middleware.py` | Rejects requests with missing / oversized Content-Length before the body hits memory |
+| `gpd_log/handler.py` | The route: auth → rate-limit → byte-quota → GCS upload |
+| `gpd_log/gcs_writer.py` | `upload_from_string(if_generation_match=0)` idempotent write |
+| `gpd_log/quota.py` | Per-key daily byte counter in Redis, fail-closed |
+
+## One-time GCP setup
+
+```bash
+# SA scoped to write-only on one bucket — no list, no read, no delete.
+gcloud iam service-accounts create gpd-log-writer --project=gpd-desktop \
+  --display-name="GPD LiteLLM-side log writer"
+
+gcloud storage buckets add-iam-policy-binding gs://gpd-desktop-logs \
+  --member=serviceAccount:gpd-log-writer@gpd-desktop.iam.gserviceaccount.com \
+  --role=roles/storage.objectCreator
+
+gcloud iam service-accounts keys create /tmp/gpd-log-writer-key.json \
+  --iam-account=gpd-log-writer@gpd-desktop.iam.gserviceaccount.com \
+  --project=gpd-desktop
+
+# Print for pasting into Railway:
+cat /tmp/gpd-log-writer-key.json | jq -c .   # single-line JSON for env var
+shred -u /tmp/gpd-log-writer-key.json         # destroy local copy
+```
+
+## Railway configuration
+
+1. **Switch the LiteLLM service from "Docker image" mode to "Dockerfile" mode.**
+   Settings → Source → point at this repo, root directory `infra/litellm/`.
+2. **Add env vars** (Settings → Variables):
+   ```
+   GOOGLE_APPLICATION_CREDENTIALS_JSON = <paste the JSON from step above>
+   GPD_LOG_BUCKET = gpd-desktop-logs
+   GPD_LOG_BYTES_PER_DAY = 1073741824    # 1 GiB/day per key (optional; default)
+   # Existing vars stay untouched: DATABASE_URL, LITELLM_MASTER_KEY,
+   # REDIS_URL (or REDIS_HOST / REDIS_PASSWORD), STORE_MODEL_IN_DB, etc.
+   ```
+   `LITELLM_WORKER_STARTUP_HOOKS` is baked into the Dockerfile, so don't
+   set it as a Railway env var (would double-register the route).
+3. **Click Redeploy.** Railway rebuilds the image (~30s), the hook fires
+   during worker startup, and the `/gpd/log` route becomes live.
+
+## Verification
+
+After deploy, with a valid LiteLLM virtual key:
+
+```bash
+KEY=sk-<your-key>
+BASE=https://litellm-production-46bb.up.railway.app
+SEQ=$(python -c 'import secrets, time; import string; \
+  alpha="0123456789ABCDEFGHJKMNPQRSTVWXYZ"; \
+  print("".join(secrets.choice(alpha) for _ in range(26)))')
+
+# Small test body (gzipped NDJSON)
+echo '{"kind":"test","ts":1}' | gzip | curl -sS -X POST \
+  "$BASE/gpd/log?session=ses_test&root_session=ses_test&seq=$SEQ" \
+  -H "Authorization: Bearer $KEY" \
+  -H "Content-Encoding: gzip" \
+  -H "Content-Type: application/x-ndjson" \
+  --data-binary @- \
+  | jq
+
+# Expected: {"ok": true, "path": "user=.../session=ses_test/parts/<SEQ>.jsonl.gz", "bytes": N}
+
+# Check the object landed
+gcloud storage cat "gs://gpd-desktop-logs/user=*/date=$(date -u +%Y-%m-%d)/session=ses_test/parts/$SEQ.jsonl.gz" \
+  | gunzip
+```
+
+Negative tests:
+
+```bash
+# Missing key → 401
+curl -sS -X POST "$BASE/gpd/log?session=s&seq=$SEQ" -H "Content-Length: 10" -d abc
+# Revoked key → 401 (after LiteLLM cache TTL)
+# Missing Content-Length → 411
+# Oversize Content-Length (>64MB) → 413
+# Malformed seq → 400
+# Over daily byte quota → 429
+```
+
+## Client-side mapping
+
+Client POSTs `POST /gpd/log` with:
+
+- `Authorization: Bearer <virtual key>`
+- `Content-Encoding: gzip`
+- `Content-Type: application/x-ndjson`
+- Query: `session=<id>&root_session=<id>&seq=<ULID>`
+- Body: gzipped NDJSON — one `GpdLog.Event` per line
+
+The server writes the object to:
+```
+gs://gpd-desktop-logs/user=<hashed_user_id>/date=YYYY-MM-DD/session=<root_id>/parts/<seq>.jsonl.gz
+```
+or for subagents:
+```
+gs://gpd-desktop-logs/user=<hashed_user_id>/date=YYYY-MM-DD/session=<root_id>/subagents/agent-<child_id>/parts/<seq>.jsonl.gz
+```
+
+A nightly compactor (separate service) fuses `parts/*.jsonl.gz` into
+a single `root.jsonl.gz` per session per day, using GCS `Objects.compose()`
+in 32-at-a-time batches.

infra/litellm/gpd_log/__init__.py

@@ -0,0 +1,170 @@
+"""Nightly compactor: fuse parts/*.jsonl.gz into one root.jsonl.gz per session-day.
+
+Why: client flushes every ~1s (during active subagent fan-out, much faster).
+Per-flush objects hit GCS's 1-write/sec-per-object ceiling if we wrote to
+a single file. So client writes `parts/<ULID>.jsonl.gz` (unique object per
+flush), and this job fuses them at rest.
+
+Fusing strategy: lexicographic part-name order == chronological (ULID is
+timestamp-prefixed). Simple concat of gzipped streams works because gzip
+is concatenation-safe per RFC 1952 §2.2.
+
+Run modes:
+  - Railway cron:    `python -m gpd_log.compactor --yesterday`
+  - Cloud Function:  `compact_all()` as the entry point, HTTP trigger
+  - Ad-hoc:          `python -m gpd_log.compactor --date 2026-04-20`
+
+What survives: `parts/` is deleted after a successful fuse, leaving just
+`root.jsonl.gz`. The lifecycle policy (30d nearline, 90d coldline) then
+flows on the compacted object as normal.
+"""
+from __future__ import annotations
+
+import argparse
+import io
+import logging
+import os
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+from google.cloud import storage
+
+from .gcs_writer import _get_bucket
+
+log = logging.getLogger("gpd_log.compactor")
+
+# GCS limits: compose() takes up to 32 components. If a session has
+# more than 32 parts (subagent-heavy day), we do two passes.
+COMPOSE_MAX = 32
+
+
+def _list_sessions_for_date(bucket: storage.Bucket, date: str) -> list[str]:
+    """Return list of object prefixes that look like
+    `user=*/date=<date>/session=*/` (one per session).
+    """
+    # `Client.list_blobs(prefix=..., delimiter="/")` returns prefixes in
+    # the response's `prefixes` attr when called via list_blobs with
+    # delimiter. We iterate two levels of prefix to enumerate sessions.
+    prefixes: set[str] = set()
+    for user_prefix in _list_subdirs(bucket, ""):
+        if not user_prefix.startswith("user="):
+            continue
+        date_prefix = f"{user_prefix}date={date}/"
+        for session_prefix in _list_subdirs(bucket, date_prefix):
+            if session_prefix.startswith(f"{date_prefix}session="):
+                prefixes.add(session_prefix)
+            # Nested subagent dirs are compacted per-subagent, not per-root.
+            # Handle them by walking the tree another level:
+            subagents_prefix = f"{session_prefix}subagents/"
+            for sub in _list_subdirs(bucket, subagents_prefix):
+                prefixes.add(sub)
+    return sorted(prefixes)
+
+
+def _list_subdirs(bucket: storage.Bucket, prefix: str) -> list[str]:
+    client = bucket.client
+    it = client.list_blobs(bucket, prefix=prefix, delimiter="/")
+    # Exhaust iterator to populate `prefixes`.
+    list(it)
+    return sorted(it.prefixes or [])
+
+
+def _list_parts(bucket: storage.Bucket, session_prefix: str) -> list[storage.Blob]:
+    parts_prefix = f"{session_prefix}parts/"
+    blobs = list(bucket.client.list_blobs(bucket, prefix=parts_prefix))
+    # Lexicographic == chronological (ULID prefix).
+    return sorted(blobs, key=lambda b: b.name)
+
+
+def _compose_batch(
+    bucket: storage.Bucket,
+    destination_name: str,
+    sources: list[storage.Blob],
+) -> storage.Blob:
+    dest = bucket.blob(destination_name)
+    dest.content_encoding = "gzip"
+    dest.content_type = "application/x-ndjson"
+    dest.compose(sources)
+    return dest
+
+
+def compact_session(bucket: storage.Bucket, session_prefix: str) -> dict:
+    """Fuse parts/*.jsonl.gz → root.jsonl.gz for one session-day.
+
+    Returns {'session': prefix, 'parts': N, 'bytes': M, 'skipped': bool}.
+    Idempotent: if root.jsonl.gz already exists and is newer than the
+    oldest part, skip.
+    """
+    parts = _list_parts(bucket, session_prefix)
+    if not parts:
+        return {"session": session_prefix, "parts": 0, "skipped": True}
+
+    root_name = f"{session_prefix}root.jsonl.gz"
+    root = bucket.blob(root_name)
+    if root.exists(bucket.client):
+        root.reload()
+        if root.updated and parts[-1].updated and root.updated > parts[-1].updated:
+            return {"session": session_prefix, "parts": len(parts), "skipped": True}
+
+    # Two-phase compose for > 32 parts.
+    intermediates: list[storage.Blob] = []
+    if len(parts) <= COMPOSE_MAX:
+        final_sources = parts
+    else:
+        batch_idx = 0
+        for start in range(0, len(parts), COMPOSE_MAX):
+            batch = parts[start : start + COMPOSE_MAX]
+            tmp_name = f"{session_prefix}_compact/tmp-{batch_idx:04d}.jsonl.gz"
+            intermediates.append(_compose_batch(bucket, tmp_name, batch))
+            batch_idx += 1
+        final_sources = intermediates
+
+    total_bytes = sum((p.size or 0) for p in parts)
+    _compose_batch(bucket, root_name, final_sources)
+
+    # Delete parts only after the compose succeeded — makes the operation
+    # resumable if compose fails halfway.
+    for p in parts:
+        p.delete()
+    for i in intermediates:
+        i.delete()
+
+    return {"session": session_prefix, "parts": len(parts), "bytes": total_bytes, "skipped": False}
+
+
+def compact_all(date: str) -> dict:
+    bucket = _get_bucket()
+    results = []
+    for session_prefix in _list_sessions_for_date(bucket, date):
+        try:
+            r = compact_session(bucket, session_prefix)
+            results.append(r)
+            log.info("compacted", extra={"result": r})
+        except Exception as e:
+            log.exception("compact failed", extra={"session": session_prefix})
+            results.append({"session": session_prefix, "error": str(e)})
+    return {"date": date, "sessions": len(results), "details": results}
+
+
+def main() -> int:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--date", help="YYYY-MM-DD (defaults to yesterday UTC)")
+    parser.add_argument("--yesterday", action="store_true")
+    args = parser.parse_args()
+
+    if args.date:
+        date = args.date
+    else:
+        date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d")
+
+    t0 = time.time()
+    summary = compact_all(date)
+    dt = time.time() - t0
+    log.info(f"done date={date} sessions={summary['sessions']} elapsed={dt:.1f}s")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

infra/litellm/gpd_log/compactor.py

@@ -0,0 +1,81 @@
+"""Upload the /gpd/log request body to GCS.
+
+Idempotent: `if_generation_match=0` means "create iff object does not
+exist". A client-side spill retry that duplicates the object path gets
+a 412 Precondition Failed, which we treat as success. Combined with
+monotonic per-flush ULID object names, this gives at-least-once
+delivery with exactly-once storage.
+
+Credentials: `GOOGLE_APPLICATION_CREDENTIALS_JSON` env var (inline SA
+JSON, Railway-standard). Falls back to Application Default Credentials
+if unset, for local testing against gcloud auth login.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+
+from google.cloud import storage
+from google.oauth2 import service_account
+
+_client: storage.Client | None = None
+_bucket: storage.Bucket | None = None
+
+
+def _get_bucket() -> storage.Bucket:
+    global _client, _bucket
+    if _bucket is not None:
+        return _bucket
+
+    bucket_name = os.environ.get("GPD_LOG_BUCKET")
+    if not bucket_name:
+        raise RuntimeError("GPD_LOG_BUCKET env var required")
+
+    sa_json = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS_JSON")
+    if sa_json:
+        info = json.loads(sa_json)
+        creds = service_account.Credentials.from_service_account_info(
+            info,
+            scopes=["https://www.googleapis.com/auth/devstorage.read_write"],
+        )
+        _client = storage.Client(credentials=creds, project=info.get("project_id"))
+    else:
+        _client = storage.Client()
+
+    _bucket = _client.bucket(bucket_name)
+    return _bucket
+
+
+async def stream_to_gcs(object_path: str, body: bytes) -> int:
+    """Upload body as a GCS object at `object_path`.
+
+    Returns bytes written. Treats 412 (object exists) as success to make
+    client-side retry-from-spill safe.
+    """
+    bucket = _get_bucket()
+    blob = bucket.blob(object_path)
+    # Client sends gzipped NDJSON — preserve that encoding on the stored object
+    # so BigQuery's external-table reader (NEWLINE_DELIMITED_JSON) auto-decodes
+    # transparently via the .gz suffix.
+    blob.content_encoding = "gzip"
+    blob.content_type = "application/x-ndjson"
+
+    def _upload() -> int:
+        try:
+            blob.upload_from_string(
+                body,
+                content_type="application/x-ndjson",
+                if_generation_match=0,
+                retry=None,
+            )
+            return len(body)
+        except Exception as e:
+            msg = str(e)
+            # 412 Precondition Failed → object already exists. Client is
+            # retrying a previously-successful write. Idempotent.
+            if "preconditionFailed" in msg or "412" in msg:
+                return len(body)
+            raise
+
+    return await asyncio.to_thread(_upload)