ci(e2e): record the e2e suite weekly and replay it on weekdays with zero egress (#38163)

Adds a scheduled GitHub Actions lane on top of the merged record/replay
transport. A Saturday cron records the `replayable` e2e tests against the
real providers and publishes the fixture bundle as a private
`e2e-fixtures-bundle` artifact with a SHA-256 sidecar. Weekday crons pull
that artifact by its pinned digest, verify the checksum before extracting,
and replay it with provider credentials set to bogus values, so a run that
ever reached a real provider fails instead of passing.

An egress sentinel pins the provider hostnames to a local sink for the whole
replay job and counts every connection that reaches them; the job asserts
that count is zero, so hermeticity is proven by measurement. A red Saturday
publishes no bundle, so the next weekday finds nothing fresh and fails loudly
rather than replaying a week-old recording, and the transport's seven-day
freshness gate hard-fails any bundle that has drifted too far. The lane also
runs on demand from the Actions tab with a record/replay `mode` input.

Tests join the lane with `@pytest.mark.replayable`. The streaming Anthropic
test now counts to twenty so its recorded response banks several content
deltas, matching the assertion that the stream arrives incrementally.

.github/scripts/e2e_egress_sentinel.py

@@ -0,0 +1,198 @@
+"""Prove an e2e replay run makes zero outbound provider calls, by counting them.
+
+`serve` pins each provider host (`--host`) to a local sink address in the hosts
+file and binds a counting listener on that address, so any connection the proxy
+or the record/replay edge opens to a real provider is redirected to the sink,
+recorded as one line in `--hits-file`, and never leaves the box. The record and
+replay edge only ever dials `127.0.0.1:<edge-port>` (a different host than the
+pinned provider names), so in a clean replay the sink sees nothing; a single hit
+means a provider call escaped the bundle. `assert-empty` turns that hit file into
+the pass/fail check.
+
+Stdlib only, so CI runs it under the system interpreter as root (binding :443 and
+editing the hosts file both need root); `--sink-address`, `--port`, and
+`--hosts-file` are injectable so it runs unprivileged against a temp hosts file on
+a high port under test.
+"""
+
+# ruff: noqa: T201  # CLI script: its stdout/stderr progress and results are the interface
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import signal
+import socket
+import sys
+import threading
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from types import FrameType
+from typing import Final
+
+_BLOCK_BEGIN: Final = "# BEGIN e2e-egress-sentinel"
+_BLOCK_END: Final = "# END e2e-egress-sentinel"
+
+
+@dataclass(frozen=True, slots=True)
+class ServeConfig:
+    hosts: tuple[str, ...]
+    sink_address: str
+    ports: tuple[int, ...]
+    hits_file: Path
+    hosts_file: Path
+    ready_file: Path | None
+    pid_file: Path | None
+
+
+def _pin_block(sink_address: str, hosts: tuple[str, ...]) -> str:
+    lines = "\n".join(f"{sink_address}\t{host}" for host in hosts)
+    return f"\n{_BLOCK_BEGIN}\n{lines}\n{_BLOCK_END}\n"
+
+
+def _install_pins(hosts_file: Path, sink_address: str, hosts: tuple[str, ...]) -> bytes:
+    original = hosts_file.read_bytes() if hosts_file.exists() else b""
+    hosts_file.write_bytes(original + _pin_block(sink_address, hosts).encode())
+    return original
+
+
+def _restore_pins(hosts_file: Path, original: bytes) -> None:
+    hosts_file.write_bytes(original)
+
+
+def _bind(sink_address: str, port: int) -> socket.socket:
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    listener.bind((sink_address, port))
+    listener.listen(128)
+    return listener
+
+
+@dataclass(frozen=True, slots=True)
+class _HitLog:
+    path: Path
+    _lock: threading.Lock
+
+    def record(self, *, port: int, peer: tuple[str, int]) -> None:
+        entry = json.dumps({"ts": time.time(), "port": port, "peer": list(peer)})
+        with self._lock:
+            with self.path.open("a", encoding="utf-8") as handle:
+                handle.write(entry + "\n")
+
+
+def _serve_socket(listener: socket.socket, port: int, hits: _HitLog, stop: threading.Event) -> None:
+    while not stop.is_set():
+        try:
+            conn, peer = listener.accept()
+        except OSError:
+            return
+        hits.record(port=port, peer=(peer[0], peer[1]))
+        try:
+            conn.close()
+        except OSError:
+            pass
+
+
+def serve(config: ServeConfig) -> int:
+    config.hits_file.write_text("", encoding="utf-8")
+    original_hosts = _install_pins(config.hosts_file, config.sink_address, config.hosts)
+    try:
+        listeners = tuple(_bind(config.sink_address, port) for port in config.ports)
+    except OSError as exc:
+        _restore_pins(config.hosts_file, original_hosts)
+        print(f"egress sentinel could not bind a sink: {exc}", file=sys.stderr)
+        return 1
+
+    stop = threading.Event()
+    hits = _HitLog(path=config.hits_file, _lock=threading.Lock())
+    threads = tuple(
+        threading.Thread(target=_serve_socket, args=(listener, port, hits, stop), daemon=True)
+        for listener, port in zip(listeners, config.ports)
+    )
+    for thread in threads:
+        thread.start()
+
+    def _handle(_signum: int, _frame: FrameType | None) -> None:
+        stop.set()
+        for listener in listeners:
+            try:
+                listener.close()
+            except OSError:
+                pass
+
+    signal.signal(signal.SIGTERM, _handle)
+    signal.signal(signal.SIGINT, _handle)
+
+    if config.pid_file is not None:
+        config.pid_file.write_text(str(os.getpid()), encoding="utf-8")
+    if config.ready_file is not None:
+        config.ready_file.write_text("ready", encoding="utf-8")
+    print(
+        f"egress sentinel up: pinned {', '.join(config.hosts)} to {config.sink_address} "
+        f"on port(s) {', '.join(str(p) for p in config.ports)}",
+        flush=True,
+    )
+
+    stop.wait()
+    _restore_pins(config.hosts_file, original_hosts)
+    if config.ready_file is not None and config.ready_file.exists():
+        config.ready_file.unlink()
+    if config.pid_file is not None and config.pid_file.exists():
+        config.pid_file.unlink()
+    return 0
+
+
+def assert_empty(hits_file: Path) -> int:
+    if not hits_file.exists():
+        print(f"egress sentinel recorded no provider calls ({hits_file} absent): zero egress")
+        return 0
+    hits = [line for line in hits_file.read_text(encoding="utf-8").splitlines() if line.strip()]
+    if not hits:
+        print("egress sentinel recorded no provider calls: zero egress")
+        return 0
+    print(f"egress sentinel recorded {len(hits)} provider call(s); replay was not hermetic:", file=sys.stderr)
+    for line in hits:
+        print(f"  {line}", file=sys.stderr)
+    return 1
+
+
+def _serve_from_args(args: argparse.Namespace) -> int:
+    config = ServeConfig(
+        hosts=tuple(args.host),
+        sink_address=args.sink_address,
+        ports=tuple(args.port),
+        hits_file=Path(args.hits_file),
+        hosts_file=Path(args.hosts_file),
+        ready_file=Path(args.ready_file) if args.ready_file else None,
+        pid_file=Path(args.pid_file) if args.pid_file else None,
+    )
+    return serve(config)
+
+
+def main(argv: tuple[str, ...]) -> int:
+    parser = argparse.ArgumentParser(description="count outbound provider calls during an e2e replay")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    serve_parser = sub.add_parser("serve", help="pin provider hosts and count connection attempts")
+    serve_parser.add_argument("--host", action="append", required=True, help="provider host to pin and watch")
+    serve_parser.add_argument("--sink-address", default="127.0.0.1")
+    serve_parser.add_argument("--port", action="append", type=int, default=None)
+    serve_parser.add_argument("--hits-file", required=True)
+    serve_parser.add_argument("--hosts-file", default="/etc/hosts")
+    serve_parser.add_argument("--ready-file", default=None)
+    serve_parser.add_argument("--pid-file", default=None)
+
+    assert_parser = sub.add_parser("assert-empty", help="exit non-zero if any provider call was recorded")
+    assert_parser.add_argument("--hits-file", required=True)
+
+    args = parser.parse_args(argv)
+    if args.command == "serve":
+        if args.port is None:
+            args.port = [443]
+        return _serve_from_args(args)
+    return assert_empty(Path(args.hits_file))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(tuple(sys.argv[1:])))

