diff --git a/.github/merge-smoke-tests.json b/.github/merge-smoke-tests.json
new file mode 100644
index 00000000000..6088953b7eb
--- /dev/null
+++ b/.github/merge-smoke-tests.json
@@ -0,0 +1,15 @@
+{
+  "cases": {
+    "CHAT-JSON": "tests/test_litellm/llms/openai/test_openai.py::test_acompletion_returns_json_reply_over_injected_transport",
+    "CHAT-TEXT-STREAM": "tests/test_litellm/llms/openai/test_openai.py::test_acompletion_streams_text_deltas_over_injected_transport",
+    "CHAT-TOOL-STREAM": "tests/test_litellm/llms/openai/test_openai.py::test_acompletion_streams_tool_call_arguments_over_injected_transport",
+    "MODEL-ALLOW": "tests/test_litellm/proxy/auth/test_auth_checks.py::test_can_object_call_model_allows_listed_model_for_key",
+    "MODEL-DENY": "tests/test_litellm/proxy/auth/test_auth_checks.py::test_can_object_call_model_denials_return_forbidden[key-key_model_access_denied]",
+    "COST-EXPLICIT": "tests/test_litellm/test_cost_calculator.py::test_completion_cost_charges_explicit_per_token_rates_over_registered_ones",
+    "COST-ZERO": "tests/test_litellm/test_cost_calculator.py::test_completion_cost_is_zero_when_explicit_rates_are_zero",
+    "LOG-CONTENT-ON": "tests/test_litellm/litellm_core_utils/test_litellm_logging.py::test_standard_logging_payload_keeps_message_content_when_message_logging_is_on",
+    "LOG-CONTENT-OFF": "tests/test_litellm/litellm_core_utils/test_litellm_logging.py::test_standard_logging_payload_redacts_message_content_when_message_logging_is_off",
+    "CALLBACK-SUCCESS": "tests/test_litellm/litellm_core_utils/test_litellm_logging.py::test_async_success_handler_delivers_standard_logging_payload_to_custom_logger",
+    "CALLBACK-FAILURE": "tests/test_litellm/litellm_core_utils/test_litellm_logging.py::test_async_failure_handler_delivers_failure_payload_to_custom_logger"
+  }
+}
diff --git a/.github/scripts/run_merge_smoke.py b/.github/scripts/run_merge_smoke.py
new file mode 100644
index 00000000000..7e74de324fc
--- /dev/null
+++ b/.github/scripts/run_merge_smoke.py
@@ -0,0 +1,493 @@
+#!/usr/bin/env python3
+"""Merge smoke harness: bounded checks run inside a loopback-only Linux network namespace."""
+
+# ruff: noqa: T201  # CLI harness: stdout/stderr lines are the reported result
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import http.client
+import json
+import os
+import secrets
+import signal
+import socket
+import subprocess
+import sys
+import time
+from collections import Counter
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from pathlib import Path
+from types import MappingProxyType
+from typing import Final, NoReturn, TextIO, cast
+
+import pytest
+
+EXPECTED_CASES: Final = (
+    "CHAT-JSON",
+    "CHAT-TEXT-STREAM",
+    "CHAT-TOOL-STREAM",
+    "MODEL-ALLOW",
+    "MODEL-DENY",
+    "COST-EXPLICIT",
+    "COST-ZERO",
+    "LOG-CONTENT-ON",
+    "LOG-CONTENT-OFF",
+    "CALLBACK-SUCCESS",
+    "CALLBACK-FAILURE",
+)
+
+
+@dataclass(frozen=True, slots=True)
+class CheckResult:
+    ok: bool
+    detail: str = ""
+
+
+@dataclass(slots=True)
+class _Args:
+    command: str = ""
+    no_child: bool = False
+    expect: str = ""
+    litellm_bin: str | None = None
+    lite_bin: str | None = None
+    diagnostics_dir: str = ""
+    ready_deadline: float = 120.0
+    shutdown_deadline: float = 20.0
+    poll_interval: float = 0.5
+    manifest: str = ""
+    rootdir: str | None = None
+
+
+def fail(reason: str) -> NoReturn:
+    print(f"merge-smoke: FAIL {reason}", file=sys.stderr)
+    sys.exit(1)
+
+
+def ok(step: str) -> None:
+    print(f"merge-smoke: OK {step}")
+
+
+def tail(path: Path, lines: int = 20) -> str:
+    try:
+        return "\n".join(path.read_text(errors="replace").splitlines()[-lines:])
+    except OSError as exc:
+        return f"<cannot read {path}: {exc}>"
+
+
+def cmd_verify_isolation(args: _Args) -> int:
+    if os.geteuid() == 0:
+        fail("verify-isolation must run unprivileged (geteuid()==0)")
+    try:
+        socket.create_connection(("192.0.2.1", 9), timeout=3)
+    except OSError as exc:
+        print(f"external connect blocked as expected: errno={exc.errno} {exc}")
+    else:
+        fail("external TCP connect to 192.0.2.1:9 succeeded; namespace is not isolated")
+    listener: Final = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.bind(("127.0.0.1", 0))
+    listener.listen(1)
+    port: Final = cast(int, listener.getsockname()[1])
+    client: Final = socket.create_connection(("127.0.0.1", port), timeout=5)
+    accepted: Final = listener.accept()
+    accepted[0].close()
+    client.close()
+    listener.close()
+    print(f"loopback connect ok on 127.0.0.1:{port}")
+    if not args.no_child:
+        proc: Final = subprocess.run(
+            [sys.executable, str(Path(__file__).resolve()), "verify-isolation", "--no-child"],
+            timeout=30,
+            capture_output=True,
+            text=True,
+        )
+        if proc.returncode != 0:
+            fail(f"child process did not inherit isolation: {proc.stderr.strip()}")
+        print("child process inherits isolation")
+    ok("verify-isolation")
+    return 0
+
+
+def cmd_interpreter(args: _Args) -> int:
+    print(sys.version)
+    print(sys.executable)
+    actual: Final = f"{sys.version_info.major}.{sys.version_info.minor}"
+    if actual != args.expect:
+        fail(f"interpreter is {actual}, expected {args.expect}")
+    ok(f"interpreter {actual}")
+    return 0
+
+
+def _run_cli(argv: Sequence[str], label: str) -> CheckResult:
+    try:
+        proc: Final = subprocess.run(list(argv), timeout=120, capture_output=True, text=True)
+    except subprocess.TimeoutExpired:
+        return CheckResult(ok=False, detail=f"{label} timed out after 120s")
+    sys.stdout.write(proc.stdout)
+    sys.stderr.write(proc.stderr)
+    if proc.returncode != 0:
+        return CheckResult(ok=False, detail=f"{label} exited {proc.returncode}")
+    return CheckResult(ok=True)
+
+
+def cmd_cli(args: _Args) -> int:
+    venv_bin: Final = Path(sys.executable).parent
+    litellm_bin: Final = Path(args.litellm_bin) if args.litellm_bin else venv_bin / "litellm"
+    lite_bin: Final = Path(args.lite_bin) if args.lite_bin else venv_bin / "lite"
+    commands: Final = (
+        ("import litellm", [sys.executable, "-c", "import litellm"]),
+        ("litellm --version", [str(litellm_bin), "--version"]),
+        ("lite version", [str(lite_bin), "version"]),
+    )
+    for label, argv in commands:
+        result = _run_cli(argv, label)
+        if not result.ok:
+            fail(result.detail)
+        ok(label)
+    return 0
+
+
+def _free_port() -> int:
+    sock: Final = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.bind(("127.0.0.1", 0))
+    port: Final = cast(int, sock.getsockname()[1])
+    sock.close()
+    return port
+
+
+_CONFIG_TEMPLATE: Final = """model_list:
+  - model_name: smoke-model
+    litellm_params:
+      model: openai/smoke-model
+      api_base: http://127.0.0.1:9/v1
+      api_key: synthetic-key
+general_settings:
+  master_key: os.environ/LITELLM_MASTER_KEY
+"""
+
+
+def _listen_inode(port: int) -> str | None:
+    target: Final = f"{port:04X}"
+    for table in ("/proc/net/tcp", "/proc/net/tcp6"):
+        try:
+            rows = Path(table).read_text().splitlines()[1:]
+        except OSError:
+            continue
+        for row in rows:
+            cols = row.split()
+            if len(cols) > 9 and cols[3] == "0A" and cols[1].rsplit(":", 1)[-1] == target:
+                return cols[9]
+    return None
+
+
+def _ancestors(pid: int) -> frozenset[int]:
+    chain: Final[set[int]] = set()
+    pending: Final[list[int]] = [pid]
+    while pending:
+        current = pending.pop()
+        if current <= 0 or current in chain:
+            continue
+        chain.add(current)
+        try:
+            stat = Path(f"/proc/{current}/stat").read_text()
+        except OSError:
+            continue
+        pending.append(int(stat.rpartition(")")[2].split()[1]))
+    return frozenset(chain)
+
+
+def _socket_owner_pid(inode: str) -> int | None:
+    for proc_dir in Path("/proc").iterdir():
+        if not proc_dir.name.isdigit():
+            continue
+        fd_dir = proc_dir / "fd"
+        try:
+            for fd in fd_dir.iterdir():
+                try:
+                    if os.readlink(fd) == f"socket:[{inode}]":
+                        return int(proc_dir.name)
+                except OSError:
+                    continue
+        except OSError:
+            continue
+    return None
+
+
+def _verify_port_owner(port: int, proc: subprocess.Popen[bytes]) -> CheckResult:
+    inode: Final = _listen_inode(port)
+    if inode is None:
+        return CheckResult(ok=False, detail=f"no LISTEN socket found for port {port} in /proc/net/tcp")
+    owner: Final = _socket_owner_pid(inode)
+    if owner is None:
+        return CheckResult(ok=False, detail=f"no process owns the listen socket inode {inode} for port {port}")
+    if owner != proc.pid and proc.pid not in _ancestors(owner):
+        return CheckResult(
+            ok=False, detail=f"port {port} owned by pid {owner} outside the launched process group {proc.pid}"
+        )
+    if proc.poll() is not None:
+        return CheckResult(ok=False, detail=f"proxy exited with code {proc.returncode} after readiness")
+    return CheckResult(ok=True)
+
+
+def cmd_proxy_startup(args: _Args) -> int:
+    diagnostics: Final = Path(args.diagnostics_dir)
+    diagnostics.mkdir(parents=True, exist_ok=True)
+    venv_bin: Final = Path(sys.executable).parent
+    litellm_bin: Final = Path(args.litellm_bin) if args.litellm_bin else venv_bin / "litellm"
+    port: Final = _free_port()
+    master_key: Final = "sk-smoke-" + secrets.token_hex(16)
+    config_path: Final = diagnostics / "config.yaml"
+    config_path.write_text(_CONFIG_TEMPLATE)
+    log_path: Final = diagnostics / "proxy.log"
+    result_path: Final = diagnostics / "result.json"
+    outcome: Final[dict[str, object]] = {
+        "port": port,
+        "time_to_ready_s": None,
+        "shutdown_s": None,
+        "readiness": None,
+        "outcome": "failed",
+    }
+    log_file: Final = log_path.open("w")
+    env: Final = {
+        **os.environ,
+        "LITELLM_MASTER_KEY": master_key,
+        "LITELLM_LOCAL_MODEL_COST_MAP": "True",
+    }
+    started: Final = time.monotonic()
+    proc: Final = subprocess.Popen(
+        [str(litellm_bin), "--config", str(config_path), "--host", "127.0.0.1", "--port", str(port)],
+        stdout=log_file,
+        stderr=subprocess.STDOUT,
+        start_new_session=True,
+        env=env,
+    )
+    body: str | None = None
+    last_status: int | None = None
+    while time.monotonic() - started < args.ready_deadline:
+        if proc.poll() is not None:
+            log_file.close()
+            result_path.write_text(json.dumps(outcome))
+            fail(f"proxy exited early with code {proc.returncode}\n{tail(log_path)}")
+        try:
+            conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
+            conn.request("GET", "/health/readiness")
+            resp = conn.getresponse()
+            last_status = resp.status
+            candidate = resp.read().decode()
+            conn.close()
+        except (http.client.HTTPException, ConnectionError, OSError):
+            time.sleep(args.poll_interval)
+            continue
+        if last_status == 200:
+            body = candidate
+            break
+        time.sleep(args.poll_interval)
+    outcome["time_to_ready_s"] = round(time.monotonic() - started, 3)
+    if body is None:
+        _terminate(proc, log_file)
+        result_path.write_text(json.dumps(outcome))
+        detail = f"last status {last_status}" if last_status is not None else "no response"
+        fail(f"readiness not reached within {args.ready_deadline}s ({detail})\n{tail(log_path)}")
+    outcome["readiness"] = body
+    try:
+        readiness = cast(object, json.loads(body))
+    except json.JSONDecodeError:
+        readiness = None
+    if readiness != {"status": "healthy", "db": "Not connected"}:
+        _terminate(proc, log_file)
+        result_path.write_text(json.dumps(outcome))
+        fail(f"unexpected readiness body: {body}")
+    owner_check: Final = _verify_port_owner(port, proc)
+    if not owner_check.ok:
+        _terminate(proc, log_file)
+        result_path.write_text(json.dumps(outcome))
+        fail(owner_check.detail)
+    shutdown_started: Final = time.monotonic()
+    os.killpg(proc.pid, signal.SIGTERM)
+    try:
+        proc.wait(timeout=args.shutdown_deadline)
+    except subprocess.TimeoutExpired:
+        os.killpg(proc.pid, signal.SIGKILL)
+        proc.wait(timeout=10)
+        outcome["shutdown_s"] = round(time.monotonic() - shutdown_started, 3)
+        log_file.close()
+        result_path.write_text(json.dumps(outcome))
+        fail(f"forced kill after {args.shutdown_deadline}s\n{tail(log_path)}")
+    outcome["shutdown_s"] = round(time.monotonic() - shutdown_started, 3)
+    try:
+        os.killpg(proc.pid, 0)
+    except ProcessLookupError:
+        pass
+    else:
+        os.killpg(proc.pid, signal.SIGKILL)
+        log_file.close()
+        result_path.write_text(json.dumps(outcome))
+        fail("process group survived SIGTERM")
+    log_file.close()
+    outcome["outcome"] = "ok"
+    result_path.write_text(json.dumps(outcome))
+    ok(f"proxy-startup ready={outcome['time_to_ready_s']}s shutdown={outcome['shutdown_s']}s")
+    return 0
+
+
+def _terminate(proc: subprocess.Popen[bytes], log_file: TextIO) -> None:
+    with contextlib.suppress(ProcessLookupError):
+        os.killpg(proc.pid, signal.SIGTERM)
+    try:
+        proc.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        with contextlib.suppress(ProcessLookupError):
+            os.killpg(proc.pid, signal.SIGKILL)
+        with contextlib.suppress(subprocess.TimeoutExpired):
+            proc.wait(timeout=10)
+    log_file.close()
+
+
+def _load_manifest(path: Path) -> MappingProxyType[str, str]:
+    def no_duplicates(pairs: list[tuple[object, object]]) -> dict[object, object]:
+        seen: dict[object, object] = {}
+        for key, value in pairs:
+            if key in seen:
+                raise ValueError(f"duplicate key in manifest: {key}")
+            seen[key] = value
+        return seen
+
+    raw_value: object = cast(object, json.loads(path.read_text(), object_pairs_hook=no_duplicates))
+    if not isinstance(raw_value, dict):
+        raise ValueError("manifest must be an object")
+    loaded: Final = cast(dict[object, object], raw_value)
+    cases_value: object = loaded.get("cases")
+    if not isinstance(cases_value, dict):
+        raise ValueError("manifest must be an object with a 'cases' object")
+    cases_any: Final = cast(dict[object, object], cases_value)
+    cases: Final = {k: v for k, v in cases_any.items() if isinstance(k, str) and isinstance(v, str)}
+    if len(cases) != len(cases_any):
+        raise ValueError("manifest 'cases' must map string ids to string node ids")
+    return MappingProxyType(cases)
+
+
+@dataclass(slots=True, eq=False)
+class _Recorder:
+    collect_failed: list[str] = field(default_factory=list)
+    collected: tuple[str, ...] = ()
+    reports: dict[str, list[tuple[str, str, bool]]] = field(default_factory=dict)
+
+    def pytest_collectreport(self, report: pytest.CollectReport) -> None:
+        if report.failed:
+            self.collect_failed.append(report.nodeid)
+
+    def pytest_collection_finish(self, session: pytest.Session) -> None:
+        self.collected = tuple(item.nodeid for item in session.items)
+
+    def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
+        self.reports.setdefault(report.nodeid, []).append((report.when, report.outcome, hasattr(report, "wasxfail")))
+
+
+def cmd_pytest(args: _Args) -> int:
+    try:
+        cases: Final = _load_manifest(Path(args.manifest))
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        fail(f"manifest invalid: {exc}")
+    if tuple(cases) != EXPECTED_CASES:
+        fail(f"manifest case ids must be exactly {list(EXPECTED_CASES)} in order, got {list(cases)}")
+    node_ids: Final = tuple(cases.values())
+    if len(set(node_ids)) != len(node_ids):
+        fail("manifest node ids are not unique")
+    argv: Final = [
+        *node_ids,
+        "-p",
+        "no:cacheprovider",
+        "-p",
+        "no:xdist",
+        "-p",
+        "no:rerunfailures",
+        "-p",
+        "no:randomly",
+        "-rA",
+        "-q",
+        *(["--rootdir", args.rootdir] if args.rootdir else []),
+    ]
+
+    recorder: Final = _Recorder()
+    code: Final = pytest.main(argv, plugins=[recorder])
+    name_of: Final = MappingProxyType({node_id: case_id for case_id, node_id in cases.items()})
+    problems: Final[list[str]] = []
+    if code != 0:
+        problems.append(f"pytest exit code {code}")
+    for failed_id in recorder.collect_failed:
+        problems.append(f"collection failed: {name_of.get(failed_id, failed_id)}")
+    expected: Final = Counter(node_ids)
+    collected: Final = Counter(recorder.collected)
+    for node_id in expected - collected:
+        problems.append(f"missing case {name_of[node_id]} ({node_id})")
+    for node_id in collected - expected:
+        problems.append(f"unexpected test collected: {node_id}")
+    for node_id, count in collected.items():
+        if count > 1:
+            problems.append(f"duplicated test id: {node_id}")
+    if len(recorder.collected) != len(EXPECTED_CASES):
+        problems.append(f"collected {len(recorder.collected)} tests, expected {len(EXPECTED_CASES)}")
+    rows: Final[list[tuple[str, bool]]] = []
+    for case_id, node_id in cases.items():
+        reports = recorder.reports.get(node_id, [])
+        case_ok = (
+            bool(reports)
+            and all(outcome == "passed" and not wasxfail for _, outcome, wasxfail in reports)
+            and {when for when, _, _ in reports} >= {"setup", "call", "teardown"}
+        )
+        rows.append((case_id, case_ok))
+        if not reports:
+            problems.append(f"{case_id} ({node_id}) produced no runtest reports")
+            continue
+        for when, outcome, wasxfail in reports:
+            if outcome != "passed":
+                problems.append(f"{case_id} ({node_id}) {when} outcome={outcome}")
+            if wasxfail:
+                problems.append(f"{case_id} ({node_id}) {when} was xfail/xpass")
+        missing_phases = {"setup", "call", "teardown"} - {when for when, _, _ in reports}
+        for phase in sorted(missing_phases):
+            problems.append(f"{case_id} ({node_id}) missing {phase} report")
+    for case_id, passed in rows:
+        print(f"{case_id}  {'PASS' if passed else 'FAIL'}  {cases[case_id]}")
+    if problems:
+        for problem in problems:
+            print(f"merge-smoke: {problem}", file=sys.stderr)
+        fail("pytest verdict failed")
+    ok("pytest 11 cases")
+    return 0
+
+
+def main() -> int:
+    parser: Final = argparse.ArgumentParser(description=__doc__)
+    subs: Final = parser.add_subparsers(dest="command", required=True)
+    p_iso: Final = subs.add_parser("verify-isolation")
+    p_iso.add_argument("--no-child", action="store_true")
+    p_interp: Final = subs.add_parser("interpreter")
+    p_interp.add_argument("--expect", required=True)
+    p_cli: Final = subs.add_parser("cli")
+    p_cli.add_argument("--litellm-bin", default=None)
+    p_cli.add_argument("--lite-bin", default=None)
+    p_proxy: Final = subs.add_parser("proxy-startup")
+    p_proxy.add_argument("--diagnostics-dir", required=True)
+    p_proxy.add_argument("--litellm-bin", default=None)
+    p_proxy.add_argument("--ready-deadline", type=float, default=120)
+    p_proxy.add_argument("--shutdown-deadline", type=float, default=20)
+    p_proxy.add_argument("--poll-interval", type=float, default=0.5)
+    p_test: Final = subs.add_parser("pytest")
+    p_test.add_argument("--manifest", required=True)
+    p_test.add_argument("--rootdir", default=None)
+    args: Final = parser.parse_args(namespace=_Args())
+    handlers: Final = {
+        "verify-isolation": cmd_verify_isolation,
+        "interpreter": cmd_interpreter,
+        "cli": cmd_cli,
+        "proxy-startup": cmd_proxy_startup,
+        "pytest": cmd_pytest,
+    }
+    return handlers[args.command](args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/.github/workflows/test-code-quality.yml b/.github/workflows/test-code-quality.yml
index 2b6adaa6af6..75f645086fb 100644
--- a/.github/workflows/test-code-quality.yml
+++ b/.github/workflows/test-code-quality.yml
@@ -80,6 +80,9 @@ jobs:
       - name: test_e2e_changed_gate
         run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py tests/code_coverage_tests/test_e2e_idp_stack.py
 
+      - name: Check merge smoke harness
+        run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_merge_smoke.py
+
       - name: router_code_coverage
         run: uv run --no-sync python ./tests/code_coverage_tests/router_code_coverage.py
 
diff --git a/.github/workflows/test-merge-smoke.yml b/.github/workflows/test-merge-smoke.yml
new file mode 100644
index 00000000000..910763c6af2
--- /dev/null
+++ b/.github/workflows/test-merge-smoke.yml
@@ -0,0 +1,95 @@
+name: Merge smoke checks
+
+on:
+  pull_request:
+    branches: [main, litellm_internal_staging, litellm_oss_staging, "litellm_**"]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: merge-smoke-${{ github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  dashboard-build:
+    name: Dashboard build
+    runs-on: ubuntu-24.04
+    timeout-minutes: 30
+    steps:
+      - name: Checkout
+        uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - name: Build the dashboard stage
+        run: docker build --target ui-builder -f Dockerfile .
+
+  core-checks:
+    name: Core checks (Python ${{ matrix.python-version }})
+    runs-on: ubuntu-24.04
+    timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
+    env:
+      LITELLM_LOCAL_MODEL_COST_MAP: "True"
+    steps:
+      - name: Checkout
+        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: ${{ matrix.python-version }}
+
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Install dependencies
+        run: .github/scripts/uv_sync_with_retries.sh --frozen --extra proxy --extra cli --group dev --group proxy-dev --python ${{ matrix.python-version }}
+
+      - name: Create the loopback-only network namespace
+        run: |
+          sudo ip netns add smoke
+          sudo ip netns exec smoke ip link set lo up
+          cat > "${RUNNER_TEMP}/in-netns" <<'WRAP'
+          #!/usr/bin/env bash
+          set -euo pipefail
+          exec sudo --preserve-env=LITELLM_LOCAL_MODEL_COST_MAP ip netns exec smoke setpriv --reuid "$(id -u)" --regid "$(id -g)" --init-groups -- env HOME="${HOME}" PATH="${PATH}" "$@"
+          WRAP
+          chmod +x "${RUNNER_TEMP}/in-netns"
+          echo "IN_NETNS=${RUNNER_TEMP}/in-netns" >> "${GITHUB_ENV}"
+
+      - name: Verify namespace isolation
+        run: $IN_NETNS .venv/bin/python .github/scripts/run_merge_smoke.py verify-isolation
+
+      - name: Verify interpreter version
+        run: $IN_NETNS .venv/bin/python .github/scripts/run_merge_smoke.py interpreter --expect ${{ matrix.python-version }}
+
+      - name: Import and CLI checks
+        run: $IN_NETNS .venv/bin/python .github/scripts/run_merge_smoke.py cli
+
+      - name: Proxy startup check
+        run: $IN_NETNS .venv/bin/python .github/scripts/run_merge_smoke.py proxy-startup --diagnostics-dir "${RUNNER_TEMP}/smoke-diagnostics"
+
+      - name: Run curated smoke cases
+        run: $IN_NETNS .venv/bin/python .github/scripts/run_merge_smoke.py pytest --manifest .github/merge-smoke-tests.json
+
+      - name: Upload smoke diagnostics
+        if: always()
+        uses: actions/upload-artifact@4cec3d8aa04e39d1a68397de0c4cd6fb9dce8ec1 # v4.6.1
+        with:
+          name: merge-smoke-diagnostics-py${{ matrix.python-version }}
+          path: ${{ runner.temp }}/smoke-diagnostics
+          if-no-files-found: ignore
+
+      - name: Remove the network namespace
+        if: always()
+        run: sudo ip netns delete smoke
diff --git a/tests/code_coverage_tests/test_merge_smoke.py b/tests/code_coverage_tests/test_merge_smoke.py
new file mode 100644
index 00000000000..e3b96e222b9
--- /dev/null
+++ b/tests/code_coverage_tests/test_merge_smoke.py
@@ -0,0 +1,402 @@
+import json
+import os
+import stat
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+from typing import Final, cast
+
+import pytest
+
+HARNESS: Final = Path(__file__).parents[2] / ".github" / "scripts" / "run_merge_smoke.py"
+
+CASE_IDS: Final = (
+    "CHAT-JSON",
+    "CHAT-TEXT-STREAM",
+    "CHAT-TOOL-STREAM",
+    "MODEL-ALLOW",
+    "MODEL-DENY",
+    "COST-EXPLICIT",
+    "COST-ZERO",
+    "LOG-CONTENT-ON",
+    "LOG-CONTENT-OFF",
+    "CALLBACK-SUCCESS",
+    "CALLBACK-FAILURE",
+)
+
+
+def _write_fake_tests(root: Path, body: str) -> Path:
+    package: Final = root / "fake_tests"
+    package.mkdir()
+    (package / "test_cases.py").write_text(body)
+    return package
+
+
+def _manifest(root: Path, **overrides: str) -> Path:
+    cases: Final[dict[str, str]] = {
+        case_id: f"fake_tests/test_cases.py::test_{case_id.lower().replace('-', '_')}" for case_id in CASE_IDS
+    }
+    cases.update(overrides)
+    path: Final = root / "manifest.json"
+    path.write_text(json.dumps({"cases": cases}))
+    return path
+
+
+def _run(root: Path, *argv: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, "-I", str(HARNESS), *argv],
+        cwd=root,
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+
+
+def _passing_tests() -> str:
+    return "\n".join(f"def test_{case_id.lower().replace('-', '_')}():\n    assert True" for case_id in CASE_IDS)
+
+
+def test_all_eleven_cases_pass(tmp_path: Path) -> None:
+    _write_fake_tests(tmp_path, _passing_tests())
+    manifest: Final = _manifest(tmp_path)
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest), "--rootdir", str(tmp_path))
+
+    assert proc.returncode == 0, proc.stderr
+    assert proc.stdout.count("PASS") >= 11
+    for case_id in CASE_IDS:
+        assert f"{case_id}  PASS" in proc.stdout
+
+
+def test_missing_test_node_id_fails(tmp_path: Path) -> None:
+    _write_fake_tests(tmp_path, _passing_tests())
+    manifest: Final = _manifest(tmp_path, **{"COST-ZERO": "fake_tests/test_cases.py::test_does_not_exist"})
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest), "--rootdir", str(tmp_path))
+
+    assert proc.returncode != 0
+    assert "COST-ZERO" in proc.stderr or "test_does_not_exist" in proc.stderr
+
+
+def test_skipped_case_fails(tmp_path: Path) -> None:
+    _write_fake_tests(
+        tmp_path,
+        _passing_tests().replace(
+            "def test_cost_zero():\n    assert True",
+            "def test_cost_zero():\n    import pytest\n    pytest.skip('nope')",
+        ),
+    )
+    manifest: Final = _manifest(tmp_path)
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest), "--rootdir", str(tmp_path))
+
+    assert proc.returncode != 0
+    assert "COST-ZERO" in proc.stderr
+
+
+def test_xfail_case_fails(tmp_path: Path) -> None:
+    _write_fake_tests(
+        tmp_path,
+        "import pytest\n"
+        + _passing_tests().replace(
+            "def test_cost_zero():\n    assert True",
+            "@pytest.mark.xfail\ndef test_cost_zero():\n    assert False",
+        ),
+    )
+    manifest: Final = _manifest(tmp_path)
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest), "--rootdir", str(tmp_path))
+
+    assert proc.returncode != 0
+    assert "COST-ZERO" in proc.stderr
+
+
+def test_xpass_case_fails(tmp_path: Path) -> None:
+    _write_fake_tests(
+        tmp_path,
+        "import pytest\n"
+        + _passing_tests().replace(
+            "def test_cost_zero():\n    assert True",
+            "@pytest.mark.xfail\ndef test_cost_zero():\n    assert True",
+        ),
+    )
+    manifest: Final = _manifest(tmp_path)
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest), "--rootdir", str(tmp_path))
+
+    assert proc.returncode != 0
+    assert "COST-ZERO" in proc.stderr
+
+
+def test_duplicate_manifest_key_fails(tmp_path: Path) -> None:
+    manifest: Final = tmp_path / "manifest.json"
+    manifest.write_text('{"cases": {"CHAT-JSON": "a::b", "CHAT-JSON": "a::c"}}')
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest))
+
+    assert proc.returncode != 0
+    assert "CHAT-JSON" in proc.stderr
+
+
+def test_missing_case_id_fails(tmp_path: Path) -> None:
+    manifest: Final = tmp_path / "manifest.json"
+    cases: Final = {c: f"t::{c}" for c in CASE_IDS[:-1]}
+    manifest.write_text(json.dumps({"cases": cases}))
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest))
+
+    assert proc.returncode != 0
+    assert "case ids" in proc.stderr
+
+
+def test_extra_case_id_fails(tmp_path: Path) -> None:
+    manifest: Final = tmp_path / "manifest.json"
+    cases: Final = {c: f"t::{c}" for c in CASE_IDS}
+    cases["EXTRA"] = "t::x"
+    manifest.write_text(json.dumps({"cases": cases}))
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest))
+
+    assert proc.returncode != 0
+    assert "case ids" in proc.stderr
+
+
+def test_teardown_error_fails(tmp_path: Path) -> None:
+    body: Final = (
+        "import pytest\n\n@pytest.fixture\ndef boom():\n    yield\n    raise RuntimeError('teardown-boom')\n\n"
+        + _passing_tests().replace(
+            "def test_cost_zero():\n    assert True",
+            "def test_cost_zero(boom):\n    assert True",
+        )
+    )
+    _write_fake_tests(tmp_path, body)
+    manifest: Final = _manifest(tmp_path)
+
+    proc: Final = _run(tmp_path, "pytest", "--manifest", str(manifest), "--rootdir", str(tmp_path))
+
+    assert proc.returncode != 0
+    assert "COST-ZERO" in proc.stderr
+
+
+def _fake_litellm(tmp_path: Path, script: str) -> Path:
+    path: Final = tmp_path / "fake-litellm"
+    path.write_text(f"#!{sys.executable}\n" + textwrap.dedent(script))
+    path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+    return path
+
+
+def test_proxy_startup_exits_early_fails(tmp_path: Path) -> None:
+    fake: Final = _fake_litellm(tmp_path, "import sys\nsys.exit(1)\n")
+    diagnostics: Final = tmp_path / "diag"
+
+    proc: Final = _run(
+        tmp_path,
+        "proxy-startup",
+        "--diagnostics-dir",
+        str(diagnostics),
+        "--litellm-bin",
+        str(fake),
+    )
+
+    assert proc.returncode != 0
+    assert "exited early" in proc.stderr
+    assert (diagnostics / "proxy.log").exists()
+
+
+def test_proxy_startup_readiness_timeout_fails(tmp_path: Path) -> None:
+    fake: Final = _fake_litellm(
+        tmp_path,
+        "import os, pathlib, sys, time\npathlib.Path(sys.argv[0]).with_name('fake.pid').write_text(str(os.getpid()))\ntime.sleep(3600)\n",
+    )
+    diagnostics: Final = tmp_path / "diag"
+
+    proc: Final = _run(
+        tmp_path,
+        "proxy-startup",
+        "--diagnostics-dir",
+        str(diagnostics),
+        "--litellm-bin",
+        str(fake),
+        "--ready-deadline",
+        "3",
+        "--shutdown-deadline",
+        "2",
+    )
+
+    assert proc.returncode != 0
+    assert "readiness" in proc.stderr
+    assert (diagnostics / "proxy.log").exists()
+    with pytest.raises(ProcessLookupError):
+        os.kill(int((tmp_path / "fake.pid").read_text()), 0)
+
+
+def test_proxy_startup_healthy_succeeds(tmp_path: Path) -> None:
+    fake: Final = _fake_litellm(
+        tmp_path,
+        """
+        import http.server, json, sys
+        port = int(sys.argv[sys.argv.index("--port") + 1])
+        class H(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                body = json.dumps({"status": "healthy", "db": "Not connected"}).encode()
+                self.send_response(200)
+                self.send_header("content-type", "application/json")
+                self.end_headers()
+                self.wfile.write(body)
+            def log_message(self, *a):
+                pass
+        http.server.HTTPServer(("127.0.0.1", port), H).serve_forever()
+        """,
+    )
+    diagnostics: Final = tmp_path / "diag"
+
+    proc: Final = _run(
+        tmp_path,
+        "proxy-startup",
+        "--diagnostics-dir",
+        str(diagnostics),
+        "--litellm-bin",
+        str(fake),
+        "--ready-deadline",
+        "15",
+    )
+
+    assert proc.returncode == 0, proc.stderr
+    result: Final = cast(dict[str, object], json.loads((diagnostics / "result.json").read_text()))
+    assert result["outcome"] == "ok"
+    assert result["readiness"] == '{"status": "healthy", "db": "Not connected"}'
+
+
+def test_proxy_startup_sigterm_ignored_forces_kill(tmp_path: Path) -> None:
+    fake: Final = _fake_litellm(
+        tmp_path,
+        """
+        import http.server, json, os, pathlib, signal, sys
+        port = int(sys.argv[sys.argv.index("--port") + 1])
+        pathlib.Path(sys.argv[0]).with_name("fake.pid").write_text(str(os.getpid()))
+        signal.signal(signal.SIGTERM, signal.SIG_IGN)
+        class H(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                body = json.dumps({"status": "healthy", "db": "Not connected"}).encode()
+                self.send_response(200)
+                self.send_header("content-type", "application/json")
+                self.end_headers()
+                self.wfile.write(body)
+            def log_message(self, *a):
+                pass
+        http.server.HTTPServer(("127.0.0.1", port), H).serve_forever()
+        """,
+    )
+    diagnostics: Final = tmp_path / "diag"
+
+    proc: Final = _run(
+        tmp_path,
+        "proxy-startup",
+        "--diagnostics-dir",
+        str(diagnostics),
+        "--litellm-bin",
+        str(fake),
+        "--ready-deadline",
+        "15",
+        "--shutdown-deadline",
+        "2",
+    )
+
+    assert proc.returncode != 0
+    assert "forced kill" in proc.stderr
+    result: Final = cast(dict[str, object], json.loads((diagnostics / "result.json").read_text()))
+    assert result["outcome"] == "failed"
+    with pytest.raises(ProcessLookupError):
+        os.kill(int((tmp_path / "fake.pid").read_text()), 0)
+
+
+def test_proxy_startup_waits_through_not_ready_status(tmp_path: Path) -> None:
+    fake: Final = _fake_litellm(
+        tmp_path,
+        """
+        import http.server, json, sys
+        port = int(sys.argv[sys.argv.index("--port") + 1])
+        hits = [0]
+        class H(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                hits[0] += 1
+                if hits[0] <= 2:
+                    self.send_response(503)
+                    self.end_headers()
+                    return
+                body = json.dumps({"status": "healthy", "db": "Not connected"}).encode()
+                self.send_response(200)
+                self.send_header("content-type", "application/json")
+                self.end_headers()
+                self.wfile.write(body)
+            def log_message(self, *a):
+                pass
+        http.server.HTTPServer(("127.0.0.1", port), H).serve_forever()
+        """,
+    )
+    diagnostics: Final = tmp_path / "diag"
+
+    proc: Final = _run(
+        tmp_path,
+        "proxy-startup",
+        "--diagnostics-dir",
+        str(diagnostics),
+        "--litellm-bin",
+        str(fake),
+        "--ready-deadline",
+        "15",
+    )
+
+    assert proc.returncode == 0, proc.stderr
+
+
+def test_proxy_startup_wrong_body_fails(tmp_path: Path) -> None:
+    fake: Final = _fake_litellm(
+        tmp_path,
+        """
+        import http.server, json, sys
+        port = int(sys.argv[sys.argv.index("--port") + 1])
+        class H(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                body = json.dumps({"status": "healthy", "db": "connected"}).encode()
+                self.send_response(200)
+                self.send_header("content-type", "application/json")
+                self.end_headers()
+                self.wfile.write(body)
+            def log_message(self, *a):
+                pass
+        http.server.HTTPServer(("127.0.0.1", port), H).serve_forever()
+        """,
+    )
+    diagnostics: Final = tmp_path / "diag"
+
+    proc: Final = _run(
+        tmp_path,
+        "proxy-startup",
+        "--diagnostics-dir",
+        str(diagnostics),
+        "--litellm-bin",
+        str(fake),
+        "--ready-deadline",
+        "15",
+    )
+
+    assert proc.returncode != 0
+    assert "connected" in proc.stderr
+
+
+def test_interpreter_expect_mismatch_fails() -> None:
+    proc: Final = _run(Path.cwd(), "interpreter", "--expect", "9.99")
+
+    assert proc.returncode != 0
+    assert "9.99" in proc.stderr
+
+
+def test_interpreter_expect_match_passes() -> None:
+    expect: Final = f"{sys.version_info.major}.{sys.version_info.minor}"
+
+    proc: Final = _run(Path.cwd(), "interpreter", "--expect", expect)
+
+    assert proc.returncode == 0
+    assert f"OK interpreter {expect}" in proc.stdout
diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
index 5bdc0e9f8b8..3d5c38c3acd 100644
--- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
+++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
@@ -14,6 +14,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import httpx
 import pytest
 from mcp.types import AudioContent, CallToolResult, ImageContent, TextContent
