fix(sandbox): PR #12655 simplification — correctness fixes + test consolidation

Phase 1 (correctness): replace hand-rolled equality helpers with
pydantic-canonicalized `_normalize_section` (via `model_dump`); future field
additions to `InternetAccessConfig` / `PythonDependenciesConfig` now track
automatically. `_env_vars_equal` switches from `frozenset` to `Counter` so
duplicate multiplicity is preserved. Extend runtime `_resolve_user_env` to
reject reserved `EnvVarLiteral.name` symmetrically with the existing
`EnvVarSecretRef.secret_key` check, closing the pre-mutation-guard-row
exposure D7 was meant to close.

Phase 2 (test consolidation): drop three integration-layer duplicate files
(`test_env_var_end_to_end.py`, `test_credential_resolution_integration.py`,
`test_reserved_credential_names.py`) after folding unique scenarios into
their unit-level counterparts; prune pydantic-framework-rechecking tests
(`test_unknown_keys_preserved`, `test_empty_config_returns_defaults`, trivial
round-trips) from `test_sandbox_config_validation.py`; drop `backend._user_env`
internal-attribute assertions from `test_user_env_forwarding.py`; replace
`_BACKEND_CACHE` dict-shape peeks in `test_sandbox_cache_invalidation.py` with
identity-based observable checks; parametrize per-adapter shape assertions in
`test_capability_advertisement.py` over `SANDBOX_ADAPTER_METADATA.keys()`; add
module-level docstrings to `test_daytona_backend.py` / `test_vercel_backend.py`
/ `test_modal_backend.py` stating the per-adapter-files-only-for-unique-lifecycle
principle; move `test_sandbox_config_validation.py` from `tests/unit/server/api/`
to `tests/unit/server/sandbox/` (tests sandbox pydantic models, not API).

Phase 3 (clarity): rename `_PHOENIX_SANDBOX_FALLBACK_CREDENTIAL_KEYS` →
`_PHOENIX_RESERVED_CREDENTIAL_ONLY_KEYS` with an inline docstring distinguishing
it from `SandboxAdapter.credential_specs` and `RESERVED_CREDENTIAL_NAMES`.

Net: +599 / -718. All 429 sandbox-relevant tests pass.

Author: anticorrelator

src/phoenix/server/sandbox/__init__.py