.github/scripts/e2e_fetch_fixture_bundle.sh

@@ -0,0 +1,55 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+REPO="${1:-${GITHUB_REPOSITORY:?REPO required}}"
+ARTIFACT_NAME="${2:-e2e-fixtures-bundle}"
+BASE_BRANCH="${3:?base branch required}"
+DEST_DIR="${4:?destination bundle dir required}"
+
+: "${GH_TOKEN:?GH_TOKEN required to query and download artifacts}"
+
+WORKDIR="$(mktemp -d)"
+trap 'rm -rf "${WORKDIR}"' EXIT
+
+echo "resolving newest non-expired '${ARTIFACT_NAME}' artifact on ${REPO}@${BASE_BRANCH}"
+
+SELECTED="$(
+  gh api "repos/${REPO}/actions/artifacts" -X GET -f per_page=100 --paginate \
+    --jq ".artifacts[] | select(.name == \"${ARTIFACT_NAME}\" and .expired == false and .workflow_run.head_branch == \"${BASE_BRANCH}\") | {id, digest, created_at, run_id: .workflow_run.id, run_number: .workflow_run.run_number}" \
+    | jq -s 'sort_by(.created_at) | reverse | .[0] // empty'
+)"
+
+if [[ -z "${SELECTED}" ]]; then
+  echo "no usable '${ARTIFACT_NAME}' artifact on ${BASE_BRANCH}: the last record run produced none (a red Saturday), so there is nothing fresh to replay; failing loudly instead of replaying a stale bundle" >&2
+  exit 1
+fi
+
+RUN_ID="$(echo "${SELECTED}" | jq -r '.run_id')"
+RUN_NUMBER="$(echo "${SELECTED}" | jq -r '.run_number')"
+ARTIFACT_ID="$(echo "${SELECTED}" | jq -r '.id')"
+GH_DIGEST="$(echo "${SELECTED}" | jq -r '.digest // "unknown"')"
+CREATED_AT="$(echo "${SELECTED}" | jq -r '.created_at')"
+
+echo "pinned bundle: run #${RUN_NUMBER} (run_id=${RUN_ID}, artifact_id=${ARTIFACT_ID}), recorded ${CREATED_AT}, github digest ${GH_DIGEST}"
+
+gh run download "${RUN_ID}" --repo "${REPO}" -n "${ARTIFACT_NAME}" -D "${WORKDIR}"
+
+TARBALL="$(find "${WORKDIR}" -name '*.tar.gz' -type f | head -n 1)"
+if [[ -z "${TARBALL}" ]]; then
+  echo "downloaded artifact contained no tarball" >&2
+  exit 1
+fi
+SIDECAR="${TARBALL}.sha256"
+if [[ ! -f "${SIDECAR}" ]]; then
+  echo "downloaded artifact has no ${SIDECAR}: cannot verify the bundle digest" >&2
+  exit 1
+fi
+
+echo "verifying bundle against its recorded sha256 digest"
+( cd "$(dirname "${TARBALL}")" && sha256sum -c "$(basename "${SIDECAR}")" )
+
+mkdir -p "${DEST_DIR}"
+tar xzf "${TARBALL}" -C "${DEST_DIR}"
+
+echo "extracted bundle into ${DEST_DIR}"
+python3 -c "import json,sys; m=json.load(open(sys.argv[1])); print('  recorded_at', m['recorded_at'], 'harness', m['harness_version'], 'format_version', m['format_version'])" "${DEST_DIR}/manifest.json"

