fix(experiments): enforce object-level access in experiment Max tools

products/experiments/backend/max_tools.py

@@ -22,6 +22,7 @@
 
 from ee.hogai.context.experiment.context import ExperimentContext
 from ee.hogai.tool import MaxTool
+from ee.hogai.tool_errors import MaxToolAccessDeniedError
 
 CREATE_EXPERIMENT_TOOL_DESCRIPTION = dedent("""
     Use this tool to create A/B test experiments that measure the impact of changes.
@@ -248,6 +249,8 @@ async def _arun_impl(self, experiment_id: int | None = None) -> tuple[str, dict[
             # Otherwise, fetch data via the data service (agent-initiated call)
             return await self._fetch_and_format(resolved_experiment_id)
 
+        except MaxToolAccessDeniedError:
+            raise
         except Exception as e:
             capture_exception(
                 e,
@@ -266,6 +269,8 @@ async def _format_from_context(self, experiment_id: int, context: dict) -> tuple
         if experiment is None:
             return f"Experiment {experiment_id} not found", {"error": "not_found"}
 
+        await self.check_object_access(experiment, "viewer", resource="experiment", action="read")
+
         try:
             primary_metrics = [MaxExperimentMetricResult(**m) for m in context.get("primary_metrics_results", [])]
             secondary_metrics = [MaxExperimentMetricResult(**m) for m in context.get("secondary_metrics_results", [])]
@@ -293,6 +298,14 @@ async def _format_from_context(self, experiment_id: int, context: dict) -> tuple
 
     async def _fetch_and_format(self, experiment_id: int) -> tuple[str, dict[str, Any]]:
         """Fetch experiment data from query runners and format it."""
+        experiment_context = ExperimentContext(team=self._team, experiment_id=experiment_id)
+        experiment = await experiment_context.aget_experiment()
+        if experiment is None:
+            return f"Experiment {experiment_id} not found", {"error": "not_found"}
+
+        # Before the metric queries run, so a denied user costs no ClickHouse work
+        await self.check_object_access(experiment, "viewer", resource="experiment", action="read")
+
         data_service = ExperimentSummaryDataService(self._team, self._user)
 
         try:
@@ -302,11 +315,6 @@ async def _fetch_and_format(self, experiment_id: int) -> tuple[str, dict[str, An
 
         summary_context = summary_data.context
 
-        experiment_context = ExperimentContext(team=self._team, experiment_id=experiment_id)
-        experiment = await experiment_context.aget_experiment()
-        if experiment is None:
-            return f"Experiment {experiment_id} not found", {"error": "not_found"}
-
         formatted_data = await experiment_context.format_experiment_results_data(
             experiment,
             exposures=summary_context.exposures,
@@ -435,6 +443,8 @@ def get_experiment():
 
             experiment = await get_experiment()
 
+            await self.check_object_access(experiment, "viewer", resource="experiment", action="read")
+
             if experiment.is_draft:
                 output = SessionReplaySummaryOutput(
                     experiment_id=experiment_id,
@@ -538,6 +548,8 @@ def count_recordings(q):
 
             return user_message, output.model_dump()
 
+        except MaxToolAccessDeniedError:
+            raise
         except ValueError as e:
             return f"❌ {str(e)}", {"error": "validation_error", "details": str(e)}
         except Exception as e:

products/experiments/backend/test/test_max_tools.py

@@ -14,6 +14,7 @@
 from products.experiments.backend.models.experiment import Experiment
 from products.feature_flags.backend.models.feature_flag import FeatureFlag
 
+from ee.hogai.tool_errors import MaxToolAccessDeniedError
 from ee.hogai.utils.types import AssistantState
 
 
@@ -655,17 +656,28 @@ async def test_agent_initiated_with_experiment_id_argument(self):
     async def test_fetch_and_format_handles_nonexistent_experiment(self):
         tool = self._create_tool({})
 
-        with patch("products.experiments.backend.max_tools.ExperimentSummaryDataService") as mock_service_class:
-            mock_service = mock_service_class.return_value
-            mock_service.fetch_experiment_data = AsyncMock(
-                side_effect=ValueError("Experiment 99999 not found or access denied")
-            )
-
-            result, artifact = await tool._arun_impl(experiment_id=99999)
+        result, artifact = await tool._arun_impl(experiment_id=99999)
 
-        assert "not found or access denied" in result
+        assert "not found" in result
         assert artifact["error"] == "not_found"
 
+    async def test_denies_object_level_restricted_experiment(self):
+        experiment = await self._create_experiment(flag_key="restricted-test")
+
+        # Agent-initiated path: denied before any metric queries run
+        tool = self._create_tool({})
+        with patch.object(tool, "user_access_control") as mock_uac:
+            mock_uac.check_access_level_for_object.return_value = False
+            with self.assertRaises(MaxToolAccessDeniedError):
+                await tool._arun_impl(experiment_id=experiment.id)
+
+        # Frontend-context path
+        tool = self._create_tool(self._build_context(experiment_id=experiment.id))
+        with patch.object(tool, "user_access_control") as mock_uac:
+            mock_uac.check_access_level_for_object.return_value = False
+            with self.assertRaises(MaxToolAccessDeniedError):
+                await tool._arun_impl()
+
     async def test_context_experiment_id_takes_priority_over_argument(self):
         experiment = await self._create_experiment(
             name="Context Experiment",

products/experiments/backend/test/test_session_replay_summary_tool.py

@@ -11,6 +11,7 @@
 from products.experiments.backend.models.experiment import Experiment
 from products.feature_flags.backend.models.feature_flag import FeatureFlag
 
+from ee.hogai.tool_errors import MaxToolAccessDeniedError
 from ee.hogai.utils.types import AssistantState
 
 
@@ -174,6 +175,15 @@ async def test_experiment_with_no_recordings(self, mock_list_recordings):
         self.assertEqual(artifact["experiment_id"], experiment.id)
         self.assertEqual(artifact["recording_counts"], {"control": 0, "test": 0})
 
+    async def test_denies_object_level_restricted_experiment(self):
+        experiment = await self.acreate_experiment(name="restricted-replay-test")
+        tool = await self.create_tool()
+
+        with patch.object(tool, "user_access_control") as mock_uac:
+            mock_uac.check_access_level_for_object.return_value = False
+            with self.assertRaises(MaxToolAccessDeniedError):
+                await tool._arun_impl(experiment_id=experiment.id)
+
     async def test_experiment_not_found(self):
         """Test error when experiment ID doesn't exist."""
         tool = await self.create_tool()