Skip to content

Commit b01e9e0

Browse files
feat(growth): turn suggestion backend
Classify each completed PostHog AI turn and publish an end-of-turn suggestion frame. The task runs on a new posthog_ai Celery queue, a short Redis lock drops duplicate turn reports before they are queued, and the offer ledger in the task state is the only stored outcome of a card: a reload reads it from the new state endpoint. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent ec508ef commit b01e9e0

25 files changed

Lines changed: 3229 additions & 5 deletions

‎bin/celery-queues.env‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@
22
# Important: Add new queues to make Celery consume tasks from them.
33

44
# NOTE: Keep in sync with posthog/tasks/utils.py
5-
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
5+
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‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,3 +42,4 @@ class CeleryQueue(Enum):
4242
INTEGRATIONS = "integrations"
4343
FEATURE_FLAGS = "feature_flags"
4444
FEATURE_FLAGS_LONG_RUNNING = "feature_flags_long_running"
45+
POSTHOG_AI = "posthog_ai"

‎posthog/egress/typesafe/README.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,11 @@ Raise both settings when real traffic outgrows them.
4444
The default reserve ladder applies, and `typesafe_request` defaults to `NORMAL`.
4545
`typesafe_request` rejects `CRITICAL`, because a `CRITICAL` call is never shed and would skip the hourly spend ceiling.
4646
Give every caller an explicit lane: `NORMAL` when a person waits for the answer, `BATCH` for background work.
47-
No caller exists on master yet. Each new caller adds itself here with its lane and its feature flag.
47+
Each new caller adds itself here with its lane and its feature flag.
48+
49+
| Caller | Source | Lane | Feature flag |
50+
| ---------------------------------------------------------------------------------------- | ----------------------------- | ------- | ----------------------------- |
51+
| PostHog AI turn suggestions (`products/posthog_ai/backend/turn_suggestions/judgment.py`) | `posthog_ai_turn_suggestions` | `BATCH` | `posthog-ai-turn-suggestions` |
4852

4953
## Rate-limit headers
5054

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
from .mcp_tools import MCPToolsViewSet
2+
from .turn_suggestions import TurnSuggestionsViewSet
23

3-
__all__ = ["MCPToolsViewSet"]
4+
__all__ = ["MCPToolsViewSet", "TurnSuggestionsViewSet"]
Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
from drf_spectacular.utils import OpenApiResponse
2+
from rest_framework import serializers
3+
from rest_framework.decorators import action
4+
from rest_framework.exceptions import NotFound
5+
from rest_framework.response import Response
6+
from rest_framework.viewsets import GenericViewSet
7+
8+
from posthog.api.mixins import ValidatedRequest, validated_request
9+
from posthog.api.routing import TeamAndOrgViewSetMixin
10+
11+
from products.posthog_ai.backend.turn_suggestions.offer_ledger import TurnSuggestionResolution
12+
from products.posthog_ai.backend.turn_suggestions.service import resolve_turn_suggestion, turn_suggestion_state
13+
from products.tasks.backend.facade.api import task_visible
14+
15+
16+
class ResolveTurnSuggestionSerializer(serializers.Serializer):
17+
task_id = serializers.UUIDField(
18+
help_text="ID of the PostHog AI conversation (task) the suggestion card belongs to."
19+
)
20+
turn_index = serializers.IntegerField(
21+
min_value=0, help_text="Zero-based index of the conversation turn the suggestion card was shown under."
22+
)
23+
resolution = serializers.ChoiceField(
24+
choices=TurnSuggestionResolution.choices,
25+
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.",
26+
)
27+
28+
29+
class ResolveTurnSuggestionResponseSerializer(serializers.Serializer):
30+
recorded = serializers.BooleanField(
31+
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."
32+
)
33+
34+
35+
class TurnSuggestionStateQuerySerializer(serializers.Serializer):
36+
task_id = serializers.UUIDField(
37+
help_text="ID of the PostHog AI conversation (task) to read suggestion outcomes for."
38+
)
39+
40+
41+
class TurnSuggestionStateSerializer(serializers.Serializer):
42+
muted = serializers.BooleanField(
43+
help_text="Whether the user dismissed a suggestion card in this conversation, which stops further cards."
44+
)
45+
resolved_turns = serializers.ListField(
46+
child=serializers.IntegerField(min_value=0),
47+
help_text="Zero-based indexes of the turns whose suggestion card the user already dismissed or accepted.",
48+
)
49+
50+
51+
class TurnSuggestionsViewSet(TeamAndOrgViewSetMixin, GenericViewSet):
52+
scope_object = "task"
53+
54+
@validated_request(
55+
query_serializer=TurnSuggestionStateQuerySerializer,
56+
responses={200: OpenApiResponse(response=TurnSuggestionStateSerializer)},
57+
summary="Read which PostHog AI turn suggestion cards the user already resolved",
58+
)
59+
@action(detail=False, methods=["get"], required_scopes=["task:read"])
60+
def state(self, request: ValidatedRequest, *args, **kwargs) -> Response:
61+
task_id = str(request.validated_query_data["task_id"])
62+
if not task_visible(task_id, self.team_id, request.user.id):
63+
raise NotFound()
64+
state = turn_suggestion_state(task_id, self.team_id)
65+
return Response(
66+
TurnSuggestionStateSerializer({"muted": state.muted, "resolved_turns": list(state.resolved_turns)}).data
67+
)
68+
69+
@validated_request(
70+
request_serializer=ResolveTurnSuggestionSerializer,
71+
responses={200: OpenApiResponse(response=ResolveTurnSuggestionResponseSerializer)},
72+
summary="Record what the user did with a PostHog AI turn suggestion card",
73+
)
74+
@action(detail=False, methods=["post"], required_scopes=["task:write"])
75+
def resolve(self, request: ValidatedRequest, *args, **kwargs) -> Response:
76+
task_id = str(request.validated_data["task_id"])
77+
# A dismissal mutes the conversation for its creator, so a teammate who can only read it must not resolve cards.
78+
if not task_visible(task_id, self.team_id, request.user.id, for_control=True):
79+
raise NotFound()
80+
recorded = resolve_turn_suggestion(
81+
task_id,
82+
self.team_id,
83+
turn_index=request.validated_data["turn_index"],
84+
resolution=TurnSuggestionResolution(request.validated_data["resolution"]),
85+
)
86+
return Response(ResolveTurnSuggestionResponseSerializer({"recorded": recorded}).data)