.github/scripts/e2e_pack_fixture_bundle.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ $# -ne 2 ]]; then
+  echo "usage: $0 <bundle-dir> <out-tarball>" >&2
+  exit 2
+fi
+
+BUNDLE_DIR="$1"
+OUT_TARBALL="$2"
+
+MANIFEST="${BUNDLE_DIR}/manifest.json"
+if [[ ! -f "${MANIFEST}" ]]; then
+  echo "no ${MANIFEST}: refusing to publish a bundle with no manifest (record produced nothing)" >&2
+  exit 1
+fi
+
+echo "packing fixture bundle from ${BUNDLE_DIR}"
+python3 -c "import json,sys; m=json.load(open(sys.argv[1])); print('  format_version', m['format_version'], 'recorded_at', m['recorded_at'], 'harness', m['harness_version'])" "${MANIFEST}"
+
+TEST_DIRS=$(find "${BUNDLE_DIR}" -mindepth 1 -maxdepth 1 -type d | wc -l | tr -d ' ')
+if [[ "${TEST_DIRS}" -eq 0 ]]; then
+  echo "bundle at ${BUNDLE_DIR} has a manifest but no recorded interactions; refusing to publish an empty bundle" >&2
+  exit 1
+fi
+echo "  ${TEST_DIRS} recorded test director(ies)"
+
+mkdir -p "$(dirname "${OUT_TARBALL}")"
+tar czf "${OUT_TARBALL}" -C "${BUNDLE_DIR}" .
+
+OUT_DIR="$(cd "$(dirname "${OUT_TARBALL}")" && pwd)"
+OUT_BASE="$(basename "${OUT_TARBALL}")"
+( cd "${OUT_DIR}" && sha256sum "${OUT_BASE}" > "${OUT_BASE}.sha256" )
+
+echo "wrote ${OUT_TARBALL} ($(du -h "${OUT_TARBALL}" | cut -f1)) and ${OUT_BASE}.sha256"
+cat "${OUT_DIR}/${OUT_BASE}.sha256"

.github/workflows/e2e_record_replay.yml

