feat(signals): share the scout creation check between endpoint and facade

products/signals/backend/facade/api.py

@@ -16,10 +16,11 @@
 from posthog.dataclasses import frozen
 from posthog.event_usage import groups
 from posthog.helpers.tiktoken_encoding import LLM_TOKEN_COUNT_PROXY_MODEL, get_tiktoken_encoding_for_model
-from posthog.models import Team
+from posthog.models import Team, User
 from posthog.sync import database_sync_to_async
 from posthog.temporal.common.client import async_connect
 
+from products.access_control.backend.facade.user_access_control import UserAccessControl
 from products.signals.backend.artefact_schemas import (
     # Re-exported so the Slack mention handler can label the task it starts from a report's
     # notification thread without naming the relationship vocabulary itself.
@@ -29,6 +30,7 @@
 from products.signals.backend.enums import SIGNAL_SOURCE_PRODUCT_LABELS, SignalSourceProduct
 from products.signals.backend.models import SignalReport, SignalScoutConfig, SignalScoutRun, SignalSourceConfig
 from products.signals.backend.report_actionability_repair import RepairedBatch, repair_latest_actionability
+from products.signals.backend.scout_harness.create_access import can_create_scout
 from products.signals.backend.scout_harness.run_gates import (
     # Re-exported so the workflows endpoint can branch on why a fire was refused without reaching
     # into the scout harness. Every decision behind them stays Signals-side.
@@ -563,7 +565,7 @@
     return props
 
 
 async def emit_signal(
     team: Team,
     source_product: str,
     source_type: str,
@@ -1018,7 +1020,7 @@
     charts: list[dict[str, Any]]
 
 
 def scout_reports_for_source(
     team_id: int,
     source_product: str,
     source_id: str,
@@ -1171,3 +1173,29 @@
     artefact write, so this only repairs rows that drifted.
     """
     return repair_latest_actionability(team_id=team_id, batch_size=batch_size, after=after)
+
+
+def scout_creation_available(*, team_id: int, user_id: int) -> bool:
+    """Whether to offer the user scout creation on the team's project.
+
+    The create endpoint's permission checks run on the requested team: project access and editor
+    access to skills. Two more checks run on the canonical project: it runs scouts (enrollment in the
+    `signals-scout` flag payload), and the user passes the endpoint's own check (editor access to skills).
+    """
+    from products.signals.backend.scout_harness.team_limits import (
+        team_is_enrolled,  # noqa: PLC0415 — keeps the flag-reading harness module off the facade import path
+    )
+
+    team = Team.objects.select_related("parent_team").filter(id=team_id).first()
+    user = User.objects.filter(id=user_id, is_active=True).first()
+    if team is None or user is None:
+        return False
+    requested_team_access = UserAccessControl(user=user, team=team)
+    if not requested_team_access.has_project_access:
+        return False
+    if not requested_team_access.check_access_level_for_resource("llm_skill", "editor"):
+        return False
+    canonical_team = team.parent_team or team
+    if not team_is_enrolled(canonical_team.id):
+        return False
+    return can_create_scout(user, canonical_team)

products/signals/backend/scout_harness/create_access.py

@@ -0,0 +1,9 @@
+from posthog.models import Team, User
+
+from products.access_control.backend.facade.user_access_control import UserAccessControl
+
+
+def can_create_scout(user: User, canonical_team: Team) -> bool:
+    """Whether ``user`` may create a scout on ``canonical_team``: editor access to skills, because the
+    skill body is the prompt the scout agent runs. Access to the project itself is a separate check."""
+    return UserAccessControl(user=user, team=canonical_team).check_access_level_for_resource("llm_skill", "editor")

products/signals/backend/scout_harness/team_limits.py

@@ -560,6 +560,11 @@ def _resolve_enrolled(canonical_team_id: int, enrollment: Enrollment) -> bool:
     return _is_team_enrolled(canonical_team_id, enrollment.explicit)
 
 
+def team_is_enrolled(canonical_team_id: int) -> bool:
+    """Whether a canonical project runs scouts, as the `signals-scout` flag payload says right now."""
+    return _resolve_enrolled(canonical_team_id, _parse_enrollment(_read_flag_payload()))
+
+
 @dataclass(frozen=True)
 class ScoutTeamLimits:
     """A team's effective scout run caps + current usage, all resolved the way dispatch enforces."""

products/signals/backend/scout_harness/views.py

@@ -73,6 +73,7 @@
 from products.signals.backend.report_charts import ChartSize
 from products.signals.backend.report_generation.resolve_reviewers import MAX_PROJECT_MEMBERS, list_project_members
 from products.signals.backend.scout_harness.config_registry import enabled_scout_count, ensure_scout_category
+from products.signals.backend.scout_harness.create_access import can_create_scout
 from products.signals.backend.scout_harness.deprecation import deprecation_metadata_of
 from products.signals.backend.scout_harness.fleet_sync import materialize_scout_fleet
 from products.signals.backend.scout_harness.lazy_seed import (
@@ -1473,7 +1474,7 @@
         required_scopes=["signal_scout_internal:write"],
         pagination_class=None,
     )
     def lighthouse_audit(self, request: Request, **kwargs) -> Response:
         run_id = _parse_run_id_or_404(kwargs)
         run = (
             SignalScoutRun.objects.select_related("task_run")
@@ -2930,8 +2931,7 @@
     pagination_class = None
 
     def _assert_can_create_scout(self, *, user: User, canonical_team: Team) -> None:
-        access = UserAccessControl(user=user, team=canonical_team)
-        if not access.check_access_level_for_resource("llm_skill", "editor"):
+        if not can_create_scout(user, canonical_team):
             raise exceptions.PermissionDenied("Creating a scout requires editor access to skills.")
 
     @validated_request(

products/signals/backend/test/test_facade_scout_creation_available.py

@@ -0,0 +1,50 @@
+from posthog.test.base import BaseTest
+from unittest.mock import patch
+
+from parameterized import parameterized
+
+from posthog.models import Team
+
+from products.signals.backend.facade.api import scout_creation_available
+
+TEAM_LIMITS = "products.signals.backend.scout_harness.team_limits"
+CREATE_ACCESS = "products.signals.backend.scout_harness.create_access"
+FACADE = "products.signals.backend.facade.api"
+
+
+class TestScoutCreationAvailable(BaseTest):
+    @parameterized.expand(
+        [
+            ("enrolled_team", lambda team_id: {"guaranteed_team_ids": [team_id]}, False, True),
+            ("wildcard", lambda _team_id: {"guaranteed_team_ids": ["*"]}, False, True),
+            ("skipped_team", lambda team_id: {"guaranteed_team_ids": ["*"], "skip_team_ids": [team_id]}, False, False),
+            ("other_team_only", lambda team_id: {"guaranteed_team_ids": [team_id + 1000]}, False, False),
+            ("child_environment_of_enrolled_project", lambda team_id: {"guaranteed_team_ids": [team_id]}, True, True),
+        ]
+    )
+    def test_follows_the_scout_enrollment_payload(self, _name, payload_for, on_child_environment: bool, expected: bool):
+        team_id = self.team.id
+        if on_child_environment:
+            team_id = Team.objects.create(organization=self.organization, parent_team=self.team, name="Environment").id
+        with patch(f"{TEAM_LIMITS}._read_flag_payload", return_value=payload_for(self.team.id)):
+            assert scout_creation_available(team_id=team_id, user_id=self.user.id) is expected
+
+    def test_requires_skill_editor_access_like_the_create_endpoint(self):
+        with (
+            patch(f"{TEAM_LIMITS}._read_flag_payload", return_value={"guaranteed_team_ids": ["*"]}),
+            patch(f"{CREATE_ACCESS}.UserAccessControl") as user_access_control,
+        ):
+            user_access_control.return_value.check_access_level_for_resource.return_value = False
+            assert scout_creation_available(team_id=self.team.id, user_id=self.user.id) is False
+
+    @parameterized.expand([("project_access", "has_project_access"), ("skill_editor_access", "skill_editor")])
+    def test_requires_the_create_endpoint_access_on_the_requested_team(self, _name: str, denied: str):
+        with (
+            patch(f"{TEAM_LIMITS}._read_flag_payload", return_value={"guaranteed_team_ids": ["*"]}),
+            patch(f"{FACADE}.UserAccessControl") as user_access_control,
+        ):
+            access = user_access_control.return_value
+            access.has_project_access = denied != "has_project_access"
+            access.check_access_level_for_resource.return_value = denied != "skill_editor"
+            assert scout_creation_available(team_id=self.team.id, user_id=self.user.id) is False
+        assert user_access_control.call_args.kwargs["team"].id == self.team.id

products/signals/backend/test/test_scout_create_api.py

@@ -315,7 +315,7 @@ def test_environment_create_requires_skill_editor_access_on_parent(self) -> None
             parent_team=self.team,
             name="Child environment",
         )
-        with patch("products.signals.backend.scout_harness.views.UserAccessControl") as user_access_control:
+        with patch("products.signals.backend.scout_harness.create_access.UserAccessControl") as user_access_control:
             user_access_control.return_value.check_access_level_for_resource.return_value = False
             response = self.client.post(
                 f"/api/projects/{environment.id}/signals/scout/",