@@ -387,10 +387,15 @@ async def _resolve_user_env(
 
     ta: TypeAdapter[EnvVarEntry] = TypeAdapter(EnvVarEntry)
     entries: list[EnvVarEntry] = [ta.validate_python(e) for e in raw_env_vars]
-    # Fail-closed: reject reserved secret_key values before any DB lookup so
-    # rows persisted before the mutation-layer guard shipped cannot be silently
-    # resolved. This mirrors the mutation-layer check in _check_env_var_collision.
+    # Fail-closed: reject reserved names before any DB lookup so rows persisted
+    # before the mutation-layer guard shipped cannot be silently resolved.
+    # Mirrors _check_env_var_collision which checks both literal name and secret_key.
     for entry in entries:
+        if isinstance(entry, EnvVarLiteral) and is_reserved_credential_name(entry.name):
+            raise MissingSecretError(
+                f"env_var name {entry.name!r} is a reserved sandbox provider "
+                "credential and cannot be used as a user-defined environment variable."
+            )
         if isinstance(entry, EnvVarSecretRef) and is_reserved_credential_name(entry.secret_key):
             raise MissingSecretError(
                 f"secret_ref.secret_key {entry.secret_key!r} is a reserved sandbox "
@@ -624,8 +629,11 @@ async def get_or_create_backend(
 # e2b adapter not registered) cannot narrow the reserved set.
 # ---------------------------------------------------------------------------
 
-_PHOENIX_SANDBOX_FALLBACK_CREDENTIAL_KEYS: frozenset[str] = frozenset(
+_PHOENIX_RESERVED_CREDENTIAL_ONLY_KEYS: frozenset[str] = frozenset(
     {
+        # Reservation-only names: NOT settable via setSandboxCredential mutation.
+        # Contrast with SandboxAdapter.credential_specs (adapter-declared, settable
+        # via mutation). RESERVED_CREDENTIAL_NAMES is the derived union of both.
         "PHOENIX_SANDBOX_TOKEN",
         "PHOENIX_SANDBOX_API_KEY",
         # Modal tokens — added here so dropping credential_specs from ModalAdapter
@@ -637,7 +645,7 @@ async def get_or_create_backend(
 
 
 def _build_reserved_credential_names() -> frozenset[str]:
-    names: set[str] = {key.lower() for key in _PHOENIX_SANDBOX_FALLBACK_CREDENTIAL_KEYS}
+    names: set[str] = {key.lower() for key in _PHOENIX_RESERVED_CREDENTIAL_ONLY_KEYS}
     for adapter in _SANDBOX_ADAPTERS.values():
         for spec in adapter.credential_specs:
             names.add(spec.key.lower())

src/phoenix/server/sandbox/types.py

@@ -323,50 +323,90 @@ async def stop_session(self, session_key: str) -> None:
 # ---------------------------------------------------------------------------
 
 
-def _env_var_key(entry: Any) -> tuple[str, str, str, str]:
-    """Stable sort/hash key for a single env_var entry dict."""
+def _normalize_env_var(entry: Any) -> dict[str, Any]:
+    """Return a stable dict representation of a single env_var entry."""
     if isinstance(entry, dict):
-        return (
-            entry.get("kind", ""),
-            entry.get("name", ""),
-            entry.get("value", ""),
-            entry.get("secret_key", ""),
-        )
-    return (
-        getattr(entry, "kind", ""),
-        getattr(entry, "name", ""),
-        getattr(entry, "value", ""),
-        getattr(entry, "secret_key", ""),
-    )
+        return dict(entry)
+    # pydantic model instance — use model_dump for canonical representation
+    if hasattr(entry, "model_dump"):
+        result: dict[str, Any] = entry.model_dump(mode="json")
+        return result
+    return {
+        "kind": getattr(entry, "kind", ""),
+        "name": getattr(entry, "name", ""),
+        "value": getattr(entry, "value", ""),
+        "secret_key": getattr(entry, "secret_key", ""),
+    }
 
 
 def _env_vars_equal(a: Any, b: Any) -> bool:
-    """Return True if two env_vars lists are semantically equal (order-independent)."""
+    """Return True if two env_vars lists are semantically equal (order-independent).
+
+    Uses Counter over canonical tuple representations so duplicate entries in
+    one list are not collapsed — [X, X] != [X].
+    """
+    from collections import Counter
+
     if not a and not b:
         return True
     if not a or not b:
         return False
-    return frozenset(_env_var_key(e) for e in a) == frozenset(_env_var_key(e) for e in b)
+
+    def _to_tuple(entry: Any) -> tuple[str, ...]:
+        d = _normalize_env_var(entry)
+        return (d.get("kind", ""), d.get("name", ""), d.get("value", ""), d.get("secret_key", ""))
+
+    return Counter(_to_tuple(e) for e in a) == Counter(_to_tuple(e) for e in b)
+
+
+def _normalize_section(value: Any, model_cls: Type[BaseModel]) -> dict[str, Any]:
+    """Normalize a config section through pydantic model_dump so comparisons track the schema."""
+    if value is None:
+        return {}
+    if isinstance(value, dict):
+        dumped: dict[str, Any] = model_cls.model_validate(value).model_dump(
+            mode="json", exclude_defaults=False
+        )
+        return dumped
+    if hasattr(value, "model_dump"):
+        dumped = value.model_dump(mode="json", exclude_defaults=False)
+        return dumped
+    return {}
 
 
 def _internet_access_equal(a: Any, b: Any) -> bool:
-    """Return True if two internet_access values are semantically equal."""
+    """Return True if two internet_access values are semantically equal.
+
+    Canonicalizes through InternetAccessConfig.model_dump so future fields
+    are automatically included rather than silently dropped.
+    """
     if a is None and b is None:
         return True
     if a is None or b is None:
         return False
-    mode_a = a.get("mode") if isinstance(a, dict) else getattr(a, "mode", None)
-    mode_b = b.get("mode") if isinstance(b, dict) else getattr(b, "mode", None)
-    return mode_a == mode_b
+    return _normalize_section(a, InternetAccessConfig) == _normalize_section(
+        b, InternetAccessConfig
+    )
 
 
 def _packages_equal(a: Any, b: Any) -> bool:
-    """Return True if two packages lists are semantically equal (set comparison)."""
+    """Return True if two dependencies sections are semantically equal.
+
+    Canonicalizes through PythonDependenciesConfig.model_dump so the lockfile
+    field is included — set(packages) alone is insufficient. Package list order
+    is not semantically meaningful, so packages are sorted before comparison.
+    """
     if not a and not b:
         return True
     if not a or not b:
         return False
-    return set(a) == set(b)
+
+    def _canonical(value: Any) -> dict[str, Any]:
+        d = _normalize_section(value, PythonDependenciesConfig)
+        d["packages"] = sorted(d.get("packages") or [])
+        return d
+
+    return _canonical(a) == _canonical(b)
 
 
 class SandboxAdapter(ABC):
@@ -580,8 +620,7 @@ def _enforce_capability_gates(
             packages = dependencies.get("packages") if isinstance(dependencies, dict) else None
             if packages:
                 stored_deps = stored_config.get("dependencies") if stored_config else None
-                stored_pkgs = stored_deps.get("packages") if isinstance(stored_deps, dict) else None
-                if not _packages_equal(packages, stored_pkgs):
+                if not _packages_equal(dependencies, stored_deps):
                     errors.append(
                         InitErrorDetails(
                             type=PydanticCustomError(

tests/unit/server/api/mutations/test_sandbox_cache_invalidation.py

@@ -198,10 +198,6 @@ async def test_rotate_vercel_token_evicts_both_vercel_backends(
                 assert py_backend is not None
                 assert ts_backend is not None
 
-                # Confirm BOTH caches are populated pre-rotation.
-                assert any(k[0] == py_adapter.key for k in _BACKEND_CACHE)
-                assert any(k[0] == ts_adapter.key for k in _BACKEND_CACHE)
-
                 # Rotate via the PY backend_type only. The key-level fan-out
                 # (invalidate_backend_cache_for_key) must evict both because
                 # both adapters list `shared_spec_key` in credential_specs.
@@ -216,12 +212,19 @@ async def test_rotate_vercel_token_evicts_both_vercel_backends(
                 )
                 assert not result.errors, result.errors
 
-                # Both cache namespaces must be empty.
-                assert not any(k[0] == py_adapter.key for k in _BACKEND_CACHE), (
-                    "PY cache entries survived rotation"
+                # Both caches must have been evicted: next call returns a new instance.
+                async with db() as session:
+                    py_backend_v2 = await get_or_create_backend(
+                        py_adapter.key, config={}, session=session, decrypt=enc.decrypt
+                    )
+                    ts_backend_v2 = await get_or_create_backend(
+                        ts_adapter.key, config={}, session=session, decrypt=enc.decrypt
+                    )
+                assert py_backend_v2 is not py_backend, (
+                    "PY cache was not evicted: same instance returned after rotation"
                 )
-                assert not any(k[0] == ts_adapter.key for k in _BACKEND_CACHE), (
-                    "TS cache entries survived rotation — shared-spec fan-out failed"
+                assert ts_backend_v2 is not ts_backend, (
+                    "TS cache was not evicted — shared-spec fan-out failed"
                 )
         finally:
             _purge_cache_for([py_adapter.key, ts_adapter.key])

tests/unit/server/api/test_capability_advertisement.py

@@ -69,21 +69,12 @@ async def test_sandbox_backends_full_ui_query_shape(
     assert response.data is not None
     backends = {b["backendType"]: b for b in response.data["sandboxBackends"]}
 
-    assert set(backends.keys()) == {
-        "WASM",
-        "E2B",
-        "DAYTONA_PYTHON",
-        "VERCEL_PYTHON",
-        "VERCEL_TYPESCRIPT",
-        "DENO",
-        "MODAL",
-    }
+    assert set(backends.keys()) == set(SANDBOX_ADAPTER_METADATA.keys())
 
     for bt, backend in backends.items():
         assert "supportsEnvVars" in backend, bt
         assert "internetAccess" in backend, bt
         assert "dependenciesLanguage" in backend, bt
-        assert backend["internetAccess"] == "NONE", bt
 
 
 @pytest.mark.parametrize("backend_type", list(SANDBOX_ADAPTER_METADATA.keys()))
@@ -213,17 +204,8 @@ async def test_dependencies_language_only_set_for_daytona(
     assert response.data is not None
     backends = {b["backendType"]: b for b in response.data["sandboxBackends"]}
 
-    assert backends["DAYTONA_PYTHON"]["dependenciesLanguage"] == "PYTHON"
-
-    no_deps_backends = [
-        "WASM",
-        "E2B",
-        "VERCEL_PYTHON",
-        "VERCEL_TYPESCRIPT",
-        "DENO",
-        "MODAL",
-    ]
-    for bt in no_deps_backends:
-        assert backends[bt]["dependenciesLanguage"] is None, (
-            f"{bt} unexpectedly advertises a dependenciesLanguage"
+    for bt, meta in SANDBOX_ADAPTER_METADATA.items():
+        expected = meta.dependencies_language
+        assert backends[bt]["dependenciesLanguage"] == expected, (
+            f"{bt}: expected dependenciesLanguage={expected!r}"
         )