@@ -0,0 +1,237 @@
+name: "E2E Record and Replay"
+
+on:
+  schedule:
+    - cron: "0 8 * * 6"
+    - cron: "0 8 * * 1-5"
+  workflow_dispatch:
+    inputs:
+      mode:
+        description: "record (hits real providers and publishes a fresh bundle) or replay (bundle only, zero provider egress)"
+        type: choice
+        options:
+          - record
+          - replay
+        default: record
+
+permissions:
+  contents: read
+
+jobs:
+  record:
+    name: "Record the e2e suite against real providers"
+    if: >-
+      (github.event_name != 'schedule' || github.repository == 'BerriAI/litellm') &&
+      (github.event.schedule == '0 8 * * 6' ||
+       (github.event_name == 'workflow_dispatch' && github.event.inputs.mode == 'record'))
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    services:
+      postgres:
+        image: postgres:16.6
+        env:
+          POSTGRES_USER: llmproxy
+          POSTGRES_PASSWORD: dbpassword9090
+          POSTGRES_DB: litellm
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U llmproxy"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
+      LITELLM_MASTER_KEY: sk-e2e-record-replay
+      LITELLM_LOCAL_MODEL_COST_MAP: "True"
+      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+      ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.12"
+
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Cache the Rust build
+        uses: ./.github/actions/cache-cargo-build
+
+      - name: Install dependencies
+        run: |
+          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra proxy
+
+      - name: Cache Prisma binaries
+        uses: ./.github/actions/cache-prisma-binaries
+
+      - name: Generate Prisma client
+        run: |
+          uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
+
+      - name: Start the proxy
+        run: |
+          nohup uv run --no-sync litellm --config tests/e2e/gateway/record_replay_ci_config.yml --port 4000 > proxy.log 2>&1 &
+          for _ in $(seq 1 90); do
+            if curl -fs http://localhost:4000/health/liveliness > /dev/null; then
+              exit 0
+            fi
+            sleep 2
+          done
+          echo "proxy never became live"
+          tail -n 100 proxy.log
+          exit 1
+
+      - name: Record the replayable e2e lane
+        env:
+          E2E_FIXTURE_MODE: record
+        run: |
+          uv run --no-sync pytest tests/e2e -m replayable --reruns 0 -v --tb=short -rA
+
+      - name: Pack the fixture bundle
+        run: |
+          .github/scripts/e2e_pack_fixture_bundle.sh tests/e2e/.fixtures "${RUNNER_TEMP}/bundle/e2e-fixtures.tar.gz"
+
+      - name: Publish the fixture bundle
+        uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1
+        with:
+          name: e2e-fixtures-bundle
+          path: |
+            ${{ runner.temp }}/bundle/e2e-fixtures.tar.gz
+            ${{ runner.temp }}/bundle/e2e-fixtures.tar.gz.sha256
+          if-no-files-found: error
+          retention-days: 30
+
+      - name: Show proxy log on failure
+        if: failure()
+        run: tail -n 300 proxy.log
+
+  replay:
+    name: "Replay the e2e suite from the pinned bundle with zero egress"
+    if: >-
+      (github.event_name != 'schedule' || github.repository == 'BerriAI/litellm') &&
+      (github.event.schedule == '0 8 * * 1-5' ||
+       (github.event_name == 'workflow_dispatch' && github.event.inputs.mode == 'replay'))
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    permissions:
+      contents: read
+      actions: read
+    services:
+      postgres:
+        image: postgres:16.6
+        env:
+          POSTGRES_USER: llmproxy
+          POSTGRES_PASSWORD: dbpassword9090
+          POSTGRES_DB: litellm
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U llmproxy"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      DATABASE_URL: postgresql://llmproxy:dbpassword9090@localhost:5432/litellm
+      LITELLM_MASTER_KEY: sk-e2e-record-replay
+      LITELLM_LOCAL_MODEL_COST_MAP: "True"
+      GH_TOKEN: ${{ github.token }}
+      OPENAI_API_KEY: sk-replay-must-never-reach-a-provider
+      ANTHROPIC_API_KEY: sk-ant-replay-must-never-reach-a-provider
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.12"
+
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Cache the Rust build
+        uses: ./.github/actions/cache-cargo-build
+
+      - name: Install dependencies
+        run: |
+          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra proxy
+
+      - name: Cache Prisma binaries
+        uses: ./.github/actions/cache-prisma-binaries
+
+      - name: Generate Prisma client
+        run: |
+          uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
+
+      - name: Fetch the pinned fixture bundle by digest
+        env:
+          BASE_BRANCH: ${{ github.ref_name }}
+        run: |
+          .github/scripts/e2e_fetch_fixture_bundle.sh \
+            "${GITHUB_REPOSITORY}" \
+            e2e-fixtures-bundle \
+            "${BASE_BRANCH}" \
+            tests/e2e/.fixtures
+
+      - name: Start the proxy
+        run: |
+          nohup uv run --no-sync litellm --config tests/e2e/gateway/record_replay_ci_config.yml --port 4000 > proxy.log 2>&1 &
+          for _ in $(seq 1 90); do
+            if curl -fs http://localhost:4000/health/liveliness > /dev/null; then
+              exit 0
+            fi
+            sleep 2
+          done
+          echo "proxy never became live"
+          tail -n 100 proxy.log
+          exit 1
+
+      - name: Start the egress sentinel
+        run: |
+          # shellcheck disable=SC2024  # the log redirect is deliberately the runner user's, so a later non-sudo cat can read it
+          sudo python3 .github/scripts/e2e_egress_sentinel.py serve \
+            --host api.openai.com \
+            --host api.anthropic.com \
+            --hits-file "${RUNNER_TEMP}/egress-hits.jsonl" \
+            --ready-file "${RUNNER_TEMP}/egress-ready" \
+            --pid-file "${RUNNER_TEMP}/egress.pid" \
+            > "${RUNNER_TEMP}/egress-sentinel.log" 2>&1 &
+          for _ in $(seq 1 30); do
+            if [[ -f "${RUNNER_TEMP}/egress-ready" ]]; then
+              cat "${RUNNER_TEMP}/egress-sentinel.log"
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "egress sentinel never became ready"
+          cat "${RUNNER_TEMP}/egress-sentinel.log"
+          exit 1
+
+      - name: Replay the replayable e2e lane
+        env:
+          E2E_FIXTURE_MODE: replay
+        run: |
+          uv run --no-sync pytest tests/e2e -m replayable --reruns 0 -v --tb=short -rA
+
+      - name: Stop the egress sentinel and assert zero provider egress
+        if: always()
+        run: |
+          if [[ -f "${RUNNER_TEMP}/egress.pid" ]]; then
+            sudo kill -TERM "$(cat "${RUNNER_TEMP}/egress.pid")" 2>/dev/null || true
+            sleep 2
+          fi
+          python3 .github/scripts/e2e_egress_sentinel.py assert-empty --hits-file "${RUNNER_TEMP}/egress-hits.jsonl"
+
+      - name: Show proxy log on failure
+        if: failure()
+        run: tail -n 300 proxy.log

