feat(growth): turn suggestion backend

bin/celery-queues.env

@@ -2,4 +2,4 @@
 # Important: Add new queues to make Celery consume tasks from them.
 
 # NOTE: Keep in sync with posthog/tasks/utils.py
-CELERY_WORKER_QUEUES=ai_research_privacy,celery,stats,email,analytics_queries,analytics_limited,long_running,exports,subscription_delivery,usage_reports,integrations,feature_flags,feature_flags_long_running
+CELERY_WORKER_QUEUES=ai_research_privacy,celery,stats,email,analytics_queries,analytics_limited,long_running,exports,subscription_delivery,usage_reports,integrations,feature_flags,feature_flags_long_running,posthog_ai

posthog/celery_queues.py

@@ -42,3 +42,4 @@ class CeleryQueue(Enum):
     INTEGRATIONS = "integrations"
     FEATURE_FLAGS = "feature_flags"
     FEATURE_FLAGS_LONG_RUNNING = "feature_flags_long_running"
+    POSTHOG_AI = "posthog_ai"

products/posthog_ai/backend/api/__init__.py

@@ -1,3 +1,4 @@
 from .mcp_tools import MCPToolsViewSet
+from .turn_suggestions import TurnSuggestionsViewSet
 
-__all__ = ["MCPToolsViewSet"]
+__all__ = ["MCPToolsViewSet", "TurnSuggestionsViewSet"]

products/posthog_ai/backend/api/turn_suggestions.py

@@ -0,0 +1,86 @@
+from drf_spectacular.utils import OpenApiResponse
+from rest_framework import serializers
+from rest_framework.decorators import action
+from rest_framework.exceptions import NotFound
+from rest_framework.response import Response
+from rest_framework.viewsets import GenericViewSet
+
+from posthog.api.mixins import ValidatedRequest, validated_request
+from posthog.api.routing import TeamAndOrgViewSetMixin
+
+from products.posthog_ai.backend.turn_suggestions.offer_ledger import TurnSuggestionResolution
+from products.posthog_ai.backend.turn_suggestions.service import resolve_turn_suggestion, turn_suggestion_state
+from products.tasks.backend.facade.api import task_visible
+
+
+class ResolveTurnSuggestionSerializer(serializers.Serializer):
+    task_id = serializers.UUIDField(
+        help_text="ID of the PostHog AI conversation (task) the suggestion card belongs to."
+    )
+    turn_index = serializers.IntegerField(
+        min_value=0, help_text="Zero-based index of the conversation turn the suggestion card was shown under."
+    )
+    resolution = serializers.ChoiceField(
+        choices=TurnSuggestionResolution.choices,
+        help_text="What the user did with the card: `dismissed` mutes suggestions for the rest of the conversation, `accepted` means the offered scout, notebook, alert or subscription was created.",
+    )
+
+
+class ResolveTurnSuggestionResponseSerializer(serializers.Serializer):
+    recorded = serializers.BooleanField(
+        help_text="Whether a suggestion card existed for that turn and this call recorded its outcome. A card keeps the first outcome recorded for it."
+    )
+
+
+class TurnSuggestionStateQuerySerializer(serializers.Serializer):
+    task_id = serializers.UUIDField(
+        help_text="ID of the PostHog AI conversation (task) to read suggestion outcomes for."
+    )
+
+
+class TurnSuggestionStateSerializer(serializers.Serializer):
+    muted = serializers.BooleanField(
+        help_text="Whether the user dismissed a suggestion card in this conversation, which stops further cards."
+    )
+    resolved_turns = serializers.ListField(
+        child=serializers.IntegerField(min_value=0),
+        help_text="Zero-based indexes of the turns whose suggestion card the user already dismissed or accepted.",
+    )
+
+
+class TurnSuggestionsViewSet(TeamAndOrgViewSetMixin, GenericViewSet):
+    scope_object = "task"
+
+    @validated_request(
+        query_serializer=TurnSuggestionStateQuerySerializer,
+        responses={200: OpenApiResponse(response=TurnSuggestionStateSerializer)},
+        summary="Read which PostHog AI turn suggestion cards the user already resolved",
+    )
+    @action(detail=False, methods=["get"], required_scopes=["task:read"])
+    def state(self, request: ValidatedRequest, *args, **kwargs) -> Response:
+        task_id = str(request.validated_query_data["task_id"])
+        if not task_visible(task_id, self.team_id, request.user.id):
+            raise NotFound()
+        state = turn_suggestion_state(task_id, self.team_id)
+        return Response(
+            TurnSuggestionStateSerializer({"muted": state.muted, "resolved_turns": list(state.resolved_turns)}).data
+        )
+
+    @validated_request(
+        request_serializer=ResolveTurnSuggestionSerializer,
+        responses={200: OpenApiResponse(response=ResolveTurnSuggestionResponseSerializer)},
+        summary="Record what the user did with a PostHog AI turn suggestion card",
+    )
+    @action(detail=False, methods=["post"], required_scopes=["task:write"])
+    def resolve(self, request: ValidatedRequest, *args, **kwargs) -> Response:
+        task_id = str(request.validated_data["task_id"])
+        # A dismissal mutes the conversation for its creator, so a teammate who can only read it must not resolve cards.
+        if not task_visible(task_id, self.team_id, request.user.id, for_control=True):
+            raise NotFound()
+        recorded = resolve_turn_suggestion(
+            task_id,
+            self.team_id,
+            turn_index=request.validated_data["turn_index"],
+            resolution=TurnSuggestionResolution(request.validated_data["resolution"]),
+        )
+        return Response(ResolveTurnSuggestionResponseSerializer({"recorded": recorded}).data)