+from openai import AsyncOpenAI
 from openai._legacy_response import HttpxBinaryResponseContent
 
 import litellm
@@ -8426,3 +8427,174 @@ class TestBudgetReservationBinding:
 
         assert logging_obj.litellm_params["metadata"]["user_api_key_budget_reservation"] is reservation
         assert reservation["callback_bound"] is False
+
+
+@pytest.mark.asyncio
+async def test_standard_logging_payload_keeps_message_content_when_message_logging_is_on(monkeypatch):
+    outbound: Final = asyncio.Queue()
+    logs: Final = asyncio.Queue()
+    monkeypatch.setattr(litellm, "turn_off_message_logging", False)
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        outbound.put_nowait(json.loads(request.content))
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-smoke",
+                "object": "chat.completion",
+                "created": 0,
+                "model": "gpt-5.6",
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "smoke-marker-reply"},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            },
+        )
+
+    async def capture(kwargs, response_obj, start_time, end_time):
+        logs.put_nowait(kwargs["standard_logging_object"])
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as http_client:
+        client: Final = AsyncOpenAI(api_key="transport-only", http_client=http_client)
+        await litellm.acompletion(
+            model="openai/gpt-5.6",
+            api_key="transport-only",
+            client=client,
+            messages=[{"role": "user", "content": "smoke-marker-request"}],
+            success_callback=[capture],
+            num_retries=0,
+            max_retries=0,
+        )
+        payload: Final = await asyncio.wait_for(logs.get(), timeout=10)
+        request: Final = await asyncio.wait_for(outbound.get(), timeout=10)
+        assert outbound.empty()
+        assert request["messages"][0]["content"] == "smoke-marker-request"
+        assert payload["messages"][0]["content"] == "smoke-marker-request"
+        assert payload["response"]["choices"][0]["message"]["content"] == "smoke-marker-reply"
+
+
+@pytest.mark.asyncio
+async def test_standard_logging_payload_redacts_message_content_when_message_logging_is_off(monkeypatch):
+    outbound: Final = asyncio.Queue()
+    logs: Final = asyncio.Queue()
+    monkeypatch.setattr(litellm, "turn_off_message_logging", False)
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        outbound.put_nowait(json.loads(request.content))
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-smoke",
+                "object": "chat.completion",
+                "created": 0,
+                "model": "gpt-5.6",
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "smoke-marker-reply"},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            },
+        )
+
+    async def capture(kwargs, response_obj, start_time, end_time):
+        logs.put_nowait(kwargs["standard_logging_object"])
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as http_client:
+        client: Final = AsyncOpenAI(api_key="transport-only", http_client=http_client)
+        await litellm.acompletion(
+            model="openai/gpt-5.6",
+            api_key="transport-only",
+            client=client,
+            messages=[{"role": "user", "content": "smoke-marker-request"}],
+            turn_off_message_logging=True,
+            success_callback=[capture],
+            num_retries=0,
+            max_retries=0,
+        )
+        payload: Final = await asyncio.wait_for(logs.get(), timeout=10)
+        assert outbound.qsize() == 1
+        assert "smoke-marker-request" not in json.dumps(payload["messages"])
+        assert "smoke-marker-reply" not in json.dumps(payload["response"])
+        assert payload["model"]
+        assert payload["total_tokens"] == 15
+
+
+@pytest.mark.asyncio
+async def test_async_success_handler_delivers_standard_logging_payload_to_custom_logger():
+    events: Final = asyncio.Queue()
+
+    class SuccessRecorder(CustomLogger):
+        async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+            events.put_nowait((kwargs, response_obj))
+
+    recorder: Final = SuccessRecorder()
+    logging_obj: Final = LitellmLogging(
+        model="openai/gpt-5.6",
+        messages=[{"role": "user", "content": "smoke-callback-request"}],
+        stream=False,
+        call_type="acompletion",
+        start_time=time.time(),
+        litellm_call_id="smoke-callback-success",
+        function_id="smoke-callback-success",
+        dynamic_async_success_callbacks=[recorder],
+    )
+    logging_obj.model_call_details["litellm_params"] = {"metadata": {}, "proxy_server_request": {}}
+    result: Final = ModelResponse(
+        model="openai/gpt-5.6",
+        choices=[
+            {"index": 0, "message": {"role": "assistant", "content": "smoke-callback-reply"}, "finish_reason": "stop"}
+        ],
+        usage=litellm.Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
+    )
+    now: Final = datetime.datetime.now()
+
+    await logging_obj.async_success_handler(result=result, start_time=now, end_time=now, cache_hit=False)
+
+    kwargs, response_obj = await asyncio.wait_for(events.get(), timeout=10)
+    assert response_obj is result
+    payload: Final = kwargs["standard_logging_object"]
+    assert payload["status"] == "success"
+    assert payload["model"] == "openai/gpt-5.6"
+    assert payload["total_tokens"] == 15
+    assert events.empty()
+
+
+@pytest.mark.asyncio
+async def test_async_failure_handler_delivers_failure_payload_to_custom_logger():
+    events: Final = asyncio.Queue()
+
+    class FailureRecorder(CustomLogger):
+        async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+            events.put_nowait((kwargs, response_obj))
+
+    recorder: Final = FailureRecorder()
+    logging_obj: Final = LitellmLogging(
+        model="openai/gpt-5.6",
+        messages=[{"role": "user", "content": "smoke-callback-request"}],
+        stream=False,
+        call_type="acompletion",
+        start_time=time.time(),
+        litellm_call_id="smoke-callback-failure",
+        function_id="smoke-callback-failure",
+        dynamic_async_failure_callbacks=[recorder],
+    )
+    logging_obj.model_call_details["litellm_params"] = {"metadata": {}, "proxy_server_request": {}}
+    failure: Final = ValueError("smoke-failure")
+    now: Final = datetime.datetime.now()
+
+    await logging_obj.async_failure_handler(exception=failure, traceback_exception="", start_time=now, end_time=now)
+
+    kwargs, response_obj = await asyncio.wait_for(events.get(), timeout=10)
+    assert kwargs["exception"] is failure
+    payload: Final = kwargs["standard_logging_object"]
+    assert payload["status"] == "failure"
+    assert "smoke-failure" in payload["error_str"]
+    assert payload["model"] == "openai/gpt-5.6"
+    assert events.empty()
diff --git a/tests/test_litellm/llms/openai/test_openai.py b/tests/test_litellm/llms/openai/test_openai.py
index 136b837f191..9539e13a802 100644
--- a/tests/test_litellm/llms/openai/test_openai.py
+++ b/tests/test_litellm/llms/openai/test_openai.py
@@ -1,5 +1,12 @@
-import pytest
+import asyncio
+import json
+from typing import Final
 