tests/e2e/CLAUDE.md

@@ -94,9 +94,9 @@ E2E_FIXTURE_MODE=record E2E_FIXTURE_DIR=/tmp/e2e-fixtures E2E_RESET_SPEND_LOGS=1
 E2E_FIXTURE_MODE=replay E2E_FIXTURE_DIR=/tmp/e2e-fixtures E2E_RESET_SPEND_LOGS=1 uv run pytest tests/e2e/llm_translation/test_chat_completions_contract_e2e.py
 ```
 
-Point the proxy at bogus provider credentials for the replay run and it still has to pass: that is the whole proof that nothing left the process. Bundles are never committed. `tests/e2e/.fixtures` is gitignored because a bundle holds verbatim provider response bodies and hard-fails after seven days, and publishing one for CI is LIT-5748
+Point the proxy at bogus provider credentials for the replay run and it still has to pass: that is the whole proof that nothing left the process. Bundles are never committed. `tests/e2e/.fixtures` is gitignored because a bundle holds verbatim provider response bodies and hard-fails after seven days. CI records and replays this lane on a schedule in `.github/workflows/e2e_record_replay.yml`, publishing the bundle as a private `e2e-fixtures-bundle` artifact instead of committing it, selecting the tests with the `@pytest.mark.replayable` marker, and proving the bogus-credentials replay hermetic by counting provider egress with `.github/scripts/e2e_egress_sentinel.py`
 
-Current limits: CI wiring is LIT-5748, Bedrock cannot be mounted (SigV4 signs the Host header, so a rewritten api_base fails signature verification), deployments baked into the proxy's config file cannot be edge-wired (only `/model/new` registrations can carry the edge api_base), and a file upload routed by `custom_llm_provider` through the proxy's `files_settings` block never passes a deployment at all, so the batches `model_param` and `provider_fallback` scenarios keep uploading live in every mode
+Current limits: Bedrock cannot be mounted (SigV4 signs the Host header, so a rewritten api_base fails signature verification), deployments baked into the proxy's config file cannot be edge-wired (only `/model/new` registrations can carry the edge api_base), and a file upload routed by `custom_llm_provider` through the proxy's `files_settings` block never passes a deployment at all, so the batches `model_param` and `provider_fallback` scenarios keep uploading live in every mode
 
 ## Typing
 

tests/e2e/CONTRIBUTING.md

@@ -61,11 +61,15 @@ E2E_FIXTURE_MODE=record E2E_FIXTURE_DIR=/tmp/e2e-fixtures uv run pytest tests/e2
 E2E_FIXTURE_MODE=replay E2E_FIXTURE_DIR=/tmp/e2e-fixtures uv run pytest tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py -v
 ```
 
-Bundles stay local. `tests/e2e/.fixtures` is gitignored because a bundle holds verbatim provider response bodies and expires seven days after it was recorded, so record the suite you want before you replay it and never commit the result; publishing bundles for CI is LIT-5748
+Bundles stay local. `tests/e2e/.fixtures` is gitignored because a bundle holds verbatim provider response bodies and expires seven days after it was recorded, so record the suite you want before you replay it and never commit the result. CI keeps its bundle out of git too, as a private GitHub Actions artifact rather than a committed file, for the same reason
+
+In CI the `.github/workflows/e2e_record_replay.yml` lane runs record and replay on a schedule. A Saturday cron records the `replayable` marker's tests against the real providers and publishes the bundle as a private `e2e-fixtures-bundle` artifact carrying a SHA-256 sidecar; weekday crons pull that artifact by its pinned digest, verify the checksum before extracting, and replay it with provider credentials deliberately set to bogus values, so a run that ever reached a real provider would fail instead of passing. An egress sentinel (`.github/scripts/e2e_egress_sentinel.py`) pins the provider hostnames to a local sink for the whole replay job and counts every connection that reaches them, and the job asserts that count is zero, so hermeticity is proven by measurement rather than by an absent bill. A red Saturday publishes no bundle, so the next weekday finds nothing fresh and fails loudly rather than replaying a week-old recording, and the seven-day freshness gate hard-fails any bundle that has drifted too far from the live providers. Run the lane on demand from the Actions tab with the `mode` input: `record` re-records and republishes, `replay` replays the current bundle. A test joins the lane by carrying `@pytest.mark.replayable` on top of its edge wiring, so add that marker only to a test whose provider traffic actually replays with zero egress
 
 One sharp edge: a replayed response reuses the recorded provider response id, and that id is the primary key of `LiteLLM_SpendLogs`, so replaying against a database that still holds the record run's rows silently dedupes the spend writes and a spend assertion fails with zero rows. Run both commands above with `E2E_RESET_SPEND_LOGS=1` (and `DATABASE_URL` set in the pytest env) so each session truncates the spend log table after itself, or point replay at a fresh database
 