‎products/posthog_ai/backend/receivers.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
"""Keep a copied chat and its task in step once the task runs for real.
22
3-
Both hooks connect through the tasks facade so no model class crosses the product boundary.
3+
Every hook connects through the tasks facade so no model class crosses the product boundary.
44
55
- Before a run starts: the copy runs a few seconds behind the chat, so a run started in that window
66
would resume without the last turn. The guard copies what is still missing and refuses the run
@@ -10,6 +10,7 @@
1010
path too, so a chat can never fork back into LangGraph after it has been continued as a task.
1111
- When tasks are read: a user without the sandbox runtime does not see the copies of their chats. The
1212
copy is written for everyone so the switch is instant later, but until then their chats are chats.
13+
- After a turn completes: the end-of-turn suggestion is queued (``turn_suggestions/dispatch.py``).
1314
"""
1415

1516
import logging
@@ -22,9 +23,11 @@
2223
from posthog.models import Team, User
2324

2425
from products.posthog_ai.backend.models.assistant import Conversation
26+
from products.posthog_ai.backend.turn_suggestions.dispatch import enqueue_turn_suggestion_on_turn_completed
2527
from products.tasks.backend.facade.task_run_signals import (
2628
TaskOriginProduct,
2729
connect_task_run_post_save,
30+
connect_task_run_turn_completed,
2831
register_task_read_exclusion,
2932
register_task_run_start_guard,
3033
)
@@ -51,6 +54,9 @@ def connect() -> None:
5154
connect_task_run_post_save(
5255
mark_conversation_sandbox_on_first_run, dispatch_uid="posthog_ai_conversation_follows_task_run"
5356
)
57+
connect_task_run_turn_completed(
58+
enqueue_turn_suggestion_on_turn_completed, dispatch_uid="posthog_ai_turn_suggestion_on_turn_completed"
59+
)
5460

5561

5662
def copied_chats_hidden_without_sandbox_mode(team_id: int, user_id: int | None) -> Iterable[UUID]:

‎products/posthog_ai/backend/routes.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
from posthog.api.routing import RouterRegistry
22

3-
from products.posthog_ai.backend.api import MCPToolsViewSet
3+
from products.posthog_ai.backend.api import MCPToolsViewSet, TurnSuggestionsViewSet
44
from products.posthog_ai.backend.presentation.terminal_ai import TerminalAIViewSet
55

66

@@ -12,3 +12,9 @@ def register_routes(routers: RouterRegistry) -> None:
1212
"project_mcp_tools",
1313
["team_id"],
1414
)
15+
routers.projects.register(
16+
r"turn_suggestions",
17+
TurnSuggestionsViewSet,
18+
"project_turn_suggestions",
19+
["team_id"],
20+
)
Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
from celery import shared_task
2+
3+
from posthog.celery_queues import CeleryQueue
4+
5+
from products.posthog_ai.backend.turn_suggestions.service import generate_turn_suggestion
6+
7+
8+
# No retries: a suggestion that misses its turn is worth less than a stale one arriving mid-conversation.
9+
@shared_task(ignore_result=True, queue=CeleryQueue.POSTHOG_AI.value, soft_time_limit=90, time_limit=120)
10+
def generate_turn_suggestion_task(*, run_id: str, team_id: int) -> None:
11+
generate_turn_suggestion(run_id, team_id)

0 commit comments

Comments
 (0)