diff --git a/scripts/redis_acl_safety.py b/scripts/redis_acl_safety.py
new file mode 100644
index 0000000..38907ff
--- /dev/null
+++ b/scripts/redis_acl_safety.py
@@ -0,0 +1,590 @@
+#!/usr/bin/env python3
+"""P1-5 incident remediation.
+
+On 2026-09-18, a production Redis FLUSHALL was executed during ad hoc
+ACL negative testing of the redis-exporter monitoring identity, because:
+
+  1. Redis's `default` user was enabled with `nopass`.
+  2. A connection opened to production Redis was therefore already
+     authenticated as `default` the instant it connected -- before any
+     AUTH was ever attempted.
+  3. An explicit `AUTH redis_monitoring <wrong password>` failed.
+  4. The failed AUTH did NOT deauthenticate the connection -- Redis
+     left it exactly as it was: authenticated as the already-logged-in
+     `default` user.
+  5. Subsequent commands issued on that same connection, including
+     FLUSHALL, therefore executed with default's unrestricted +@all
+     permissions instead of failing.
+
+This module exists to make that entire failure class structurally
+impossible for any caller that uses it, rather than relying on
+operator discipline the next time someone needs to validate a Redis
+ACL identity:
+
+  - `RedisEnvironment` is a closed three-way classification
+    (DISPOSABLE / PRODUCTION / UNKNOWN). UNKNOWN is treated at least as
+    strictly as PRODUCTION everywhere below -- there is no code path
+    that relaxes any check merely because classification evidence is
+    absent. Disposable status is never inferred from a hostname,
+    port difference, container name, or an unset/empty environment
+    variable; it requires a `DisposableAttestation` that this module
+    independently re-verifies live against the target instance (a
+    fresh nonce round-trip plus a hard rejection of this
+    architecture's known production-adjacent port), mirroring the
+    existing `tests/_redis_integration_guard.py` pattern used
+    elsewhere in this repo for exactly the same class of mistake.
+  - Every authenticated session is built on a brand-new connection
+    (never a reused/pooled one that might already be authenticated as
+    someone else). `authenticate()` sends AUTH as the very first
+    command on that connection; on ANY failure the connection is
+    closed immediately and unconditionally, before a second command
+    can ever reach the wire -- see `authenticate()`'s own test
+    coverage for the exact incident sequence reproduced end to end.
+  - After AUTH succeeds, the session is unusable until `ACL WHOAMI`
+    has confirmed the *actual* authenticated identity equals the
+    identity the caller asked for. A mismatch is a hard failure, never
+    a warning.
+  - Every command a caller wants to run passes through exactly one
+    gate (`CommandGate.authorize`) before it reaches the wire.
+    `ProductionValidator` enforces a narrow allowlist; disposable
+    callers get `DisposableValidator` (allowlist-free, but a
+    hard-denylist for the always-dangerous administrative command
+    family still applies) or, for the one legitimate case where a test
+    must prove Redis's own ACL denies a dangerous command,
+    `DisposableNegativeTestGate` -- the only gate that passes
+    dangerous commands through, and the only gate that requires a
+    freshly-reverified `DisposableAttestation` to even construct.
+  - `AuthenticatedSession` never exposes the underlying `redis.Redis`
+    client through any public method or attribute -- "get the raw
+    client and do whatever you want" is exactly the escape hatch that
+    would defeat every guarantee above. (Python has no true private
+    attributes; see this module's own tests and the accompanying
+    report for the honest limit of what "no bypass" means here.)
+
+Nothing in this module ever prints, logs, or includes in an exception
+message a password, a credential-bearing URL, a JWT, or an ACL hash --
+see `RedisSafetyError` and `_safe_repr_command`.
+"""
+from __future__ import annotations
+
+import secrets
+from dataclasses import dataclass
+from typing import Sequence
+from urllib.parse import urlsplit
+
+import redis
+
+# This architecture's shared, production-adjacent Redis port (matches
+# tests/_redis_integration_guard.py's FORBIDDEN_PORT convention exactly --
+# kept as a plain module constant here rather than imported from there,
+# since that module is pytest-only and this one is not).
+PRODUCTION_PORT = 6380
+
+
+# ---------------------------------------------------------------------------
+# Errors -- never include secret material in a message.
+# ---------------------------------------------------------------------------
+
+
+class RedisSafetyError(RuntimeError):
+    """Base class for every error this module raises. Messages may
+    freely include: environment classification, expected/actual
+    identity names, command category, and reason. They must never
+    include a password, a credential-bearing URL, a JWT, or an ACL
+    hash."""
+
+
+class EnvironmentClassificationError(RedisSafetyError):
+    """A target could not be positively proven DISPOSABLE (or a
+    disposable-only construct was asked to trust a target that isn't
+    one)."""
+
+
+class AuthenticationFailedError(RedisSafetyError):
+    """AUTH did not unambiguously succeed. The connection that produced
+    this error has already been closed by the time this is raised --
+    see `authenticate()`."""
+
+
+class IdentityMismatchError(RedisSafetyError):
+    """`ACL WHOAMI`'s answer did not exactly equal the identity the
+    caller asserted before issuing it."""
+
+
+class ProhibitedCommandError(RedisSafetyError):
+    """A caller asked a `CommandGate` to run a command that its active
+    policy does not permit."""
+
+
+class DangerousCommandError(ProhibitedCommandError):
+    """The specific case of `ProhibitedCommandError` for a command on
+    the hard, environment-independent dangerous-command denylist."""
+
+
+# ---------------------------------------------------------------------------
+# Environment classification -- disposable status is proven, never inferred.
+# ---------------------------------------------------------------------------
+
+
+class RedisEnvironment:
+    """Closed three-way classification. Deliberately not an IntEnum/str
+    subclass a caller could accidentally coerce from an arbitrary
+    string -- these three instances are the only values that exist."""
+
+    DISPOSABLE = None  # assigned below
+    PRODUCTION = None
+    UNKNOWN = None
+
+    def __init__(self, name: str):
+        self._name = name
+
+    def __repr__(self) -> str:
+        return f"RedisEnvironment.{self._name}"
+
+    def __eq__(self, other):
+        return self is other
+
+    def __hash__(self):
+        return id(self)
+
+
+RedisEnvironment.DISPOSABLE = RedisEnvironment("DISPOSABLE")
+RedisEnvironment.PRODUCTION = RedisEnvironment("PRODUCTION")
+RedisEnvironment.UNKNOWN = RedisEnvironment("UNKNOWN")
+
+
+@dataclass(frozen=True)
+class DisposableAttestation:
+    """Positive, independently-reverifiable proof that a specific
+    Redis endpoint is disposable test infrastructure.
+
+    Every field is required, with no default -- there is no way to
+    construct one of these by omission. `nonce_value` must already
+    have been written to `nonce_key` on the target instance by the
+    caller's own disposable-Redis bootstrap code (using whatever
+    access that caller's disposable instance grants -- this module
+    never writes it). `classify_environment` independently reads it
+    back; a caller that has not actually written a matching key to the
+    actual target instance cannot manufacture a passing attestation by
+    editing Python values alone.
+    """
+
+    host: str
+    port: int
+    nonce_key: str
+    nonce_value: str
+
+    def __post_init__(self):
+        if self.port == PRODUCTION_PORT:
+            raise EnvironmentClassificationError(
+                f"port {self.port} is this architecture's shared, "
+                "production-adjacent Redis port -- refusing to attest "
+                "any endpoint on it as disposable, regardless of other "
+                "evidence"
+            )
+        if not self.host:
+            raise EnvironmentClassificationError("attestation requires a non-empty host")
+        if not self.nonce_key or not self.nonce_value:
+            raise EnvironmentClassificationError(
+                "attestation requires a non-empty nonce_key/nonce_value pair "
+                "-- an empty nonce proves nothing"
+            )
+
+    @staticmethod
+    def generate_nonce() -> str:
+        """Convenience for callers: a fresh, unguessable nonce value.
+        Still the caller's responsibility to actually write it to the
+        disposable instance before constructing the attestation."""
+        return secrets.token_hex(16)
+
+
+def classify_environment(
+    *, host: str, port: int, attestation: DisposableAttestation | None
+) -> RedisEnvironment:
+    """Returns DISPOSABLE only after independently re-verifying the
+    attestation against the live instance. Returns UNKNOWN whenever no
+    attestation was supplied -- there is no fallback path that infers
+    DISPOSABLE from a hostname, a port merely being *different* from
+    PRODUCTION_PORT, a container name, an empty database, or an unset
+    environment variable alone."""
+    if port == PRODUCTION_PORT:
+        return RedisEnvironment.PRODUCTION
+    if attestation is None:
+        return RedisEnvironment.UNKNOWN
+    if attestation.host != host or attestation.port != port:
+        raise EnvironmentClassificationError(
+            "attestation host/port do not match the target host/port -- "
+            "refusing to trust an attestation for a different endpoint "
+            "than the one being classified"
+        )
+    probe = None
+    try:
+        # Construction itself may connect eagerly (redis-py does this
+        # when single_connection_client=True), so it must be inside
+        # this same try/except -- a connection failure at construction
+        # time must fail closed exactly like a failure during the
+        # command that follows it.
+        probe = redis.Redis(
+            host=host, port=port, socket_timeout=2, socket_connect_timeout=2,
+            decode_responses=True, single_connection_client=True,
+        )
+        got = probe.execute_command("GET", attestation.nonce_key)
+    except redis.RedisError as exc:
+        raise EnvironmentClassificationError(
+            f"could not independently verify disposable attestation: probe "
+            f"connection/command failed ({type(exc).__name__})"
+        ) from None
+    finally:
+        if probe is not None:
+            try:
+                probe.close()
+            except Exception:  # noqa: BLE001 -- best-effort cleanup only
+                pass
+    if got != attestation.nonce_value:
+        raise EnvironmentClassificationError(
+            "disposable attestation nonce did not match the value actually "
+            "stored on the live instance -- refusing to classify as "
+            "DISPOSABLE"
+        )
+    return RedisEnvironment.DISPOSABLE
+
+
+# ---------------------------------------------------------------------------
+# Command classification -- allowlist-first, denylist as defense in depth.
+# ---------------------------------------------------------------------------
+
+# Commands/subcommands that are unconditionally prohibited through this
+# module's gates, regardless of environment -- including on disposable
+# Redis, via every gate except DisposableNegativeTestGate (whose entire,
+# sole purpose is proving Redis's own ACL denies these to a restricted
+# identity; see its own docstring).
+DANGEROUS_COMMANDS: frozenset[tuple[str, ...]] = frozenset({
+    ("FLUSHALL",),
+    ("FLUSHDB",),
+    ("SHUTDOWN",),
+    ("DEBUG",),
+    ("CONFIG", "SET"),
+    ("ACL", "SETUSER"),
+    ("ACL", "DELUSER"),
+    ("ACL", "LOAD"),
+    ("ACL", "SAVE"),
+    ("ACL", "LOG"),
+    ("MODULE", "LOAD"),
+    ("MODULE", "UNLOAD"),
+    ("MIGRATE",),
+    ("RESTORE",),
+    ("RESTORE-ASKING",),
+    ("SWAPDB",),
+    ("REPLICAOF",),
+    ("SLAVEOF",),
+})
+
+# The minimum set of operations needed to demonstrate connectivity and
+# harmless read-only capability against PRODUCTION Redis. Deliberately
+# small: this module's own job is to prove "this identity authenticates
+# and is who we expect", not to exercise application-level behavior --
+# that belongs in each service's own tests against its own scoped
+# identity (see e.g. the redis_lims_cache/redis_audit_health_reader
+# validation performed during the P1-5 migrations, which used the real
+# application client code, not this harness).
+PRODUCTION_ALLOWED_COMMANDS: frozenset[tuple[str, ...]] = frozenset({
+    ("PING",),
+    ("ACL", "WHOAMI"),
+    ("INFO",),
+})
+
+# Commands whose ACL representation in this codebase's generated
+# users.acl is a two-token "COMMAND SUBCOMMAND" form (see
+# ~/omnibioai-redis-backups/bin/generate_production_acl.sh) -- used to
+# decide whether to normalize a caller's args to one or two tokens.
+_KNOWN_SUBCOMMAND_PREFIXES = frozenset({
+    "ACL", "CONFIG", "CLIENT", "XGROUP", "XINFO", "SCRIPT", "FUNCTION",
+    "MODULE", "COMMAND", "CLUSTER", "LATENCY", "SLOWLOG", "MEMORY",
+    "OBJECT", "PUBSUB",
+})
+
+
+def _normalize_command(args: Sequence) -> tuple[str, ...]:
+    """Case-insensitive, bytes-or-str-insensitive, whitespace-insensitive
+    normalization to a canonical (COMMAND,) or (COMMAND, SUBCOMMAND)
+    tuple. This is the single place command identity is decided --
+    every gate calls this, so a bypass would have to avoid this
+    function entirely, not merely pass a differently-cased string
+    through it."""
+    if not args:
+        raise ProhibitedCommandError("refusing to authorize an empty command")
+    parts = []
+    for a in args:
+        if isinstance(a, bytes):
+            a = a.decode("utf-8", errors="replace")
+        parts.append(str(a).strip().upper())
+    if not parts or not parts[0]:
+        raise ProhibitedCommandError("refusing to authorize an empty command")
+    if len(parts) >= 2 and parts[0] in _KNOWN_SUBCOMMAND_PREFIXES:
+        return (parts[0], parts[1])
+    return (parts[0],)
+
+
+def _safe_repr_command(args: Sequence) -> str:
+    """A command's name/subcommand only, for use in error messages --
+    never the full argv, which could contain a value being written
+    (e.g. a credential) even for an otherwise-harmless-looking
+    command."""
+    try:
+        cmd = _normalize_command(args)
+    except ProhibitedCommandError:
+        return "<empty>"
+    return " ".join(cmd)
+
+
+# ---------------------------------------------------------------------------
+# Command gates.
+# ---------------------------------------------------------------------------
+
+
+class CommandGate:
+    """Base class. Every concrete gate must implement `authorize`,
+    which either returns None (permitted) or raises a
+    `ProhibitedCommandError` subclass. No subclass may add a method
+    that lets a caller run a command without going through
+    `authorize` first."""
+
+    def authorize(self, args: Sequence) -> None:  # pragma: no cover - abstract
+        raise NotImplementedError
+
+
+class ProductionValidator(CommandGate):
+    """The only gate usable for PRODUCTION or UNKNOWN environments.
+    Allowlist-first: a command must be explicitly on
+    `PRODUCTION_ALLOWED_COMMANDS` to pass, and is additionally checked
+    against `DANGEROUS_COMMANDS` first as defense in depth (redundant
+    today, since the allowlist is already a strict subset of safe
+    commands, but kept so a future accidental allowlist expansion
+    still can't silently permit a dangerous command)."""
+
+    def authorize(self, args: Sequence) -> None:
+        cmd = _normalize_command(args)
+        if cmd in DANGEROUS_COMMANDS:
+            raise DangerousCommandError(
+                f"command {' '.join(cmd)!r} is unconditionally prohibited "
+                "in production validation"
+            )
+        if cmd not in PRODUCTION_ALLOWED_COMMANDS:
+            raise ProhibitedCommandError(
+                f"command {' '.join(cmd)!r} is not on the production "
+                f"validation allowlist "
+                f"{sorted(' '.join(c) for c in PRODUCTION_ALLOWED_COMMANDS)}"
+            )
+
+
+class DisposableValidator(CommandGate):
+    """For ordinary (non-destructive-negative-test) validation against
+    a proven-disposable target -- e.g. exercising the same
+    authenticate()/assert-identity flow used in production, but against
+    disposable Redis. Broad allow, EXCEPT the hard dangerous-command
+    denylist still applies here too: this gate is not the sanctioned
+    path for negative destructive tests -- see
+    `DisposableNegativeTestGate` for that.
+
+    Construction re-verifies disposability itself (via
+    `classify_environment`) rather than trusting a `RedisEnvironment`
+    value the caller might have computed incorrectly or reused stale."""
+
+    def __init__(self, *, host: str, port: int, attestation: DisposableAttestation):
+        environment = classify_environment(host=host, port=port, attestation=attestation)
+        if environment is not RedisEnvironment.DISPOSABLE:
+            raise EnvironmentClassificationError(
+                "DisposableValidator requires a live-verified DISPOSABLE "
+                "environment; construction refused"
+            )
+        self.environment = environment
+
+    def authorize(self, args: Sequence) -> None:
+        cmd = _normalize_command(args)
+        if cmd in DANGEROUS_COMMANDS:
+            raise DangerousCommandError(
+                f"command {' '.join(cmd)!r} is on the hard denylist; use "
+                "DisposableNegativeTestGate if a test specifically needs to "
+                "prove Redis's own ACL denies this command to a restricted "
+                "identity"
+            )
+
+
+class DisposableNegativeTestGate(CommandGate):
+    """The ONLY gate that passes dangerous commands through to the
+    wire. Exists solely so a negative test can prove that a restricted
+    identity's own Redis ACL denies e.g. FLUSHALL, by actually sending
+    FLUSHALL and observing Redis's NOPERM response -- never by this
+    module silently agreeing not to send it.
+
+    Structurally cannot be constructed for a non-disposable target:
+    like `DisposableValidator`, construction re-runs
+    `classify_environment` itself. There is no flag, kwarg, or
+    subclass that widens this gate's reach to PRODUCTION or UNKNOWN --
+    the only environment `classify_environment` will ever return
+    besides DISPOSABLE is PRODUCTION or UNKNOWN, both of which raise
+    here.
+    """
+
+    def __init__(self, *, host: str, port: int, attestation: DisposableAttestation):
+        environment = classify_environment(host=host, port=port, attestation=attestation)
+        if environment is not RedisEnvironment.DISPOSABLE:
+            raise EnvironmentClassificationError(
+                "DisposableNegativeTestGate requires a live-verified "
+                "DISPOSABLE environment; construction refused"
+            )
+        self.environment = environment
+
+    def authorize(self, args: Sequence) -> None:
+        _normalize_command(args)  # validate shape only; everything permitted
+
+
+# ---------------------------------------------------------------------------
+# Authenticated sessions.
+# ---------------------------------------------------------------------------
+
+
+class AuthenticatedSession:
+    """A single Redis connection that has just been proven to
+    authenticate as exactly the identity the caller expected, paired
+    with the `CommandGate` that authorizes every command sent on it.
+
+    Deliberately exposes no method or attribute that returns the
+    underlying `redis.Redis` client. This is a genuine API design
+    choice, not a hard security boundary -- Python has no true private
+    attributes, so code inside this same process that specifically
+    reaches for `session._client` can still get at it. What this
+    guarantees is that no *ordinary* caller path (including every
+    other function in this module) ever needs or is given that
+    access; see `test_redis_acl_safety.py::test_no_raw_client_escape_hatch`
+    for exactly what is and is not enforced.
+    """
+
+    def __init__(self, *, environment: RedisEnvironment, gate: CommandGate, client: "redis.Redis"):
+        self.environment = environment
+        self._gate = gate
+        self._client = client
+        self._closed = False
+
+    def run(self, *args: str):
+        if self._closed:
+            raise RedisSafetyError("cannot run a command on a closed session")
+        self._gate.authorize(args)
+        return self._client.execute_command(*args)
+
+    def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        try:
+            self._client.close()
+        except Exception:  # noqa: BLE001 -- best-effort cleanup only
+            pass
+
+    def __enter__(self) -> "AuthenticatedSession":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        self.close()
+
+
+def authenticate(
+    *,
+    host: str,
+    port: int,
+    username: str,
+    password: str,
+    expected_identity: str,
+    gate: CommandGate,
+    environment: RedisEnvironment,
+    socket_timeout: float = 3.0,
+) -> AuthenticatedSession:
+    """Opens a brand-new connection (never a reused/pooled one),
+    attempts AUTH as the very first command on it, and -- this is the
+    exact property the incident depended on NOT holding -- closes that
+    connection unconditionally on ANY failure before returning control
+    to the caller. There is no path through this function that leaves
+    a caller holding a connection that failed AUTH but remains
+    implicitly authenticated as whatever the server treats a fresh
+    connection as (typically `default`).
+
+    After AUTH succeeds, immediately asserts identity via `ACL WHOAMI`
+    (itself passed through `gate.authorize`, so every gate's allowlist
+    must include it -- `PRODUCTION_ALLOWED_COMMANDS` does). A mismatch
+    is a hard `IdentityMismatchError`; there is no warning-only path.
+    """
+    client = None
+    try:
+        try:
+            # Construction itself may connect eagerly (redis-py does
+            # this when single_connection_client=True), so it must be
+            # inside this same try/except -- an unreachable host must
+            # fail closed exactly like an AUTH failure does, not leak
+            # a raw redis.ConnectionError past this function.
+            client = redis.Redis(
+                host=host, port=port, socket_timeout=socket_timeout,
+                socket_connect_timeout=socket_timeout, decode_responses=True,
+                single_connection_client=True,
+            )
+            ok = client.execute_command("AUTH", username, password)
+        except redis.AuthenticationError:
+            raise AuthenticationFailedError(
+                f"AUTH failed for expected identity {expected_identity!r}"
+            ) from None
+        except redis.RedisError as exc:
+            raise AuthenticationFailedError(
+                f"AUTH did not succeed for expected identity "
+                f"{expected_identity!r}: {type(exc).__name__}"
+            ) from None
+        if ok is not True and ok != "OK":
+            raise AuthenticationFailedError(
+                f"AUTH returned an unexpected result for expected identity "
+                f"{expected_identity!r}"
+            )
+        gate.authorize(("ACL", "WHOAMI"))
+        try:
+            actual = client.execute_command("ACL", "WHOAMI")
+        except redis.RedisError as exc:
+            raise IdentityMismatchError(
+                f"could not confirm identity for expected {expected_identity!r}: "
+                f"{type(exc).__name__}"
+            ) from None
+        if actual != expected_identity:
+            raise IdentityMismatchError(
+                f"expected identity {expected_identity!r}, ACL WHOAMI reported "
+                f"{actual!r}"
+            )
+    except RedisSafetyError:
+        if client is not None:
+            client.close()
+        raise
+    except Exception as exc:  # noqa: BLE001 -- convert anything unexpected to a safety error, never leak it raw
+        if client is not None:
+            client.close()
+        raise AuthenticationFailedError(
+            f"unexpected error establishing identity for {expected_identity!r}: "
+            f"{type(exc).__name__}"
+        ) from exc
+    return AuthenticatedSession(environment=environment, gate=gate, client=client)
+
+
+def authenticate_production(
+    *, host: str, port: int, username: str, password: str, expected_identity: str,
+    socket_timeout: float = 3.0,
+) -> AuthenticatedSession:
+    """Convenience wrapper: production validation, always. Refuses to
+    proceed if `port` is not this architecture's known production
+    port, so a caller cannot accidentally point "production" validation
+    at something else and get the narrow allowlist confused for a
+    disposable-appropriate check."""
+    if port != PRODUCTION_PORT:
+        raise EnvironmentClassificationError(
+            f"authenticate_production requires port {PRODUCTION_PORT}; got "
+            f"{port}. Use authenticate() directly with an explicit gate for "
+            "any other target."
+        )
+    return authenticate(
+        host=host, port=port, username=username, password=password,
+        expected_identity=expected_identity, gate=ProductionValidator(),
+        environment=RedisEnvironment.PRODUCTION, socket_timeout=socket_timeout,
+    )
diff --git a/tests/test_redis_acl_safety.py b/tests/test_redis_acl_safety.py
new file mode 100644
index 0000000..9104fd6
--- /dev/null
+++ b/tests/test_redis_acl_safety.py
@@ -0,0 +1,692 @@
+"""Tests for scripts/redis_acl_safety.py -- the P1-5 incident remediation
+framework (see that module's own docstring for the full incident
+narrative).
+
+Two tiers:
+
+  - Pure unit tests (command normalization, gate authorize/deny
+    decisions, error taxonomy, secret hygiene) need no Redis at all and
+    always run.
+
+  - Real-Redis tests need an actual disposable Redis server, because
+    fakeredis does not implement Redis's real ACL/AUTH semantics
+    (specifically: that a failed AUTH leaves a connection's
+    pre-existing identity intact rather than deauthenticating it --
+    the exact incident behavior this framework exists to survive).
+    `disposable_redis` (session-scoped fixture below) starts one via
+    Docker on a random high port, mirroring
+    tests/_redis_integration_guard.py's existing convention: no
+    implicit default, and port 6380 (this architecture's shared
+    production-adjacent Redis port) is refused outright even if
+    something upstream ever tried to hand it to this fixture. Skipped
+    automatically if the `docker` CLI is unavailable.
+"""
+from __future__ import annotations
+
+import secrets
+import shutil
+import subprocess
+import time
+
+import pytest
+import redis
+
+from scripts.redis_acl_safety import (
+    DANGEROUS_COMMANDS,
+    PRODUCTION_ALLOWED_COMMANDS,
+    PRODUCTION_PORT,
+    AuthenticatedSession,
+    AuthenticationFailedError,
+    DangerousCommandError,
+    DisposableAttestation,
+    DisposableNegativeTestGate,
+    DisposableValidator,
+    EnvironmentClassificationError,
+    IdentityMismatchError,
+    ProductionValidator,
+    ProhibitedCommandError,
+    RedisEnvironment,
+    _normalize_command,
+    _safe_repr_command,
+    authenticate,
+    authenticate_production,
+    classify_environment,
+)
+
+# ---------------------------------------------------------------------------
+# Pure unit tests: command normalization.
+# ---------------------------------------------------------------------------
+
+
+class TestNormalizeCommand:
+    def test_simple_command_uppercased(self):
+        assert _normalize_command(["ping"]) == ("PING",)
+        assert _normalize_command(["PiNg"]) == ("PING",)
+
+    def test_subcommand_family_normalized_to_two_tokens(self):
+        assert _normalize_command(["config", "set", "maxmemory", "0"]) == ("CONFIG", "SET")
+        assert _normalize_command(["ACL", "setuser", "x"]) == ("ACL", "SETUSER")
+        assert _normalize_command(["xgroup", "CREATE", "s", "g"]) == ("XGROUP", "CREATE")
+        assert _normalize_command(["script", "flush"]) == ("SCRIPT", "FLUSH")
+
+    def test_non_subcommand_family_stays_one_token_even_with_extra_args(self):
+        assert _normalize_command(["GET", "somekey"]) == ("GET",)
+        assert _normalize_command(["SET", "k", "v"]) == ("SET",)
+
+    def test_bytes_args_normalized_same_as_str(self):
+        assert _normalize_command([b"FlUsHaLl"]) == ("FLUSHALL",)
+        assert _normalize_command([b"config", b"SET"]) == ("CONFIG", "SET")
+
+    def test_whitespace_stripped(self):
+        assert _normalize_command([" flushall "]) == ("FLUSHALL",)
+
+    def test_mixed_case_variations_all_equal(self):
+        variants = ["FLUSHALL", "flushall", "FlUsHaLl", "fLUSHALL"]
+        assert len({_normalize_command([v]) for v in variants}) == 1
+
+    def test_empty_command_rejected(self):
+        with pytest.raises(ProhibitedCommandError):
+            _normalize_command([])
+
+    def test_empty_string_command_rejected(self):
+        with pytest.raises(ProhibitedCommandError):
+            _normalize_command([""])
+
+    def test_safe_repr_never_includes_extra_args(self):
+        assert _safe_repr_command(["SET", "k", "supersecretvalue"]) == "SET"
+        assert "supersecretvalue" not in _safe_repr_command(["SET", "k", "supersecretvalue"])
+        assert _safe_repr_command(["ACL", "SETUSER", "x", ">password"]) == "ACL SETUSER"
+        assert "password" not in _safe_repr_command(["ACL", "SETUSER", "x", ">password"])
+
+
+# ---------------------------------------------------------------------------
+# Pure unit tests: ProductionValidator (allowlist-first).
+# ---------------------------------------------------------------------------
+
+
+class TestProductionValidatorAllowlist:
+    @pytest.fixture
+    def gate(self):
+        return ProductionValidator()
+
+    @pytest.mark.parametrize("cmd", [["PING"], ["ping"], ["ACL", "WHOAMI"], ["acl", "whoami"], ["INFO"], ["info"]])
+    def test_allowed_commands_pass(self, gate, cmd):
+        gate.authorize(cmd)  # must not raise
+
+    def test_allowlist_is_exactly_the_documented_minimum(self):
+        assert PRODUCTION_ALLOWED_COMMANDS == frozenset({("PING",), ("ACL", "WHOAMI"), ("INFO",)})
+
+    @pytest.mark.parametrize("cmd", [
+        ["GET", "somekey"], ["SET", "k", "v"], ["DEL", "k"], ["XADD", "s", "*", "f", "v"],
+        ["XLEN", "s"], ["SCAN", "0"], ["KEYS", "*"], ["DBSIZE"], ["CLIENT", "LIST"],
+    ])
+    def test_non_allowlisted_harmless_looking_commands_still_denied(self, gate, cmd):
+        with pytest.raises(ProhibitedCommandError):
+            gate.authorize(cmd)
+
+    @pytest.mark.parametrize("cmd", [
+        ["FLUSHALL"], ["flushall"], ["FlUsHaLl"], ["FLUSHDB"], ["SHUTDOWN"],
+        ["SHUTDOWN", "NOSAVE"], ["DEBUG", "SLEEP", "0"], ["CONFIG", "SET", "maxmemory", "0"],
+        ["config", "set", "x", "y"], ["ACL", "SETUSER", "x"], ["acl", "setuser", "x"],
+        ["ACL", "DELUSER", "x"], ["ACL", "LOAD"], ["ACL", "SAVE"], ["ACL", "LOG", "RESET"],
+        ["MODULE", "LOAD", "x"], ["MODULE", "UNLOAD", "x"], ["MIGRATE"], ["RESTORE"],
+        ["RESTORE-ASKING"], ["SWAPDB", "0", "1"], ["REPLICAOF", "no", "one"], ["SLAVEOF", "no", "one"],
+    ])
+    def test_dangerous_commands_denied_as_dangerous_specifically(self, gate, cmd):
+        with pytest.raises(DangerousCommandError):
+            gate.authorize(cmd)
+
+    def test_dangerous_error_is_a_prohibited_command_error(self, gate):
+        with pytest.raises(ProhibitedCommandError):
+            gate.authorize(["FLUSHALL"])
+
+    def test_bytes_command_cannot_bypass_dangerous_check(self, gate):
+        with pytest.raises(DangerousCommandError):
+            gate.authorize([b"FLUSHALL"])
+
+    def test_subcommand_split_cannot_bypass_dangerous_check(self, gate):
+        # A caller cannot dodge the ("CONFIG", "SET") tuple by passing the
+        # subcommand as a separate positional differently-cased token.
+        with pytest.raises(DangerousCommandError):
+            gate.authorize(["CONFIG", "set"])
+        with pytest.raises(DangerousCommandError):
+            gate.authorize(["config", "SET"])
+
+    def test_error_message_never_contains_argument_values(self, gate):
+        with pytest.raises(ProhibitedCommandError) as excinfo:
+            gate.authorize(["SET", "k", "topsecretvalue123"])
+        assert "topsecretvalue123" not in str(excinfo.value)
+
+
+# ---------------------------------------------------------------------------
+# Pure unit tests: environment classification without a real Redis.
+# ---------------------------------------------------------------------------
+
+
+class TestClassifyEnvironmentWithoutRedis:
+    def test_production_port_always_classified_production_even_with_no_attestation(self):
+        assert classify_environment(host="redis", port=PRODUCTION_PORT, attestation=None) is RedisEnvironment.PRODUCTION
+
+    def test_unknown_when_no_attestation_supplied(self):
+        # No live probe is even attempted when there's no attestation --
+        # a bogus host/port here would still correctly resolve to UNKNOWN.
+        assert classify_environment(host="nonexistent.invalid", port=59999, attestation=None) is RedisEnvironment.UNKNOWN
+
+    def test_attestation_construction_rejects_production_port(self):
+        with pytest.raises(EnvironmentClassificationError):
+            DisposableAttestation(host="redis", port=PRODUCTION_PORT, nonce_key="k", nonce_value="v")
+
+    def test_attestation_construction_rejects_empty_nonce(self):
+        with pytest.raises(EnvironmentClassificationError):
+            DisposableAttestation(host="localhost", port=16399, nonce_key="", nonce_value="")
+
+    def test_attestation_construction_rejects_empty_host(self):
+        with pytest.raises(EnvironmentClassificationError):
+            DisposableAttestation(host="", port=16399, nonce_key="k", nonce_value="v")
+
+    def test_mismatched_host_port_attestation_rejected(self):
+        attestation = DisposableAttestation(host="127.0.0.1", port=16399, nonce_key="k", nonce_value="v")
+        with pytest.raises(EnvironmentClassificationError):
+            classify_environment(host="127.0.0.1", port=16400, attestation=attestation)
+
+    def test_unreachable_disposable_target_fails_closed_not_disposable(self):
+        # Well-formed attestation, but nothing is actually listening --
+        # must raise, never silently fall through to DISPOSABLE.
+        attestation = DisposableAttestation(host="127.0.0.1", port=1, nonce_key="k", nonce_value="v")
+        with pytest.raises(EnvironmentClassificationError):
+            classify_environment(host="127.0.0.1", port=1, attestation=attestation)
+
+    def test_disposable_validator_construction_fails_closed_when_unreachable(self):
+        attestation = DisposableAttestation(host="127.0.0.1", port=1, nonce_key="k", nonce_value="v")
+        with pytest.raises(EnvironmentClassificationError):
+            DisposableValidator(host="127.0.0.1", port=1, attestation=attestation)
+
+    def test_negative_test_gate_construction_fails_closed_when_unreachable(self):
+        attestation = DisposableAttestation(host="127.0.0.1", port=1, nonce_key="k", nonce_value="v")
+        with pytest.raises(EnvironmentClassificationError):
+            DisposableNegativeTestGate(host="127.0.0.1", port=1, attestation=attestation)
+
+
+# ---------------------------------------------------------------------------
+# Pure unit tests: authenticate() production-port guard (no real Redis
+# needed -- this specific check happens before any connection attempt).
+# ---------------------------------------------------------------------------
+
+
+class TestAuthenticateProductionGuard:
+    def test_authenticate_production_rejects_non_production_port(self):
+        with pytest.raises(EnvironmentClassificationError):
+            authenticate_production(
+                host="127.0.0.1", port=16399, username="x", password="y",
+                expected_identity="x",
+            )
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis fixture: a genuinely disposable instance, mirroring
+# tests/_redis_integration_guard.py's own safety conventions.
+# ---------------------------------------------------------------------------
+
+_DOCKER_AVAILABLE = shutil.which("docker") is not None
+
+
+def _run(cmd, **kw):
+    return subprocess.run(cmd, capture_output=True, text=True, timeout=30, **kw)
+
+
+@pytest.fixture(scope="session")
+def disposable_redis():
+    """Starts a throwaway redis:7-alpine container on a random high
+    port, with `default` left on/nopass (reproducing the exact
+    precondition of the incident) plus a restricted `redis_monitoring`
+    test identity. Never uses PRODUCTION_PORT -- refuses to proceed if
+    Docker ever handed back that port by coincidence.
+    """
+    if not _DOCKER_AVAILABLE:
+        pytest.skip("docker CLI not available; skipping real-Redis safety tests")
+
+    name = f"p15-safety-test-{secrets.token_hex(4)}"
+    _run(["docker", "rm", "-f", name])
+    started = _run(["docker", "run", "-d", "--rm", "--name", name, "-p", "127.0.0.1::6379", "redis:7-alpine"])
+    if started.returncode != 0:
+        pytest.skip(f"could not start disposable Redis container: {started.stderr.strip()[:200]}")
+
+    try:
+        port_out = _run(["docker", "port", name, "6379/tcp"])
+        host_port = int(port_out.stdout.strip().rsplit(":", 1)[-1])
+        if host_port == PRODUCTION_PORT:
+            pytest.fail(
+                "disposable Redis container was allocated this architecture's "
+                "production-adjacent port by coincidence -- refusing to use it"
+            )
+
+        for _ in range(20):
+            ping = _run(["docker", "exec", name, "redis-cli", "PING"])
+            if ping.stdout.strip() == "PONG":
+                break
+            time.sleep(0.5)
+        else:
+            pytest.fail("disposable Redis container never became reachable")
+
+        nonce_key = "p15_safety_test_nonce"
+        nonce_value = secrets.token_hex(16)
+        _run(["docker", "exec", name, "redis-cli", "SET", nonce_key, nonce_value])
+
+        monitoring_password = secrets.token_hex(16)
+        _run([
+            "docker", "exec", name, "redis-cli", "ACL", "SETUSER", "redis_monitoring_test",
+            "on", f">{monitoring_password}", "resetkeys", "resetchannels", "-@all",
+            "+ping", "+info", "+acl|whoami",
+        ])
+
+        yield {
+            "host": "127.0.0.1",
+            "port": host_port,
+            "container": name,
+            "nonce_key": nonce_key,
+            "nonce_value": nonce_value,
+            "monitoring_password": monitoring_password,
+        }
+    finally:
+        _run(["docker", "rm", "-f", name])
+
+
+@pytest.fixture
+def attestation(disposable_redis):
+    return DisposableAttestation(
+        host=disposable_redis["host"], port=disposable_redis["port"],
+        nonce_key=disposable_redis["nonce_key"], nonce_value=disposable_redis["nonce_value"],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: environment classification against a genuine instance.
+# ---------------------------------------------------------------------------
+
+
+class TestClassifyEnvironmentRealRedis:
+    def test_valid_attestation_classified_disposable(self, disposable_redis, attestation):
+        env = classify_environment(host=disposable_redis["host"], port=disposable_redis["port"], attestation=attestation)
+        assert env is RedisEnvironment.DISPOSABLE
+
+    def test_wrong_nonce_value_fails_closed(self, disposable_redis):
+        bad = DisposableAttestation(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            nonce_key=disposable_redis["nonce_key"], nonce_value="not-the-real-nonce",
+        )
+        with pytest.raises(EnvironmentClassificationError):
+            classify_environment(host=disposable_redis["host"], port=disposable_redis["port"], attestation=bad)
+
+    def test_wrong_nonce_key_fails_closed(self, disposable_redis):
+        bad = DisposableAttestation(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            nonce_key="nonexistent_key_never_set", nonce_value=disposable_redis["nonce_value"],
+        )
+        with pytest.raises(EnvironmentClassificationError):
+            classify_environment(host=disposable_redis["host"], port=disposable_redis["port"], attestation=bad)
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: THE incident regression test.
+# ---------------------------------------------------------------------------
+
+
+class TestIncidentReproductionAndDefense:
+    """Phase 5 of the remediation task: reproduce the actual incident
+    condition in disposable Redis, confirm Redis's real behavior is
+    exactly what caused it, then prove this framework survives it."""
+
+    def test_raw_redis_reproduces_the_incident_condition(self, disposable_redis):
+        """Ground truth, using the raw redis-py client directly (NOT
+        this framework) -- proves the server-level behavior this
+        framework has to defend against is real, on this Redis
+        version, right now. A connection that fails AUTH remains
+        authenticated as whatever it was before (here: `default`,
+        since it's `nopass`/enabled on this disposable instance) and a
+        subsequent command still executes.
+        """
+        client = redis.Redis(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            decode_responses=True, single_connection_client=True,
+        )
+        try:
+            with pytest.raises(redis.AuthenticationError):
+                client.execute_command("AUTH", "redis_monitoring_test", "definitely-the-wrong-password")
+            # This is the crux of the incident: despite the AUTH failure
+            # above, the connection is still usable and still `default`.
+            whoami = client.execute_command("ACL", "WHOAMI")
+            assert whoami == "default"
+            probe_key = "p15_incident_repro_probe"
+            # A real write succeeds here -- this is what FLUSHALL did in
+            # production. Using a harmless SET on disposable Redis,
+            # scoped to this test's own throwaway key, to avoid needing
+            # a second destructive assertion for the same point.
+            assert client.execute_command("SET", probe_key, "1") in (True, "OK")
+            assert client.execute_command("GET", probe_key) == "1"
+        finally:
+            client.close()
+
+    def test_framework_closes_connection_on_failed_auth_zero_commands_after(self, disposable_redis):
+        """The actual regression test: authenticate() with a wrong
+        password must raise AuthenticationFailedError and must not
+        leave a usable connection behind -- verified by checking
+        `_closed`/that `.run()` refuses to proceed."""
+        gate = ProductionValidator()
+        with pytest.raises(AuthenticationFailedError):
+            session = authenticate(
+                host=disposable_redis["host"], port=disposable_redis["port"],
+                username="redis_monitoring_test", password="definitely-the-wrong-password",
+                expected_identity="redis_monitoring_test", gate=gate,
+                environment=RedisEnvironment.UNKNOWN,
+            )
+            session.run("PING")  # unreachable if authenticate() raised, as required
+
+    def test_framework_never_returns_a_session_after_failed_auth(self, disposable_redis, monkeypatch):
+        """Belt-and-suspenders: instrument the underlying client's
+        execute_command to prove nothing beyond AUTH is ever sent on a
+        connection that failed AUTH."""
+        sent = []
+        real_execute = redis.Redis.execute_command
+
+        def spy(self, *args, **kwargs):
+            sent.append(args[0] if args else None)
+            return real_execute(self, *args, **kwargs)
+
+        monkeypatch.setattr(redis.Redis, "execute_command", spy)
+        gate = ProductionValidator()
+        with pytest.raises(AuthenticationFailedError):
+            authenticate(
+                host=disposable_redis["host"], port=disposable_redis["port"],
+                username="redis_monitoring_test", password="definitely-the-wrong-password",
+                expected_identity="redis_monitoring_test", gate=gate,
+                environment=RedisEnvironment.UNKNOWN,
+            )
+        assert sent == ["AUTH"], f"expected only AUTH to have been sent, got {sent!r}"
+
+    def test_framework_succeeds_with_correct_credential_and_matching_identity(self, disposable_redis):
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        try:
+            assert session.run("PING") is True or session.run("PING") == "PONG"
+        finally:
+            session.close()
+
+    def test_session_after_close_refuses_further_commands(self, disposable_redis):
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        session.close()
+        with pytest.raises(RuntimeError):
+            session.run("PING")
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: identity mismatch is a hard failure.
+# ---------------------------------------------------------------------------
+
+
+class TestIdentityMismatchRealRedis:
+    def test_expected_identity_not_matching_actual_is_hard_failure(self, disposable_redis):
+        """Authenticate correctly as redis_monitoring_test, but assert
+        the WRONG expected identity -- must be a hard
+        IdentityMismatchError, never a warning, never silently
+        continuing with a differently-privileged identity than the
+        caller asked for."""
+        gate = ProductionValidator()
+        with pytest.raises(IdentityMismatchError):
+            authenticate(
+                host=disposable_redis["host"], port=disposable_redis["port"],
+                username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+                expected_identity="default",  # deliberately wrong
+                gate=gate, environment=RedisEnvironment.UNKNOWN,
+            )
+
+    def test_mismatch_error_does_not_leave_a_usable_session(self, disposable_redis, monkeypatch):
+        sent = []
+        real_execute = redis.Redis.execute_command
+
+        def spy(self, *args, **kwargs):
+            sent.append(args[0] if args else None)
+            return real_execute(self, *args, **kwargs)
+
+        monkeypatch.setattr(redis.Redis, "execute_command", spy)
+        gate = ProductionValidator()
+        with pytest.raises(IdentityMismatchError):
+            authenticate(
+                host=disposable_redis["host"], port=disposable_redis["port"],
+                username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+                expected_identity="nonexistent_identity", gate=gate,
+                environment=RedisEnvironment.UNKNOWN,
+            )
+        assert sent == ["AUTH", "ACL"], f"expected AUTH then ACL WHOAMI only, got {sent!r}"
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: other authentication-failure shapes (Phase 9).
+# ---------------------------------------------------------------------------
+
+
+class TestAuthenticationFailureShapesRealRedis:
+    def test_nonexistent_username(self, disposable_redis):
+        gate = ProductionValidator()
+        with pytest.raises(AuthenticationFailedError):
+            authenticate(
+                host=disposable_redis["host"], port=disposable_redis["port"],
+                username="this_user_was_never_created", password="whatever",
+                expected_identity="this_user_was_never_created", gate=gate,
+                environment=RedisEnvironment.UNKNOWN,
+            )
+
+    def test_correct_username_correct_password_succeeds(self, disposable_redis):
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        session.close()
+
+    def test_unreachable_host_fails_closed(self):
+        gate = ProductionValidator()
+        with pytest.raises(AuthenticationFailedError):
+            authenticate(
+                host="127.0.0.1", port=1, username="x", password="y",
+                expected_identity="x", gate=gate, environment=RedisEnvironment.UNKNOWN,
+                socket_timeout=1,
+            )
+
+    def test_reconnect_after_failure_does_not_inherit_prior_state(self, disposable_redis):
+        """A second, fresh authenticate() call after a failed one must
+        not be affected by the prior failure -- each call opens its
+        own brand-new connection."""
+        gate = ProductionValidator()
+        with pytest.raises(AuthenticationFailedError):
+            authenticate(
+                host=disposable_redis["host"], port=disposable_redis["port"],
+                username="redis_monitoring_test", password="wrong-again",
+                expected_identity="redis_monitoring_test", gate=gate,
+                environment=RedisEnvironment.UNKNOWN,
+            )
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        session.close()
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: production allowlist enforced end to end.
+# ---------------------------------------------------------------------------
+
+
+class TestProductionAllowlistEndToEndRealRedis:
+    def test_allowed_command_runs(self, disposable_redis):
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        try:
+            info = session.run("INFO")
+            assert info  # non-empty
+        finally:
+            session.close()
+
+    def test_flushall_never_reaches_the_wire_through_production_session(self, disposable_redis, monkeypatch):
+        sent = []
+        real_execute = redis.Redis.execute_command
+
+        def spy(self, *args, **kwargs):
+            sent.append(args[0] if args else None)
+            return real_execute(self, *args, **kwargs)
+
+        monkeypatch.setattr(redis.Redis, "execute_command", spy)
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        try:
+            with pytest.raises(DangerousCommandError):
+                session.run("FLUSHALL")
+        finally:
+            session.close()
+        assert "FLUSHALL" not in sent, "FLUSHALL must never reach execute_command through a production session"
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: disposable-only destructive proof (Phase 8).
+# ---------------------------------------------------------------------------
+
+
+class TestDisposableDestructiveProofRealRedis:
+    def test_restricted_identity_denied_flushall_by_redis_itself(self, disposable_redis, attestation):
+        """This is the one place a dangerous command is actually sent
+        -- through DisposableNegativeTestGate, against a live-verified
+        disposable target, to observe Redis's own NOPERM. Proves the
+        restricted test identity genuinely cannot FLUSHALL; does not
+        merely trust this framework's own denylist."""
+        gate = DisposableNegativeTestGate(
+            host=disposable_redis["host"], port=disposable_redis["port"], attestation=attestation,
+        )
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.DISPOSABLE,
+        )
+        try:
+            with pytest.raises(redis.exceptions.NoPermissionError):
+                session.run("FLUSHALL")
+        finally:
+            session.close()
+
+    def test_restricted_identity_denied_set_on_unauthorized_key(self, disposable_redis, attestation):
+        gate = DisposableNegativeTestGate(
+            host=disposable_redis["host"], port=disposable_redis["port"], attestation=attestation,
+        )
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.DISPOSABLE,
+        )
+        try:
+            with pytest.raises(redis.exceptions.NoPermissionError):
+                session.run("SET", "some:unrelated:key", "v")
+        finally:
+            session.close()
+
+    def test_negative_test_gate_cannot_be_constructed_for_production_port(self, disposable_redis):
+        # Even with a technically-well-formed attestation pointed at the
+        # disposable instance, asking classify_environment to check
+        # PRODUCTION_PORT directly must never say DISPOSABLE.
+        assert classify_environment(host=disposable_redis["host"], port=PRODUCTION_PORT, attestation=None) is RedisEnvironment.PRODUCTION
+
+    def test_disposable_validator_denies_dangerous_commands_even_though_target_is_disposable(self, disposable_redis, attestation):
+        """DisposableValidator (not the negative-test gate) must still
+        refuse dangerous commands -- it is not the sanctioned path for
+        destructive negative testing, even against a proven-disposable
+        target."""
+        gate = DisposableValidator(host=disposable_redis["host"], port=disposable_redis["port"], attestation=attestation)
+        with pytest.raises(DangerousCommandError):
+            gate.authorize(["FLUSHALL"])
+
+
+# ---------------------------------------------------------------------------
+# Real-Redis: unknown environment behaves at least as strictly as production.
+# ---------------------------------------------------------------------------
+
+
+class TestUnknownEnvironmentRealRedis:
+    def test_unknown_environment_with_production_validator_still_denies_dangerous(self, disposable_redis):
+        # Simulates a caller that never supplied a DisposableAttestation
+        # (environment resolves to UNKNOWN) but still, correctly, uses
+        # ProductionValidator for anything of unproven status.
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        try:
+            with pytest.raises(DangerousCommandError):
+                session.run("FLUSHALL")
+        finally:
+            session.close()
+
+    def test_missing_attestation_never_implies_disposable(self):
+        assert classify_environment(host="127.0.0.1", port=16399, attestation=None) is RedisEnvironment.UNKNOWN
+        assert classify_environment(host="localhost", port=16399, attestation=None) is RedisEnvironment.UNKNOWN
+        assert classify_environment(host="some-test-container", port=16399, attestation=None) is RedisEnvironment.UNKNOWN
+
+
+# ---------------------------------------------------------------------------
+# Raw-client bypass (Phase 11).
+# ---------------------------------------------------------------------------
+
+
+class TestRawClientBypassRealRedis:
+    def test_authenticated_session_exposes_no_public_raw_client_accessor(self, disposable_redis):
+        gate = ProductionValidator()
+        session = authenticate(
+            host=disposable_redis["host"], port=disposable_redis["port"],
+            username="redis_monitoring_test", password=disposable_redis["monitoring_password"],
+            expected_identity="redis_monitoring_test", gate=gate,
+            environment=RedisEnvironment.UNKNOWN,
+        )
+        try:
+            public_attrs = [a for a in dir(session) if not a.startswith("_")]
+            assert "client" not in public_attrs
+            assert "redis_client" not in public_attrs
+            assert "raw_client" not in public_attrs
+            assert "connection_pool" not in public_attrs
+            assert "execute_command" not in public_attrs
+            assert set(public_attrs) <= {"environment", "run", "close"}
+        finally:
+            session.close()
+
+    def test_only_run_and_close_are_the_public_surface(self):
+        assert AuthenticatedSession.run is not None
+        assert AuthenticatedSession.close is not None
+        # documents, rather than technically enforces, that this is the
+        # entire intended public surface -- see the module/class
+        # docstrings for the honest limit of this guarantee (no true
+        # private attributes exist in Python).