-Replay answers any provider call that drifted from the recording with an HTTP 599 whose body names the computed and closest recorded keys, so the test fails loudly instead of silently going live, and a bundle older than seven days fails at collection time naming its age; either way the fix is to re-record. Only tests that register edge-wired deployments participate: everything else hits its provider live in every mode, so record exactly the suite you replay. If the proxy runs in a container, set `E2E_PROVIDER_EDGE_ADVERTISE_HOST` (e.g. `host.docker.internal`) so the api_base the proxy stores can reach the edge on the pytest host, and `E2E_PROVIDER_EDGE_BIND_HOST=0.0.0.0` so the edge accepts it. The suites wired to the edge today are `quota_management/spend_tracking/test_provider_edge_spend_e2e.py`, `llm_translation/test_chat_completions_contract_e2e.py`, the OpenAI registrations in `llm_translation/test_embeddings_endpoint_e2e.py`, the Anthropic tests in `llm_translation/test_messages_e2e.py`, streamed and not, and the OpenAI batch deployment behind `batches/`. A streamed response replays as the chunk sequence the provider sent rather than one buffered body. See `CLAUDE.md` in this directory for the bundle format, the edge design, and the current limits (Bedrock, CI wiring)
+Another sharp edge, same root: record and replay derive every per-test token deterministically (the model name included, so a replay regenerates the exact requests the record run sent), which means an edge-wired deployment left in the database by an interrupted earlier run carries the same model name as the fresh one the current run registers. The proxy then holds two deployments under one model group and load-balances across both, and because the leftover's `api_base` points at the earlier run's edge process, which is gone, the calls that land on it fail with a connection error that reads like a transport bug rather than the stale row it is. Give each record or replay run a fresh database, or let a run finish so its own teardown deletes what it registered, and never reuse one long-lived proxy across back-to-back record/replay sessions. CI hands every job its own empty database and its own proxy, so it never sees this
+
+Replay answers any provider call that drifted from the recording with an HTTP 599 whose body names the computed and closest recorded keys, so the test fails loudly instead of silently going live, and a bundle older than seven days fails at collection time naming its age; either way the fix is to re-record. Only tests that register edge-wired deployments participate: everything else hits its provider live in every mode, so record exactly the suite you replay. If the proxy runs in a container, set `E2E_PROVIDER_EDGE_ADVERTISE_HOST` (e.g. `host.docker.internal`) so the api_base the proxy stores can reach the edge on the pytest host, and `E2E_PROVIDER_EDGE_BIND_HOST=0.0.0.0` so the edge accepts it. The suites wired to the edge today are `quota_management/spend_tracking/test_provider_edge_spend_e2e.py`, `llm_translation/test_chat_completions_contract_e2e.py`, the OpenAI registrations in `llm_translation/test_embeddings_endpoint_e2e.py`, the Anthropic tests in `llm_translation/test_messages_e2e.py`, streamed and not, and the OpenAI batch deployment behind `batches/`. A streamed response replays as the chunk sequence the provider sent rather than one buffered body. See `CLAUDE.md` in this directory for the bundle format, the edge design, and the current limits (Bedrock). The scheduled CI record/replay lane is described above
 
 Tests marked `@pytest.mark.e2e` hard-fail when no proxy answers `/health/liveliness`, so a run that goes red with `No live proxy` at setup means the proxy isn't up; they never skip for a missing proxy, so an absent proxy can't be mistaken for a pass
 

tests/e2e/conftest.py

@@ -43,6 +43,11 @@ def pytest_configure(config: pytest.Config) -> None:
         "markers",
         "covers(cell_id, *, exercised_on=()): coverage-registry cell(s) this test covers",
     )
