@@ -63,8 +63,14 @@ def wrapper(
         nonlocal cache_key, cache_value
 
         kwargs_key = tuple(sorted(kwargs.items())) if kwargs else ()
-        key = tuple(
-            [id(registry_proto), registry_proto.version_id, project, tags, kwargs_key]
+        # Snapshot the filter so caller mutations cannot change a cached key.
+        tags_key = frozenset(tags.items()) if tags is not None else None
+        key = (
+            id(registry_proto),
+            registry_proto.version_id,
+            project,
+            tags_key,
+            kwargs_key,
         )
 
         if key == cache_key:
 
@@ -1,11 +1,38 @@
-from feast.infra.registry.proto_registry_utils import list_saved_datasets
+from unittest.mock import patch
+
+from feast import Entity
+from feast.infra.registry.proto_registry_utils import list_entities, list_saved_datasets
 from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
 from feast.protos.feast.core.SavedDataset_pb2 import SavedDataset as SavedDatasetProto
 from feast.protos.feast.core.SavedDataset_pb2 import (
     SavedDatasetSpec,
     SavedDatasetStorage,
 )
 from feast.saved_dataset import SavedDataset
+from feast.value_type import ValueType
+
+
+def test_equivalent_tag_filters_reuse_cached_entities() -> None:
+    registry = RegistryProto(version_id="test_version")
+    entity = Entity(
+        name="driver",
+        value_type=ValueType.INT64,
+        tags={"environment": "production", "team": "ml"},
+    ).to_proto()
+    entity.spec.project = "test_project"
+    registry.entities.append(entity)
+
+    with patch.object(Entity, "from_proto", wraps=Entity.from_proto) as from_proto:
+        first = list_entities(
+            registry, "test_project", {"environment": "production", "team": "ml"}
+        )
+        second = list_entities(
+            registry, "test_project", {"team": "ml", "environment": "production"}
+        )
+
+        assert [obj.name for obj in first] == ["driver"]
+        assert [obj.name for obj in second] == ["driver"]
+        from_proto.assert_called_once()
 
 
 class TestRegistryProto:
 
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from feast import Entity, FeatureStore, FeatureView, Field
+from feast.infra.offline_stores.file_source import FileSource
+from feast.infra.registry.registry import Registry
+from feast.infra.registry.sql import SqlRegistry
+from feast.repo_config import RepoConfig
+from feast.types import Int64
+from feast.value_type import ValueType
+
+
+@pytest.mark.parametrize("registry_type", ["file", "sql"])
+@pytest.mark.parametrize("resource_type", ["entities", "feature_views"])
+@pytest.mark.parametrize(
+    "initial_tags,updated_tags,initial_names,updated_names",
+    [
+        (
+            {"environment": "production"},
+            {"environment": "staging"},
+            {"production"},
+            {"staging"},
+        ),
+        ({}, {"environment": "production"}, {"production", "staging"}, {"production"}),
+        ({"environment": "production"}, {}, {"production"}, {"production", "staging"}),
+    ],
+    ids=["change-value", "add-tag", "remove-tag"],
+)
+def test_cached_list_uses_current_tag_filter(
+    tmp_path: Path,
+    registry_type: str,
+    resource_type: str,
+    initial_tags: dict[str, str],
+    updated_tags: dict[str, str],
+    initial_names: set[str],
+    updated_names: set[str],
+) -> None:
+    """Reusing a tag dictionary must not return results for its previous contents."""
+    registry_path = tmp_path / "registry.db"
+    registry_config = (
+        {"path": str(registry_path), "cache_ttl_seconds": 0}
+        if registry_type == "file"
+        else {
+            "registry_type": "sql",
+            "path": f"sqlite:///{registry_path}",
+            "cache_ttl_seconds": 0,
+        }
+    )
+    store = FeatureStore(
+        config=RepoConfig(
+            project="test_project", registry=registry_config, provider="local"
+        )
+    )
+    try:
+        for environment in ("production", "staging"):
+            tags = {"environment": environment}
+            entity = Entity(name=environment, value_type=ValueType.INT64, tags=tags)
+            store.registry.apply_entity(entity, store.project)
+            store.registry.apply_feature_view(
+                FeatureView(
+                    name=environment,
+                    entities=[entity],
+                    schema=[
+                        Field(name=environment, dtype=Int64),
+                        Field(name="value", dtype=Int64),
+                    ],
+                    source=FileSource(
+                        path="unused.parquet", timestamp_field="event_timestamp"
+                    ),
+                    tags=tags,
+                ),
+                store.project,
+            )
+        store.refresh_registry()
+
+        def list_names(tags: dict[str, str]) -> set[str]:
+            if resource_type == "entities":
+                return {
+                    obj.name for obj in store.list_entities(allow_cache=True, tags=tags)
+                }
+            return {
+                obj.name
+                for obj in store.list_feature_views(allow_cache=True, tags=tags)
+            }
+
+        query_tags = dict(initial_tags)
+        assert list_names(query_tags) == initial_names
+
+        query_tags.clear()
+        query_tags.update(updated_tags)
+        assert list_names(query_tags) == updated_names
+        # A separate caller with an equivalent filter must also see correct results.
+        assert list_names(dict(updated_tags)) == updated_names
+    finally:
+        registry = store.registry
+        assert isinstance(registry, (Registry, SqlRegistry))
+        registry.teardown()