feat(seer): Add trigger_explorer method to SeerOperator

src/sentry/seer/entrypoints/metrics.py

@@ -9,6 +9,7 @@
 
 class SeerOperatorInteractionType(StrEnum):
     OPERATOR_TRIGGER_AUTOFIX = "trigger_autofix"
+    OPERATOR_TRIGGER_EXPLORER = "trigger_explorer"
     OPERATOR_PROCESS_AUTOFIX_UPDATE = "process_autofix_update"
     OPERATOR_CACHE_POPULATE_PRE_AUTOFIX = "cache_populate_pre_autofix"
     OPERATOR_CACHE_POPULATE_POST_AUTOFIX = "cache_populate_post_autofix"
@@ -19,6 +20,9 @@ class SeerOperatorInteractionType(StrEnum):
     ENTRYPOINT_ON_TRIGGER_AUTOFIX_ALREADY_EXISTS = "entrypoint_on_trigger_autofix_already_exists"
     ENTRYPOINT_CREATE_AUTOFIX_CACHE_PAYLOAD = "entrypoint_create_autofix_cache_payload"
     ENTRYPOINT_ON_AUTOFIX_UPDATE = "entrypoint_on_autofix_update"
+    ENTRYPOINT_ON_TRIGGER_EXPLORER = "entrypoint_on_trigger_explorer"
+    ENTRYPOINT_CREATE_EXPLORER_CACHE_PAYLOAD = "entrypoint_create_explorer_cache_payload"
+    ENTRYPOINT_ON_EXPLORER_UPDATE = "entrypoint_on_explorer_update"
     OPERATOR_PROCESS_EXPLORER_COMPLETION = "process_explorer_completion"
     OPERATOR_CACHE_SET_EXPLORER = "cache_set_explorer"
     OPERATOR_CACHE_GET_EXPLORER = "cache_get_explorer"

src/sentry/seer/entrypoints/operator.py

@@ -28,9 +28,15 @@
     autofix_entrypoint_registry,
     explorer_entrypoint_registry,
 )
-from sentry.seer.entrypoints.types import SeerAutofixEntrypoint, SeerEntrypointKey
+from sentry.seer.entrypoints.types import (
+    SeerAutofixEntrypoint,
+    SeerEntrypointKey,
+    SeerExplorerEntrypoint,
+)
+from sentry.seer.explorer.client import SeerExplorerClient
 from sentry.seer.explorer.client_models import SeerRunState
 from sentry.seer.explorer.on_completion_hook import ExplorerOnCompletionHook
+from sentry.seer.models import SeerPermissionError
 from sentry.seer.seer_setup import has_seer_access
 from sentry.sentry_apps.metrics import SentryAppEventType
 from sentry.tasks.base import instrumented_task
@@ -438,6 +444,116 @@ def trigger_autofix_legacy(
             )
 
 