+    config.addinivalue_line(
+        "markers",
+        "replayable: edge-wired test whose provider traffic replays from a fixture bundle, so it makes "
+        "zero provider calls in replay mode; the record/replay CI lane selects it with -m replayable",
+    )
     config.addinivalue_line(
         "markers",
         "load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites",

tests/e2e/gateway/record_replay_ci_config.yml

@@ -0,0 +1,3 @@
+general_settings:
+  master_key: os.environ/LITELLM_MASTER_KEY
+  store_model_in_db: true

tests/e2e/llm_translation/test_chat_completions_contract_e2e.py

@@ -13,7 +13,7 @@ from models import ChatBody, ChatMessage, ChatResponse, LiteLLMParamsBody
 from proxy_client import ProxyClient
 from pydantic import BaseModel
 
-pytestmark = pytest.mark.e2e
+pytestmark = [pytest.mark.e2e, pytest.mark.replayable]
 
 OPENAI_BACKEND = "openai/gpt-4o-mini"
 CHAT_PATH = "/chat/completions"

tests/e2e/llm_translation/test_embeddings_endpoint_e2e.py

@@ -40,6 +40,7 @@ def _openai_embeddings_params() -> LiteLLMParamsBody:
 
 
 class TestEmbeddingsEndpoint:
+    @pytest.mark.replayable
     @pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works")
     def test_embeddings_returns_vector(
         self, endpoints_client: EndpointsClient, resources: ResourceManager
@@ -109,6 +110,7 @@ class TestEmbeddingsEndpoint:
             f"embedding vector is all zeros: {result.body[:300]}"
         )
 
+    @pytest.mark.replayable
     @pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works")
     def test_array_input_returns_vectors(
         self, endpoints_client: EndpointsClient, resources: ResourceManager
@@ -129,6 +131,7 @@ class TestEmbeddingsEndpoint:
         parsed = EmbeddingsResult.model_validate_json(result.body)
         assert len(parsed.data) == 3, f"expected 3 vectors: {result.body[:300]}"
 
+    @pytest.mark.replayable
     @pytest.mark.covers("llm.embeddings.openai.input_validation.nonstream.works")
     def test_missing_model_returns_client_error(
         self, endpoints_client: EndpointsClient, resources: ResourceManager
@@ -141,6 +144,7 @@ class TestEmbeddingsEndpoint:
         )
         assert_client_error(result, "embeddings missing model")
 
+    @pytest.mark.replayable
     @pytest.mark.covers("llm.embeddings.openai.input_validation.nonstream.works")
     def test_missing_input_returns_error(
         self, endpoints_client: EndpointsClient, resources: ResourceManager

tests/e2e/llm_translation/test_messages_e2e.py

@@ -24,7 +24,7 @@ from models import (
 )
 from pydantic import BaseModel
 
-pytestmark = pytest.mark.e2e
+pytestmark = [pytest.mark.e2e, pytest.mark.replayable]
 
 
 class _OptionalMessagesBody(BaseModel):
@@ -171,7 +171,7 @@ class TestAnthropicMessages:
                 model=model,
                 max_tokens=64,
                 stream=True,
-                messages=[ChatMessage(role="user", content="Count from one to three.")],
+                messages=[ChatMessage(role="user", content="Count from 1 to 20, one number per line.")],
             ),
         )
         require_successful_call(result)

tests/e2e/pytest.ini

@@ -5,6 +5,7 @@
 addopts = --strict-markers --strict-config --reruns 1 --only-rerun "kind='network'" --only-rerun "status_code=5[0-9][0-9]"
 markers =
     e2e: live test that requires a running proxy and real provider keys
+    replayable: edge-wired test whose provider traffic replays from a fixture bundle, so it makes zero provider calls in replay mode; the record/replay CI lane selects it with -m replayable
     load: heavy throughput/load test; collected last so it never perturbs latency-sensitive suites
     weekly: real-provider anomaly load test that spends real money; deselected unless E2E_WEEKLY_ANOMALY is set
     managed_files: needs a proxy running with require_managed_files enabled; deselected unless E2E_MANAGED_FILES_STACK is set

tests/e2e/quota_management/spend_tracking/test_provider_edge_spend_e2e.py

@@ -18,7 +18,7 @@ from lifecycle import ResourceManager
 from models import LiteLLMParamsBody
 from spend_e2e_client import SpendClient, unique_marker, unwrap
 
-pytestmark = pytest.mark.e2e
+pytestmark = [pytest.mark.e2e, pytest.mark.replayable]
 
 
 @pytest.mark.covers("quota_management.spend_tracking.chat_completions.logs_cost")

tests/test_litellm/test_e2e_egress_sentinel.py

@@ -0,0 +1,172 @@
+"""Tests for .github/scripts/e2e_egress_sentinel.py.
+
+The replay lane's zero-egress proof is only as good as this sentinel: it pins the
+provider hosts to a local sink and counts every connection that reaches them, so
+a single escaped provider call turns the run red. The contract locked in here is
+that the counter counts (each accepted connection is exactly one recorded hit),
+that ``assert-empty`` is the pass/fail gate around that count, and that the hosts
+file it edits is always handed back exactly as it was found.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import socket
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+from typing import Final
+
+_REPO_ROOT: Final = Path(__file__).resolve().parents[2]
+_MODULE_PATH: Final = _REPO_ROOT / ".github" / "scripts" / "e2e_egress_sentinel.py"
+_spec: Final = importlib.util.spec_from_file_location("e2e_egress_sentinel", _MODULE_PATH)
+sentinel: Final = importlib.util.module_from_spec(_spec)
+sys.modules[_spec.name] = sentinel  # @dataclass(slots=True) rebuilds via sys.modules
+_spec.loader.exec_module(sentinel)
+
+
+def _hit_lines(hits_file: Path) -> list[str]:
+    if not hits_file.exists():
+        return []
+    return [line for line in hits_file.read_text(encoding="utf-8").splitlines() if line.strip()]
+
+
+def test_pin_block_lists_every_host_against_the_sink():
+    block = sentinel._pin_block("127.0.0.1", ("api.openai.com", "api.anthropic.com"))
+    assert "127.0.0.1\tapi.openai.com" in block
+    assert "127.0.0.1\tapi.anthropic.com" in block
+    assert sentinel._BLOCK_BEGIN in block and sentinel._BLOCK_END in block
+
+
+def test_install_and_restore_round_trips_an_existing_hosts_file(tmp_path):
+    hosts = tmp_path / "hosts"
+    original = "127.0.0.1\tlocalhost\n255.255.255.255\tbroadcasthost\n"
+    hosts.write_text(original, encoding="utf-8")
+
+    saved = sentinel._install_pins(hosts, "127.0.0.1", ("api.openai.com",))
+    assert saved == original.encode()
+    assert "api.openai.com" in hosts.read_text(encoding="utf-8")
+
+    sentinel._restore_pins(hosts, saved)
+    assert hosts.read_text(encoding="utf-8") == original
+
+
+def test_install_on_a_missing_hosts_file_creates_then_restore_empties(tmp_path):
+    hosts = tmp_path / "hosts"
+    saved = sentinel._install_pins(hosts, "127.0.0.1", ("api.anthropic.com",))
+    assert saved == b""
+    assert "api.anthropic.com" in hosts.read_text(encoding="utf-8")
+
+    sentinel._restore_pins(hosts, saved)
+    assert hosts.read_text(encoding="utf-8") == ""
+
+
+def test_assert_empty_passes_when_no_calls(tmp_path):
+    absent = tmp_path / "absent.jsonl"
+    assert sentinel.assert_empty(absent) == 0
+
+    empty = tmp_path / "empty.jsonl"
+    empty.write_text("\n  \n", encoding="utf-8")
+    assert sentinel.assert_empty(empty) == 0
+
+
+def test_assert_empty_fails_when_calls_recorded(tmp_path):
+    hits = tmp_path / "hits.jsonl"
+    hits.write_text('{"port": 443, "peer": ["127.0.0.1", 5]}\n', encoding="utf-8")
+    assert sentinel.assert_empty(hits) == 1
+
+
+def test_accept_loop_records_exactly_one_hit_per_connection(tmp_path):
+    hits_file = tmp_path / "hits.jsonl"
+    hits_file.write_text("", encoding="utf-8")
+    listener = sentinel._bind("127.0.0.1", 0)
+    port = listener.getsockname()[1]
+    stop = threading.Event()
+    hits = sentinel._HitLog(path=hits_file, _lock=threading.Lock())
+    worker = threading.Thread(target=sentinel._serve_socket, args=(listener, port, hits, stop), daemon=True)
+    worker.start()
+
+    try:
+        for _ in range(3):
+            conn = socket.create_connection(("127.0.0.1", port), timeout=2)
+            conn.close()
+        deadline = time.time() + 3
+        while time.time() < deadline and len(_hit_lines(hits_file)) < 3:
+            time.sleep(0.02)
+    finally:
+        stop.set()
+        listener.close()
+        worker.join(timeout=3)
+
+    assert len(_hit_lines(hits_file)) == 3
+
+
+def test_serve_end_to_end_pins_counts_and_restores(tmp_path):
+    hosts = tmp_path / "hosts"
+    hosts.write_text("127.0.0.1\tlocalhost\n", encoding="utf-8")
+    hits = tmp_path / "hits.jsonl"
+    ready = tmp_path / "ready"
+    pidf = tmp_path / "pid"
+    port = _free_port()
+
+    proc = subprocess.Popen(
+        [
+            sys.executable,
+            str(_MODULE_PATH),
+            "serve",
+            "--host",
+            "api.openai.com",
+            "--host",
+            "api.anthropic.com",
+            "--sink-address",
+            "127.0.0.1",
+            "--port",
+            str(port),
+            "--hits-file",
+            str(hits),
+            "--ready-file",
+            str(ready),
+            "--pid-file",
+            str(pidf),
+            "--hosts-file",
+            str(hosts),
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        _wait_for(lambda: ready.exists(), timeout=10)
+        pinned = hosts.read_text(encoding="utf-8")
+        assert "api.openai.com" in pinned and "api.anthropic.com" in pinned
+
+        for _ in range(2):
+            conn = socket.create_connection(("127.0.0.1", port), timeout=2)
+            conn.close()
+        _wait_for(lambda: len(_hit_lines(hits)) >= 2, timeout=5)
+        assert sentinel.assert_empty(hits) == 1
+    finally:
+        proc.terminate()
+        proc.wait(timeout=10)
+
+    assert hosts.read_text(encoding="utf-8") == "127.0.0.1\tlocalhost\n"
+    assert not ready.exists()
+
+
+def _free_port() -> int:
+    probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    probe.bind(("127.0.0.1", 0))
+    port = probe.getsockname()[1]
+    probe.close()
+    return port
+
+
+def _wait_for(predicate, *, timeout: float) -> None:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if predicate():
+            return
+        time.sleep(0.05)
+    raise AssertionError("condition never became true within the timeout")