products/posthog_ai/backend/receivers.py

@@ -1,6 +1,6 @@
 """Keep a copied chat and its task in step once the task runs for real.
 
-Both hooks connect through the tasks facade so no model class crosses the product boundary.
+Every hook connects through the tasks facade so no model class crosses the product boundary.
 
 - Before a run starts: the copy runs a few seconds behind the chat, so a run started in that window
   would resume without the last turn. The guard copies what is still missing and refuses the run
@@ -10,6 +10,7 @@
   path too, so a chat can never fork back into LangGraph after it has been continued as a task.
 - When tasks are read: a user without the sandbox runtime does not see the copies of their chats. The
   copy is written for everyone so the switch is instant later, but until then their chats are chats.
+- After a turn completes: the end-of-turn suggestion is queued (``turn_suggestions/dispatch.py``).
 """
 
 import logging
@@ -22,9 +23,11 @@
 from posthog.models import Team, User
 
 from products.posthog_ai.backend.models.assistant import Conversation
+from products.posthog_ai.backend.turn_suggestions.dispatch import enqueue_turn_suggestion_on_turn_completed
 from products.tasks.backend.facade.task_run_signals import (
     TaskOriginProduct,
     connect_task_run_post_save,
+    connect_task_run_turn_completed,
     register_task_read_exclusion,
     register_task_run_start_guard,
 )
@@ -51,6 +54,9 @@ def connect() -> None:
     connect_task_run_post_save(
         mark_conversation_sandbox_on_first_run, dispatch_uid="posthog_ai_conversation_follows_task_run"
     )
+    connect_task_run_turn_completed(
+        enqueue_turn_suggestion_on_turn_completed, dispatch_uid="posthog_ai_turn_suggestion_on_turn_completed"
+    )
 
 
 def copied_chats_hidden_without_sandbox_mode(team_id: int, user_id: int | None) -> Iterable[UUID]:

products/posthog_ai/backend/routes.py

@@ -1,6 +1,6 @@
 from posthog.api.routing import RouterRegistry
 
-from products.posthog_ai.backend.api import MCPToolsViewSet
+from products.posthog_ai.backend.api import MCPToolsViewSet, TurnSuggestionsViewSet
 from products.posthog_ai.backend.presentation.terminal_ai import TerminalAIViewSet
 
 
@@ -12,3 +12,9 @@ def register_routes(routers: RouterRegistry) -> None:
         "project_mcp_tools",
         ["team_id"],
     )
+    routers.projects.register(
+        r"turn_suggestions",
+        TurnSuggestionsViewSet,
+        "project_turn_suggestions",
+        ["team_id"],
+    )

products/posthog_ai/backend/tasks.py

@@ -0,0 +1,29 @@
+import structlog
+from celery import shared_task
+from prometheus_client import Counter
+
+from posthog.celery_queues import CeleryQueue
+
+from products.posthog_ai.backend.turn_suggestions.service import generate_turn_suggestion
+
+logger = structlog.get_logger(__name__)
+
+TURN_SUGGESTION_OUTCOMES_TOTAL = Counter(
+    "posthog_ai_turn_suggestion_outcomes_total",
+    "End-of-turn suggestion jobs by outcome, so a gate that skips every turn in a region shows up",
+    labelnames=["status", "reason"],
+)
+
+
+# No retries: a suggestion that misses its turn is worth less than a stale one arriving mid-conversation.
+@shared_task(ignore_result=True, queue=CeleryQueue.POSTHOG_AI.value, soft_time_limit=90, time_limit=120)
+def generate_turn_suggestion_task(*, run_id: str, team_id: int) -> None:
+    outcome = generate_turn_suggestion(run_id, team_id)
+    TURN_SUGGESTION_OUTCOMES_TOTAL.labels(status=outcome.status, reason=outcome.reason).inc()
+    logger.info(
+        "posthog_ai_turn_suggestion_outcome",
+        run_id=run_id,
+        team_id=team_id,
+        status=outcome.status,
+        reason=outcome.reason,
+    )