diff --git a/.github/e2e-stack/redact_output.py b/.github/e2e-stack/redact_output.py
new file mode 100644
index 00000000000..233a8be3e2d
--- /dev/null
+++ b/.github/e2e-stack/redact_output.py
@@ -0,0 +1,83 @@
+import argparse
+import os
+import sys
+from functools import reduce
+from pathlib import Path
+from typing import Final
+from xml.sax.saxutils import escape
+
+from pydantic import JsonValue, TypeAdapter, ValidationError
+from secrets_to_env import MIN_MASKED_LENGTH
+
+REDACTED: Final = "***"
+json_adapter: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
+
+
+def string_leaves(node: JsonValue) -> tuple[str, ...]:
+    match node:
+        case str():
+            return (node,)
+        case list():
+            return tuple(leaf for child in node for leaf in string_leaves(child))
+        case dict():
+            return tuple(leaf for child in node.values() for leaf in string_leaves(child))
+    return ()
+
+
+def field_lines(value: str) -> tuple[str, ...]:
+    try:
+        return tuple(line for leaf in string_leaves(json_adapter.validate_json(value)) for line in leaf.splitlines())
+    except ValidationError:
+        return ()
+
+
+def masked_values(values_files: tuple[Path, ...]) -> tuple[str, ...]:
+    values: Final = frozenset(
+        line.split("=", 1)[1].strip().strip("'")
+        for path in values_files
+        for line in path.read_text().splitlines()
+        if "=" in line
+    )
+    texts: Final = frozenset(text for value in values for text in (value, *field_lines(value)))
+    renderings: Final = frozenset(
+        rendering
+        for text in texts
+        if len(text) >= MIN_MASKED_LENGTH
+        for rendering in (text, escape(text), escape(text, {'"': """}))
+    )
+    return tuple(sorted(renderings, key=lambda rendering: (-len(rendering), rendering)))
+
+
+def redact(text: str, values: tuple[str, ...]) -> str:
+    return reduce(lambda redacted, value: redacted.replace(value, REDACTED), values, text)
+
+
+def write_redacted(source: Path, out_dir: Path, values: tuple[str, ...]) -> None:
+    target: Final = out_dir / source.name
+    with os.fdopen(os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600), "w") as handle:
+        _ = handle.write(redact(source.read_text(errors="replace"), values))
+
+
+def main() -> int:
+    parser: Final = argparse.ArgumentParser()
+    _ = parser.add_argument("--values", action="append", type=Path, required=True)
+    _ = parser.add_argument("--out", type=Path, required=True)
+    _ = parser.add_argument("files", nargs="*", type=Path)
+    args: Final = parser.parse_args()
+    values_files: Final = tuple(args.values)
+    out_dir: Final[Path] = args.out
+    sources: Final = tuple(args.files)
+    try:
+        values: Final = masked_values(values_files)
+        out_dir.mkdir(mode=0o700, exist_ok=True)
+        for source in sources:
+            write_redacted(source, out_dir, values)
+    except OSError as error:
+        _ = sys.stderr.write(f"could not redact {error.filename}\n")
+        return 1
+    _ = sys.stdout.write(f"redacted {len(sources)} file(s) into {out_dir}\n")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/.github/e2e-stack/secrets_to_env.py b/.github/e2e-stack/secrets_to_env.py
new file mode 100644
index 00000000000..691c10203bf
--- /dev/null
+++ b/.github/e2e-stack/secrets_to_env.py
@@ -0,0 +1,55 @@
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Final
+
+from pydantic import TypeAdapter, ValidationError
+
+secrets_adapter: Final[TypeAdapter[dict[str, str]]] = TypeAdapter(dict[str, str])
+ENV_NAME: Final = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
+MIN_MASKED_LENGTH: Final = 8
+ACTIONS_RUNNER_FLAG: Final = "GITHUB_ACTIONS"
+
+
+def main() -> int:
+    env_path: Final = Path(sys.argv[1])
+    try:
+        secrets: Final = {
+            key: value.rstrip("\r\n") for key, value in secrets_adapter.validate_json(sys.stdin.read()).items()
+        }
+    except (ValidationError, UnicodeError):
+        _ = sys.stderr.write("expected a JSON object containing string environment values\n")
+        return 1
+    unusable: Final = tuple(
+        key
+        for key, value in secrets.items()
+        if ENV_NAME.fullmatch(key) is None or any(char in value for char in "'\n\r\0")
+    )
+    if unusable:
+        _ = sys.stderr.write(
+            f"these names or values cannot be represented in both bash and dotenv: {' '.join(sorted(unusable))}\n"
+        )
+        return 1
+    if os.environ.get(ACTIONS_RUNNER_FLAG) == "true":
+        _ = sys.stdout.write(
+            "".join(
+                f"::add-mask::{value.replace('%', '%25')}\n"
+                for value in secrets.values()
+                if len(value) >= MIN_MASKED_LENGTH
+            )
+        )
+        sys.stdout.flush()
+    lines: Final = tuple(f"{key}='{value}'" for key, value in secrets.items() if value)
+    try:
+        with os.fdopen(os.open(env_path, os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_NOFOLLOW, 0o600), "w") as handle:
+            os.fchmod(handle.fileno(), 0o600)
+            _ = handle.write("\n".join(lines) + "\n")
+    except OSError:
+        _ = sys.stderr.write("could not write the environment file\n")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/.github/e2e-stack/select_tests.py b/.github/e2e-stack/select_tests.py
new file mode 100644
index 00000000000..792da5ae09c
--- /dev/null
+++ b/.github/e2e-stack/select_tests.py
@@ -0,0 +1,52 @@
+import re
+import sys
+from typing import Final
+
+SELECTABLE: Final = re.compile(r"^tests/e2e/([A-Za-z0-9_.-]+/)*test_[A-Za-z0-9_.-]+\.py$")
+UNSUPPORTED: Final = re.compile(
+    r"^tests/e2e/(ui|claude_code|load|migrations)/"
+    r"|^tests/e2e/mcp/test_mcp_oauth_happy_path_e2e\.py$"
+    r"|^tests/e2e/llm_translation/realtime/test_realtime_pipecat_audio_e2e\.py$"
+    r"|^tests/e2e/batches/test_managed_files_enforcement_e2e\.py$"
+    r"|^tests/e2e/guardrails/test_presidio_masking_e2e\.py$"
+    r"|^tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e\.py$"
+    r"|^tests/e2e/logging/test_langsmith_batch_serialization_e2e\.py$"
+    r"|^tests/e2e/logging/test_s3_log_e2e\.py$"
+    r"|^tests/e2e/secret_manager/"
+)
+HARNESS: Final = re.compile(
+    r"^tests/e2e/[A-Za-z0-9_.-]+\.(py|ini)$"
+    r"|^tests/e2e/idp_realm\.json$"
+    r"|^tests/e2e/management/(management_client|jwt_actors|conftest)\.py$"
+    r"|^tests/e2e/coverage_registry/management_cases\.py$"
+    r"|^tests/e2e/gateway/"
+    r"|^\.github/e2e-stack/"
+    r"|^\.github/workflows/test-e2e-changed\.yml$"
+)
+UNEXPANDED: Final = re.compile(r"[*?\[]")
+
+
+def is_selectable(path: str) -> bool:
+    return SELECTABLE.match(path) is not None and UNSUPPORTED.match(path) is None
+
+
+def select(changed: tuple[str, ...], canary: tuple[str, ...]) -> tuple[str, ...]:
+    direct: Final = frozenset(path for path in changed if is_selectable(path))
+    harness_changed: Final = any(HARNESS.match(path) for path in changed)
+    canary_tests: Final = frozenset(path for path in canary if harness_changed and is_selectable(path))
+    return tuple(sorted(direct | canary_tests))
+
+
+def main() -> int:
+    canary: Final = tuple(sys.argv[1:])
+    unexpanded: Final = tuple(path for path in canary if UNEXPANDED.search(path))
+    if unexpanded:
+        _ = sys.stderr.write(f"the canary paths reached the selector unexpanded: {' '.join(unexpanded)}\n")
+        return 1
+    changed: Final = tuple(line.strip() for line in sys.stdin if line.strip())
+    _ = sys.stdout.write(" ".join(select(changed, canary)) + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/code_coverage_tests/test_e2e_changed_gate.py b/tests/code_coverage_tests/test_e2e_changed_gate.py
index c4d12d1282c..758d579f67d 100644
--- a/tests/code_coverage_tests/test_e2e_changed_gate.py
+++ b/tests/code_coverage_tests/test_e2e_changed_gate.py
@@ -8,6 +8,10 @@ from typing import Final
 import pytest
 
 GATE: Final = Path(__file__).resolve().parents[2] / ".github/e2e-stack/assert_tests_ran.py"
+SECRETS_TO_ENV: Final = GATE.with_name("secrets_to_env.py")
+SELECT_TESTS: Final = GATE.with_name("select_tests.py")
+REDACT_OUTPUT: Final = GATE.with_name("redact_output.py")
+CANARY: Final = ("tests/e2e/access_control/test_a.py", "tests/e2e/access_control/test_b.py")
 SELECTED: Final = ("tests/e2e/access_control/test_a.py", "tests/e2e/access_control/test_b.py")
 
 
@@ -98,6 +102,238 @@ def test_one_passing_management_case_cannot_hide_a_missing_actor(tmp_path: Path,
     assert f"test_actor_subject_and_database_role[{omitted_role}]" in result.stdout
 
 
+def test_short_values_are_written_without_masking_every_digit_in_the_log(tmp_path: Path) -> None:
+    env_path: Final = tmp_path / ".env"
+
+    result: Final = subprocess.run(
+        [sys.executable, "-I", str(SECRETS_TO_ENV), str(env_path)],
+        input='{"FLAG": "1", "API_KEY": "sk-0123456789abcdef"}',
+        capture_output=True,
+        text=True,
+        env={**os.environ, "GITHUB_ACTIONS": "true"},
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout == "::add-mask::sk-0123456789abcdef\n"
+    assert env_path.read_text() == "FLAG='1'\nAPI_KEY='sk-0123456789abcdef'\n"
+
+
+def test_outside_actions_no_value_is_printed(tmp_path: Path) -> None:
+    env_path: Final = tmp_path / ".env"
+    local_env: Final = {key: value for key, value in os.environ.items() if key != "GITHUB_ACTIONS"}
+
+    result: Final = subprocess.run(
+        [sys.executable, "-I", str(SECRETS_TO_ENV), str(env_path)],
+        input='{"FLAG": "1", "API_KEY": "sk-0123456789abcdef"}',
+        capture_output=True,
+        text=True,
+        env=local_env,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout == ""
+    assert "sk-0123456789abcdef" not in result.stderr
+    assert env_path.read_text() == "FLAG='1'\nAPI_KEY='sk-0123456789abcdef'\n"
+
+
+def redact_output(tmp_path: Path, values: tuple[str, ...], text: str) -> tuple[subprocess.CompletedProcess[str], Path]:
+    env_path: Final = tmp_path / ".env"
+    _ = env_path.write_text("".join(f"{name}='{value}'\n" for name, value in zip(("A", "B", "C"), values)))
+    stack_env: Final = tmp_path / "stack.env"
+    _ = stack_env.write_text("LITELLM_MASTER_KEY=sk-e2e-master0123\nREDIS_PORT=6379\n")
+    log: Final = tmp_path / "e2e-pass-1.log"
+    _ = log.write_text(text)
+    out_dir: Final = tmp_path / "redacted"
+    result: Final = subprocess.run(  # test-quality-ok: standalone script that imports its sibling by script directory
+        [
+            sys.executable,
+            str(REDACT_OUTPUT),
+            "--values",
+            str(env_path),
+            "--values",
+            str(stack_env),
+            "--out",
+            str(out_dir),
+            str(log),
+        ],
+        capture_output=True,
+        text=True,
+    )
+    return result, out_dir / log.name
+
+
+def test_redacted_output_hides_every_masked_value_and_keeps_the_rest(tmp_path: Path) -> None:
+    text: Final = (
+        "FAILED key=sk-0123456789abcdef master=sk-e2e-master0123 flag=1 port=6379 message=Missing credentials\n"
+    )
+
+    result, redacted = redact_output(tmp_path, ("sk-0123456789abcdef", "1"), text)
+
+    assert result.returncode == 0, result.stderr
+    assert redacted.read_text() == "FAILED key=*** master=*** flag=1 port=6379 message=Missing credentials\n"
+    assert (redacted.stat().st_mode & 0o777) == 0o600
+    assert (tmp_path / "e2e-pass-1.log").read_text() == text
+    assert "sk-" not in result.stdout + result.stderr
+
+
+def test_a_masked_value_that_prefixes_a_longer_one_leaves_no_tail(tmp_path: Path) -> None:
+    result, redacted = redact_output(tmp_path, ("sk-0123456789", "sk-0123456789abcdef"), "token sk-0123456789abcdef\n")
+
+    assert result.returncode == 0, result.stderr
+    assert redacted.read_text() == "token ***\n"
+
+
+def test_a_json_secret_is_hidden_field_by_field_however_it_is_escaped(tmp_path: Path) -> None:
+    credentials: Final = (
+        '{"type": "service_account", "signing_key": "MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\\n'
+        'c2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\\n", "client_id": "104857600000000000001"}'
+    )
+    text: Final = (
+        "decoded MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\n"
+        "c2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\n"
+        "escaped MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\\nc2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\\n\n"
+        "twice MIIEvAIBADANBgkqhkiG9w0BAQEFAASC\\\\nc2VjcmV0LWtleS1ib2R5LWxpbmUtdHdv\n"
+        "client 104857600000000000001 status 403\n"
+    )
+
+    result, redacted = redact_output(tmp_path, (credentials,), text)
+
+    assert result.returncode == 0, result.stderr
+    assert redacted.read_text() == "decoded ***\n***\nescaped ***\\n***\\n\ntwice ***\\\\n***\nclient *** status 403\n"
+
+
+def test_a_secret_with_xml_special_characters_is_hidden_in_the_junit_file(tmp_path: Path) -> None:
+    text: Final = '<failure message="got p&amp;ss&lt;w&quot;rd-1">body p&amp;ss&lt;w"rd-1</failure>\n'
+
+    result, redacted = redact_output(tmp_path, ('p&ss<w"rd-1',), text)
+
+    assert result.returncode == 0, result.stderr
+    assert redacted.read_text() == '<failure message="got ***">body ***</failure>\n'
+
+
+def select_tests(changed: tuple[str, ...]) -> tuple[str, ...]:
+    result: Final = subprocess.run(
+        [sys.executable, str(SELECT_TESTS), *CANARY],
+        input="".join(f"{path}\n" for path in changed),
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0, result.stderr
+    return tuple(result.stdout.split())
+
+
+@pytest.mark.parametrize(
+    ("changed", "expected"),
+    (
+        (("tests/e2e/logging/test_datadog_e2e.py", "litellm/router.py"), ("tests/e2e/logging/test_datadog_e2e.py",)),
+        (("tests/e2e/ui/test_keys.py", "tests/e2e/claude_code/test_cli.py", "tests/e2e/load/test_burst.py"), ()),
+        (("tests/e2e/migrations/test_startup.py", "tests/e2e/migrations/test_recovery.py"), ()),
+        (("tests/e2e/batches/test_managed_files_enforcement_e2e.py",), ()),
+        (("tests/e2e/guardrails/test_presidio_masking_e2e.py",), ()),
+        (("tests/e2e/llm_translation/realtime/test_realtime_pipecat_audio_e2e.py",), ()),
+        (("tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e.py",), ()),
+        (
+            ("tests/e2e/logging/test_team_langfuse_callback_e2e.py",),
+            ("tests/e2e/logging/test_team_langfuse_callback_e2e.py",),
+        ),
+        (
+            ("tests/e2e/llm_translation/realtime/test_realtime_e2e.py",),
+            ("tests/e2e/llm_translation/realtime/test_realtime_e2e.py",),
+        ),
+        (
+            ("tests/e2e/guardrails/test_bedrock_guardrail_e2e.py",),
+            ("tests/e2e/guardrails/test_bedrock_guardrail_e2e.py",),
+        ),
+        (("tests/e2e/logging/helpers.py", "docs/my-website/docs/index.md", "tests/e2e/AGENTS.md"), ()),
+        (
+            ("tests/e2e/logging/test_datadog_e2e.py", "tests/e2e/logging/test_datadog_e2e.py"),
+            ("tests/e2e/logging/test_datadog_e2e.py",),
+        ),
+    ),
+)
+def test_changed_suite_files_are_selected_unless_the_stack_cannot_run_them(
+    changed: tuple[str, ...], expected: tuple[str, ...]
+) -> None:
+    assert select_tests(changed) == expected
+
+
+@pytest.mark.parametrize(
+    "harness_file",
+    (
+        "tests/e2e/proxy_client.py",
+        "tests/e2e/conftest.py",
+        "tests/e2e/management/management_client.py",
+        "tests/e2e/management/jwt_actors.py",
+        "tests/e2e/management/conftest.py",
+        "tests/e2e/coverage_registry/management_cases.py",
+        "tests/e2e/pytest.ini",
+        "tests/e2e/gateway/stage_mirror_ci_config.yml",
+        ".github/e2e-stack/up.sh",
+        ".github/e2e-stack/start-idp.sh",
+        "tests/e2e/idp_realm.json",
+        ".github/workflows/test-e2e-changed.yml",
+    ),
+)
+def test_harness_changes_run_the_canary_suite(harness_file: str) -> None:
+    assert select_tests((harness_file, "litellm/router.py")) == CANARY
+
+
+def test_a_changed_canary_file_is_selected_once_alongside_a_harness_change() -> None:
+    assert select_tests((CANARY[1], "tests/e2e/proxy_client.py")) == CANARY
+
+
+def test_dedicated_migration_tests_do_not_suppress_shared_harness_canaries() -> None:
+    assert select_tests(("tests/e2e/migrations/test_startup.py", "tests/e2e/conftest.py")) == CANARY
+
+
+def test_the_canary_joins_directly_selected_files_in_sorted_order() -> None:
+    assert select_tests(("tests/e2e/logging/test_datadog_e2e.py", ".github/e2e-stack/up.sh")) == (
+        *CANARY,
+        "tests/e2e/logging/test_datadog_e2e.py",
+    )
+
+
+def test_a_harness_unit_test_change_runs_itself_and_the_canary() -> None:
+    assert select_tests(("tests/e2e/test_proxy_client.py",)) == (*CANARY, "tests/e2e/test_proxy_client.py")
+
+
+def test_a_canary_argument_the_shell_never_expanded_fails_the_selector() -> None:
+    result: Final = subprocess.run(
+        [sys.executable, str(SELECT_TESTS), "tests/e2e/access_control/test_*.py"],
+        input="tests/e2e/proxy_client.py\n",
+        capture_output=True,
+        text=True,
+    )
+
+    assert result.returncode == 1
+    assert "tests/e2e/access_control/test_*.py" in result.stderr
+    assert result.stdout == ""
+
+
+@pytest.mark.parametrize(
+    ("secrets", "offender", "unprintable"),
+    (
+        ('{"AWS_ACCESS_KEY_ID": "AKIAEXAMPLE", "BAD-NAME": "shibboleth"}', "BAD-NAME", "shibboleth"),
+        ("""{"AWS_SECRET_ACCESS_KEY": "quote'shibboleth"}""", "AWS_SECRET_ACCESS_KEY", "shibboleth"),
+        ('{"DD_API_KEY": "line\\nshibboleth"}', "DD_API_KEY", "shibboleth"),
+    ),
+)
+def test_an_unusable_secret_is_named_without_printing_its_value(
+    tmp_path: Path, secrets: str, offender: str, unprintable: str
+) -> None:
+    env_path: Final = tmp_path / ".env"
+
+    result: Final = subprocess.run(
+        [sys.executable, str(SECRETS_TO_ENV), str(env_path)], input=secrets, capture_output=True, text=True
+    )
+
+    assert result.returncode == 1
+    assert offender in result.stderr
+    assert unprintable not in result.stderr
+    assert result.stdout == ""
+    assert not env_path.exists()
+
+
 @pytest.mark.parametrize("phase", ("setup", "call", "teardown"))
 @pytest.mark.parametrize("required_count", ("1", "4"))
 def test_oauth_failure_diagnostics_do_not_publish_private_payloads(