+import httpx
+import pytest
+from openai import AsyncOpenAI
+
+import litellm
 from litellm.llms.openai.openai import OpenAIChatCompletion
 
 
@@ -50,3 +57,199 @@ def test_get_stream_options_passes_caller_stream_options_through_on_any_host(api
     assert OpenAIChatCompletion().get_stream_options(stream_options=caller_options, api_base=api_base) == {
         "stream_options": caller_options
     }
+
+
+@pytest.mark.asyncio
+async def test_acompletion_returns_json_reply_over_injected_transport():
+    outbound: Final = asyncio.Queue()
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        outbound.put_nowait(json.loads(request.content))
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-smoke",
+                "object": "chat.completion",
+                "created": 0,
+                "model": "gpt-5.6",
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "smoke-json-reply"},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            },
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as http_client:
+        client: Final = AsyncOpenAI(api_key="transport-only", http_client=http_client)
+        response: Final = await asyncio.wait_for(
+            litellm.acompletion(
+                model="openai/gpt-5.6",
+                api_key="transport-only",
+                client=client,
+                messages=[{"role": "user", "content": "smoke-json-request"}],
+                num_retries=0,
+                max_retries=0,
+            ),
+            timeout=10,
+        )
+        request: Final = await asyncio.wait_for(outbound.get(), timeout=10)
+        assert request["model"] == "gpt-5.6"
+        assert request["messages"] == [{"role": "user", "content": "smoke-json-request"}]
+        assert not request.get("stream")
+        assert outbound.empty()
+        assert response.choices[0].message.content == "smoke-json-reply"
+        assert response.choices[0].finish_reason == "stop"
+        assert response.usage.total_tokens == 15
+
+
+@pytest.mark.asyncio
+async def test_acompletion_streams_text_deltas_over_injected_transport():
+    outbound: Final = asyncio.Queue()
+
+    def chunk(delta: dict, finish: str | None) -> bytes:
+        body: Final = {
+            "id": "chatcmpl-smoke",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "gpt-5.6",
+            "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
+        }
+        return f"data: {json.dumps(body)}\n\n".encode()
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        outbound.put_nowait(json.loads(request.content))
+        usage: Final = {
+            "id": "chatcmpl-smoke",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "gpt-5.6",
+            "choices": [],
+            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+        }
+        content: Final = b"".join(
+            (
+                chunk({"role": "assistant", "content": "Hel"}, None),
+                chunk({"content": "lo"}, "stop"),
+                f"data: {json.dumps(usage)}\n\n".encode(),
+                b"data: [DONE]\n\n",
+            )
+        )
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=content)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as http_client:
+        client: Final = AsyncOpenAI(api_key="transport-only", http_client=http_client)
+        stream: Final = await litellm.acompletion(
+            model="openai/gpt-5.6",
+            api_key="transport-only",
+            client=client,
+            messages=[{"role": "user", "content": "smoke-stream-request"}],
+            stream=True,
+            num_retries=0,
+            max_retries=0,
+        )
+        chunks: Final = []
+
+        async def drain() -> None:
+            async for part in stream:
+                chunks.append(part)
+
+        await asyncio.wait_for(drain(), timeout=10)
+        request: Final = await asyncio.wait_for(outbound.get(), timeout=10)
+        assert request["stream"] is True
+        assert outbound.empty()
+        assert (
+            "".join(part.choices[0].delta.content or "" for part in chunks if part.choices and part.choices[0].delta)
+            == "Hello"
+        )
+        last_finish: Final = next(
+            part.choices[0].finish_reason for part in reversed(chunks) if part.choices and part.choices[0].finish_reason
+        )
+        assert last_finish == "stop"
+
+
+@pytest.mark.asyncio
+async def test_acompletion_streams_tool_call_arguments_over_injected_transport():
+    outbound: Final = asyncio.Queue()
+    tools: Final = [
+        {
+            "type": "function",
+            "function": {
+                "name": "get_weather",
+                "description": "Look up weather for a city",
+                "parameters": {
+                    "type": "object",
+                    "properties": {"city": {"type": "string"}},
+                    "required": ["city"],
+                },
+            },
+        }
+    ]
+
+    def chunk(delta: dict, finish: str | None) -> bytes:
+        body: Final = {
+            "id": "chatcmpl-smoke",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "gpt-5.6",
+            "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
+        }
+        return f"data: {json.dumps(body)}\n\n".encode()
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        outbound.put_nowait(json.loads(request.content))
+        content: Final = b"".join(
+            (
+                chunk(
+                    {
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call-1",
+                                "type": "function",
+                                "function": {"name": "get_weather", "arguments": ""},
+                            }
+                        ]
+                    },
+                    None,
+                ),
+                chunk({"tool_calls": [{"index": 0, "function": {"arguments": '{"city":'}}]}, None),
+                chunk({"tool_calls": [{"index": 0, "function": {"arguments": '"Paris"}'}}]}, "tool_calls"),
+                b"data: [DONE]\n\n",
+            )
+        )
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=content)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as http_client:
+        client: Final = AsyncOpenAI(api_key="transport-only", http_client=http_client)
+        messages: Final = [{"role": "user", "content": "weather in Paris"}]
+        stream: Final = await litellm.acompletion(
+            model="openai/gpt-4o",
+            api_key="transport-only",
+            client=client,
+            messages=messages,
+            tools=tools,
+            stream=True,
+            num_retries=0,
+            max_retries=0,
+        )
+        chunks: Final = []
+
+        async def drain() -> None:
+            async for part in stream:
+                chunks.append(part)
+
+        await asyncio.wait_for(drain(), timeout=10)
+        request: Final = await asyncio.wait_for(outbound.get(), timeout=10)
+        assert request["stream"] is True
+        assert request["tools"][0]["function"]["name"] == "get_weather"
+        assert outbound.empty()
+        rebuilt: Final = litellm.stream_chunk_builder(chunks, messages=messages)
+        tool_call: Final = rebuilt.choices[0].message.tool_calls[0]
+        assert tool_call.id == "call-1"
+        assert tool_call.function.name == "get_weather"
+        assert json.loads(tool_call.function.arguments) == {"city": "Paris"}
+        assert rebuilt.choices[0].finish_reason == "tool_calls"
diff --git a/tests/test_litellm/proxy/auth/test_auth_checks.py b/tests/test_litellm/proxy/auth/test_auth_checks.py
index ffba53a5b7a..aa4bb2e49d3 100644
--- a/tests/test_litellm/proxy/auth/test_auth_checks.py
+++ b/tests/test_litellm/proxy/auth/test_auth_checks.py
@@ -9314,3 +9314,14 @@ async def test_agent_key_without_an_echoed_caller_keeps_its_own_models():
     await _check_caller_models(agent_key, "claude-sonnet", load_team, load_user)
 
     assert asked == []