+class SeerExplorerOperator[CachePayloadT]:
+    """
+    A class that connects to entrypoint implementations and runs Explorer operations for Seer.
+    It does this to ensure all entrypoints have consistent behavior and responses.
+    """
+
+    def __init__(self, entrypoint: SeerExplorerEntrypoint[CachePayloadT]):
+        self.entrypoint = entrypoint
+
+    def trigger_explorer(
+        self,
+        *,
+        organization: Organization,
+        user: User | RpcUser | None,
+        prompt: str,
+        on_page_context: str | None = None,
+        category_key: str,
+        category_value: str,
+    ) -> int | None:
+        """
+        Start or continue an Explorer run and return the run_id.
+        If a run exists for this category (e.g. slack thread), continues it; otherwise starts new.
+        Uses the entrypoint's Explorer callbacks for success/error handling.
+        """
+        event_lifecycle = SeerOperatorEventLifecycleMetric(
+            interaction_type=SeerOperatorInteractionType.OPERATOR_TRIGGER_EXPLORER,
+            entrypoint_key=self.entrypoint.key,
+        )
+
+        with event_lifecycle.capture() as lifecycle:
+            lifecycle.add_extras(
+                {
+                    "category_key": category_key,
+                    "category_value": category_value,
+                }
+            )
+
+            try:
+                # RpcUser is not in SeerExplorerClient's type signature but works at runtime
+                client = SeerExplorerClient(
+                    organization=organization,
+                    user=user,
+                    category_key=category_key,
+                    category_value=category_value,
+                    on_completion_hook=SeerOperatorCompletionHook,
+                    is_interactive=True,
+                    enable_coding=False,
+                )
+            except SeerPermissionError as e:
+                with SeerOperatorEventLifecycleMetric(
+                    interaction_type=SeerOperatorInteractionType.ENTRYPOINT_ON_TRIGGER_EXPLORER,
+                    entrypoint_key=self.entrypoint.key,
+                ).capture(assume_success=False):
+                    self.entrypoint.on_trigger_explorer_error(error=str(e))
+                lifecycle.record_failure(failure_reason=e)
+                return None
+
+            try:
+                existing_runs = client.get_runs(
+                    category_key=category_key,
+                    category_value=category_value,
+                    limit=1,
+                    only_current_user=False,
+                )
+
+                if existing_runs:
+                    run_id = client.continue_run(
+                        run_id=existing_runs[0].run_id,
+                        prompt=prompt,
+                        on_page_context=on_page_context,
+                    )
+                    lifecycle.add_extra("continued", "true")
+                else:
+                    run_id = client.start_run(
+                        prompt=prompt,
+                        on_page_context=on_page_context,
+                    )
+                    lifecycle.add_extra("continued", "false")
+            except Exception as e:
+                with SeerOperatorEventLifecycleMetric(
+                    interaction_type=SeerOperatorInteractionType.ENTRYPOINT_ON_TRIGGER_EXPLORER,
+                    entrypoint_key=self.entrypoint.key,
+                ).capture(assume_success=False):
+                    self.entrypoint.on_trigger_explorer_error(error="An unexpected error occurred")
+                lifecycle.record_failure(failure_reason=e)
+                return None
+
+            lifecycle.add_extra("run_id", str(run_id))
+
+            with SeerOperatorEventLifecycleMetric(
+                interaction_type=SeerOperatorInteractionType.ENTRYPOINT_ON_TRIGGER_EXPLORER,
+                entrypoint_key=self.entrypoint.key,
+            ).capture():
+                self.entrypoint.on_trigger_explorer_success(run_id=run_id)
+
+            with SeerOperatorEventLifecycleMetric(
+                interaction_type=SeerOperatorInteractionType.ENTRYPOINT_CREATE_EXPLORER_CACHE_PAYLOAD,
+                entrypoint_key=self.entrypoint.key,
+            ).capture():
+                cache_payload = self.entrypoint.create_explorer_cache_payload()
+
+            SeerOperatorExplorerCache.set(
+                entrypoint_key=str(self.entrypoint.key),
+                run_id=run_id,
+                cache_payload=cache_payload,
+            )
+
+            return run_id
+
+
 @instrumented_task(
     name="sentry.seer.entrypoints.operator.process_autofix_updates",
     namespace=seer_tasks,
@@ -693,8 +809,22 @@ def execute(cls, organization: Organization, run_id: int) -> None:
                 if not cache_payload:
                     continue
 
-                entrypoint_cls.on_explorer_update(
-                    cache_payload=cache_payload,
-                    summary=summary,
-                    run_id=run_id,
-                )
+                if cache_payload.get("organization_id") != organization.id:
+                    # run_id is globally unique in Seer, so only one entrypoint will
+                    # have a cache entry per run. An org mismatch here is anomalous;
+                    # return rather than continue to abort the entire method.
+                    lifecycle.record_failure(failure_reason="org_mismatch")
+                    return
+
+                with SeerOperatorEventLifecycleMetric(
+                    interaction_type=SeerOperatorInteractionType.ENTRYPOINT_ON_EXPLORER_UPDATE,
+                    entrypoint_key=str(entrypoint_key),
+                ).capture() as ept_lifecycle:
+                    try:
+                        entrypoint_cls.on_explorer_update(
+                            cache_payload=cache_payload,
+                            summary=summary,
+                            run_id=run_id,
+                        )
+                    except Exception as e:
+                        ept_lifecycle.record_failure(failure_reason=e)

src/sentry/seer/explorer/client.py

@@ -35,6 +35,7 @@
 from sentry.seer.seer_setup import has_seer_access_with_detail
 from sentry.seer.signed_seer_api import SeerViewerContext
 from sentry.users.models.user import User
+from sentry.users.services.user import RpcUser
 
 logger = logging.getLogger(__name__)
 
@@ -170,7 +171,7 @@ def execute(cls, organization: Organization, run_id: int) -> None:
 
         Args:
             organization: Sentry organization
-            user: User for permission checks and user-specific context (can be User, AnonymousUser, or None)
+            user: User for permission checks and user-specific context (can be User, RpcUser, AnonymousUser, or None)
             project: Optional project for project-scoped runs (e.g. autofix for an issue)
             category_key: Optional category key for filtering/grouping runs (e.g., "bug-fixer", "trace-analyzer"). Must be provided together with category_value. Makes it easy to retrieve runs for your feature later.
             category_value: Optional category value for filtering/grouping runs (e.g., issue ID, trace ID). Must be provided together with category_key. Makes it easy to retrieve a specific run for your feature later.
@@ -185,7 +186,7 @@ def execute(cls, organization: Organization, run_id: int) -> None:
     def __init__(
         self,
         organization: Organization,
-        user: User | AnonymousUser | None = None,
+        user: User | RpcUser | AnonymousUser | None = None,
         project: Project | None = None,
         category_key: str | None = None,
         category_value: str | None = None,

src/sentry/seer/explorer/client_utils.py

@@ -189,7 +189,7 @@ def has_seer_explorer_access_with_detail(
 
 
 def collect_user_org_context(
-    user: SentryUser | AnonymousUser | None,
+    user: SentryUser | RpcUser | AnonymousUser | None,
     organization: Organization,
     request: Request | None = None,
 ) -> dict[str, Any]:
@@ -235,7 +235,7 @@ def collect_user_org_context(
 
     # Handle name attribute - SentryUser has name
     user_name: str | None = None
-    if isinstance(user, SentryUser):
+    if isinstance(user, (SentryUser, RpcUser)):
         user_name = user.name
 
     # Get user's timezone setting (IANA timezone name, e.g., "America/Los_Angeles")

tests/sentry/seer/entrypoints/test_operator.py

@@ -25,6 +25,7 @@
 )
 from sentry.seer.explorer.client_models import MemoryBlock, Message, RepoPRState, SeerRunState
 from sentry.sentry_apps.metrics import SentryAppEventType
+from sentry.testutils.asserts import assert_failure_metric
 from sentry.testutils.cases import TestCase
 
 
@@ -774,7 +775,7 @@ def test_execute_skips_entrypoint_without_access(self, mock_fetch):
         mock_no_access.has_access.return_value = False
         mock_has_access = Mock(spec=SeerExplorerEntrypoint)
         mock_has_access.has_access.return_value = True
-        cache_payload = {"thread_id": "abc"}
+        cache_payload = {"thread_id": "abc", "organization_id": self.organization.id}
 
         with (
             patch.dict(
@@ -816,6 +817,28 @@ def test_execute_skips_entrypoint_without_cache(self, mock_fetch):
 
         mock_entrypoint_cls.on_explorer_update.assert_not_called()
 
+    @patch("sentry.integrations.utils.metrics.EventLifecycle.record_event")
+    @patch("sentry.seer.explorer.client_utils.fetch_run_status")
+    def test_execute_records_failure_on_org_mismatch(self, mock_fetch, mock_record):
+        state = self._make_state(
+            blocks=[
+                MemoryBlock(
+                    id="1",
+                    message=Message(role="assistant", content="summary"),
+                    timestamp="2024-01-01T00:00:00Z",
+                ),
+            ]
+        )
+        other_org = self.create_organization()
+        mock_entrypoint_cls = self._execute_with_mock_entrypoint(
+            mock_fetch,
+            state,
+            cache_return_value={"thread_id": "abc", "organization_id": other_org.id},
+        )
+
+        mock_entrypoint_cls.on_explorer_update.assert_not_called()
+        assert_failure_metric(mock_record, "org_mismatch")
+
     @patch("sentry.seer.explorer.client_utils.fetch_run_status")
     def test_execute_with_empty_blocks(self, mock_fetch):
         state = self._make_state(blocks=[])