+
+
+def test_can_object_call_model_allows_listed_model_for_key():
+    result: Final = _can_object_call_model(
+        model="allowed-model",
+        llm_router=None,
+        models=["allowed-model"],
+        object_type="key",
+    )
+
+    assert result is True
diff --git a/tests/test_litellm/test_cost_calculator.py b/tests/test_litellm/test_cost_calculator.py
index ddcdeb61e02..10a904d6141 100644
--- a/tests/test_litellm/test_cost_calculator.py
+++ b/tests/test_litellm/test_cost_calculator.py
@@ -4633,3 +4633,49 @@ def test_gemini_live_native_audio_limits_and_capabilities_match_vendor_model_car
     assert info["supports_response_schema"] is False
     assert info["supports_url_context"] is False
     assert info["supports_pdf_input"] is False
+
+
+def test_completion_cost_charges_explicit_per_token_rates_over_registered_ones(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "smoke-priced-model",
+        {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02, "litellm_provider": "openai", "mode": "chat"},
+    )
+    response: Final = ModelResponse(
+        model="smoke-priced-model",
+        choices=[],
+        usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
+    )
+
+    cost: Final = completion_cost(
+        completion_response=response,
+        model="smoke-priced-model",
+        custom_llm_provider="openai",
+        custom_cost_per_token={"input_cost_per_token": 0.001, "output_cost_per_token": 0.002},
+    )
+
+    assert cost == pytest.approx(100 * 0.001 + 50 * 0.002)
+
+
+def test_completion_cost_is_zero_when_explicit_rates_are_zero(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setitem(
+        litellm.model_cost,
+        "smoke-priced-model",
+        {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02, "litellm_provider": "openai", "mode": "chat"},
+    )
+    response: Final = ModelResponse(
+        model="smoke-priced-model",
+        choices=[],
+        usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
+    )
+
+    cost: Final = completion_cost(
+        completion_response=response,
+        model="smoke-priced-model",
+        custom_llm_provider="openai",
+        custom_cost_per_token={"input_cost_per_token": 0.0, "output_cost_per_token": 0.0},
+    )
+
+    assert cost == 0.0