feat(signals): start the personal ranking build when a first activity summary lands

products/signals/backend/activity_summary/tick.py

@@ -65,6 +65,7 @@
 from products.signals.backend.activity_summary.render import DEFAULT_LIMITS, RENDER_VERSION, RenderLimits
 from products.signals.backend.activity_summary.summaries import DEFAULT_MODEL, PIPELINE_DISTINCT_ID
 from products.signals.backend.models import UserActivitySummary
+from products.signals.backend.personal_ranking_tasks import start_first_summary_build
 from products.signals.backend.report_embeddings import EMBEDDING_PRODUCT
 
 logger = structlog.get_logger(__name__)
@@ -371,7 +372,7 @@
     return team is not None and is_team_enrolled_for_generation(read_config(), team)
 
 
 def run_pair(input: ActivitySummaryPairInput, *, last_attempt: bool = True) -> ActivitySummaryPairResult:
     if not _enrolled_at_dispatch(input.team_id):
         return ActivitySummaryPairResult(
             status=UpdateStatus.SKIPPED_NOT_ENROLLED.value, cost_usd=None, called_model=False
@@ -404,6 +405,14 @@
     except Exception:
         # The pair stays due, so the next tick picks it, skips it as not eligible yet, and syncs it then.
         logger.exception("activity_summary_queue_sync_failed", team_id=input.team_id, user_id=input.user_id)
+    if update.status == UpdateStatus.OK and update.new_summary:
+        try:
+            start_first_summary_build(input.team_id, input.user_id)
+        except Exception:
+            # The personal ranking tick builds the first list later, because the pair still has no list.
+            logger.exception(
+                "activity_summary_first_ranking_build_failed", team_id=input.team_id, user_id=input.user_id
+            )
     if update.status == UpdateStatus.OK and input.emit_embeddings:
         try:
             _emit_embedding(input.team_id, input.user_id, emit_profile=input.emit_profile_embeddings)

products/signals/backend/management/AGENTS.md

@@ -344,6 +344,10 @@ A build with no ready activity summary, or for an organization that has not appr
 When a limit refuses a build, the read serves the latest row as stale, or the candidates in inbox order with `fallback_reason: rate_limited`.
 The scheduled tick spends from the same bucket.
 
+When the activity summary tick stores a person's first summary and the person has no list, it queues the same Celery task with `trigger: first_summary`, so the first list does not wait for the personal ranking tick.
+That build takes the per-person lock, needs `"enabled"` in the payload, and runs only when it calls the model, so it also needs candidates and AI approval, and it spends the shared `max_builds_per_day`.
+It does not need tick enrollment. A build that a gate refuses leaves the pair to the tick.
+
 ### Scheduled personal ranking tick
 
 `personal-inbox-ranking-tick` (`../personal_ranking/tick.py`) is a Temporal schedule on `SELF_DRIVING_TASK_QUEUE` with overlap `SKIP`.
@@ -380,7 +384,7 @@ The `tick` object of the `personal-inbox-ranking` payload holds its levers. With
 
 Each enabled tick captures `personal inbox ranking tick` with the due pairs, the pairs it picked, a count per pair status (`built`, `failed`, `unchanged`, `running`, `not_enrolled`, `tick_limited`, `shared_limited`), the stop reason, `oldest_due_hours`, `fresh_share`, the enrolled organizations and teams, and the duration.
 It also carries `unchanged_pair_seconds_p50`, `unchanged_pair_seconds_p95`, `built_pair_seconds_p50` and `built_pair_seconds_p95`: the seconds from the start of a pair to its result, for sizing `max_concurrency` and `max_pairs_per_tick`.
-`personal inbox ranking built` and the model call's trace carry `trigger` (`tick` or `on_demand`), so cost can be split by what started a build.
+`personal inbox ranking built` and the model call's trace carry `trigger` (`tick`, `on_demand` or `first_summary`), so cost can be split by what started a build.
 
 `due_pairs`, `due_without_list` and `oldest_due_hours` are read when the tick plans, before it does any work.
 A healthy tick reads `oldest_due_hours` a little above `target_interval_minutes`, and one old list that just became due makes it spike, so do not alert on it.

products/signals/backend/personal_ranking/logic.py

@@ -87,7 +87,7 @@
 
 BuildStatus = Literal["ok", "fallback", "empty", "disabled"]
 # What started a build, on the build event and the trace, so cost can be split by it.
-BuildTrigger = Literal["on_demand", "tick"]
+BuildTrigger = Literal["on_demand", "tick", "first_summary"]
 
 _MODEL_ORDER_REASON = "Kept in inbox order."
 _NEEDS_YOU_REASONS = {
@@ -499,7 +499,7 @@ def _capture(
                 distinct_id=user.distinct_id or PIPELINE_DISTINCT_ID,
                 event=BUILT_EVENT,
                 properties={
-                    **({"$update_person_last_seen_at": False} if trigger == "tick" else {}),
+                    **({"$update_person_last_seen_at": False} if trigger != "on_demand" else {}),
                     "team_id": team.id,
                     "user_id": user.id,
                     "ranking_id": build.ranking_id,

products/signals/backend/personal_ranking_tasks.py

@@ -6,9 +6,10 @@
 from posthog.models import Team, User
 from posthog.models.scoping import with_team_scope
 
+from products.signals.backend.models import PersonalInboxRanking
 from products.signals.backend.personal_ranking.build_budget import spend_shared_build
 from products.signals.backend.personal_ranking.config import PersonalRankingConfig, read_config
-from products.signals.backend.personal_ranking.logic import build_personal_ranking, check_inputs
+from products.signals.backend.personal_ranking.logic import BuildTrigger, build_personal_ranking, check_inputs
 
 # Above the task's time_limit, so a live build never loses its lock. It also frees the lock of a worker that died.
 BUILD_LOCK_TTL_SECONDS = 5 * 60
@@ -18,33 +19,67 @@ def build_lock_key(team_id: int, user_id: int) -> str:
     return f"signals_personal_ranking_build/{team_id}/{user_id}"
 
 
-def _skip_build(team: Team, user: User, config: PersonalRankingConfig, limit: int | None) -> bool:
-    """Compare the latest list with the current inputs, as the tick does. True when the build must not run."""
+def _skip_build(
+    team: Team, user: User, config: PersonalRankingConfig, limit: int | None, *, model_only: bool = False
+) -> bool:
+    """Compare the latest list with the current inputs, as the tick does. True when the build must not run.
+
+    With `model_only`, a build that would not call the model does not run either.
+    """
     check = check_inputs(team, user, config, now=timezone.now(), limit=limit)
     if check.unchanged:
         return True
+    if not check.calls_model:
+        return model_only
     # The read did not gather the candidates, so it could not know whether the build calls the model.
-    return check.calls_model and spend_shared_build(config) is not None
+    return spend_shared_build(config) is not None
 
 
 @shared_task(ignore_result=True, max_retries=0, soft_time_limit=3 * 60, time_limit=4 * 60)
 @with_team_scope()
 def build_personal_inbox_ranking(
-    team_id: int, user_id: int, focus: str | None, limit: int | None, check_first: bool = False
+    team_id: int,
+    user_id: int,
+    focus: str | None,
+    limit: int | None,
+    check_first: bool = False,
+    trigger: BuildTrigger = "on_demand",
 ) -> None:
-    """Build and store one person's ranked inbox list. The `for_me` read takes the lock before it queues this task.
+    """Build and store one person's ranked inbox list. The caller takes the lock before it queues this task.
 
     With `check_first`, a list that still matches the current inputs gets `checked_at` and no model call, and a
-    build that calls the model spends the shared daily limit first.
+    build that calls the model spends the shared daily limit first. A `first_summary` build runs only when it
+    calls the model.
     """
     try:
         team = Team.objects.select_related("organization").filter(id=team_id).first()
         user = User.objects.filter(id=user_id).first()
         if team is None or user is None:
             return
         config = read_config()
-        if check_first and _skip_build(team, user, config, limit):
+        if check_first and _skip_build(team, user, config, limit, model_only=trigger == "first_summary"):
             return
-        build_personal_ranking(team, user, focus=focus, limit=limit, config=config)
+        build_personal_ranking(team, user, focus=focus, limit=limit, config=config, trigger=trigger)
     finally:
         cache.delete(build_lock_key(team_id, user_id))
+
+
+def start_first_summary_build(team_id: int, user_id: int) -> bool:
+    """Queue a build for a person whose first activity summary was just stored. True when the build is queued.
+
+    The person has no list until a build stores one, so without this build the first list waits for the
+    personal ranking tick. A pair that already has a list, or that this function does not queue, stays with the tick.
+    """
+    if not read_config().enabled:
+        return False
+    if PersonalInboxRanking.objects.for_team(team_id).filter(user_id=user_id).exists():
+        return False
+    lock_key = build_lock_key(team_id, user_id)
+    if not cache.add(lock_key, "1", timeout=BUILD_LOCK_TTL_SECONDS):
+        return False
+    try:
+        build_personal_inbox_ranking.delay(team_id, user_id, None, None, True, "first_summary")
+    except Exception:
+        cache.delete(lock_key)
+        raise
+    return True

products/signals/backend/test/test_activity_summary_tick.py

@@ -8,8 +8,9 @@
 import pytest
 import time_machine
 from posthog.test.base import BaseTest
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, call, patch
 
+from django.core.cache import cache
 from django.test import override_settings
 
 import redis
@@ -90,6 +91,7 @@
     run_pair,
 )
 from products.signals.backend.models import SignalSourceConfig, UserActivitySummary, UserActivitySummaryQueueState
+from products.signals.backend.personal_ranking_tasks import build_lock_key
 from products.signals.backend.test.test_activity_summary_render import FakeLLM
 from products.tasks.backend.facade.user_task_summaries import UserTaskTime
 
@@ -1136,11 +1138,15 @@ def test_an_ok_update_emits_the_embedding_and_an_event_without_the_summary(
         self, trigger: str | None, wait_hours: int
     ) -> None:
         self._changes(self.user, 3)
+        cache.delete(build_lock_key(self.team.id, self.user.id))
+        self.addCleanup(cache.delete, build_lock_key(self.team.id, self.user.id))
         capture = MagicMock()
         with (
             patch("products.signals.backend.activity_summary.logic.gateway_client", return_value=FakeLLM([ANSWER])),
             patch("posthog.api.embedding_worker.emit_embedding_request") as emit,
             patch.object(tick, "ph_background_capture", return_value=capture),
+            patch("products.signals.backend.personal_ranking.config.read_flag_payload", return_value={}),
+            patch("products.signals.backend.personal_ranking_tasks.build_personal_inbox_ranking.delay") as queue,
         ):
             result = run_pair(
                 ActivitySummaryPairInput(
@@ -1164,6 +1170,7 @@ def test_an_ok_update_emits_the_embedding_and_an_event_without_the_summary(
         assert event["event"] == "activity summary generated"
         assert event["properties"]["trigger"] == trigger
         assert "signup funnel" not in json.dumps(event["properties"]).lower()
+        assert queue.call_args_list == [call(self.team.id, self.user.id, None, None, True, "first_summary")]
 
     @parameterized.expand(
         [

products/signals/backend/test/test_personal_ranking.py

@@ -55,7 +55,11 @@
 )
 from products.signals.backend.personal_ranking.prompt import render_person, system_prompt, user_message
 from products.signals.backend.personal_ranking.validation import MalformedRankingError, validate_ranking
-from products.signals.backend.personal_ranking_tasks import build_lock_key, build_personal_inbox_ranking
+from products.signals.backend.personal_ranking_tasks import (
+    build_lock_key,
+    build_personal_inbox_ranking,
+    start_first_summary_build,
+)
 from products.signals.backend.ranking.inbox_sort import PERSONAL_RECOMMENDED_FLAG, RECOMMENDED_INBOX_SORT_FLAG
 from products.signals.backend.ranking.report_predictions import OutcomePrediction, ReportPredictions
 
@@ -948,6 +952,43 @@ def test_a_build_keeps_only_the_newest_rows_of_that_person(self) -> None:
         assert PersonalInboxRanking.objects.for_team(self.team.id).filter(user_id=other_user.id).count() == 1
         assert PersonalInboxRanking.objects.for_team(other_team.id).filter(user_id=self.user.id).count() == 1
 
+    @parameterized.expand(
+        [
+            ("first summary with candidates", {}, True, False, ["first_summary"]),
+            ("already has a list", {}, True, True, ["on_demand"]),
+            ("disabled payload", {"enabled": False}, True, False, []),
+            ("no candidates", {}, False, False, []),
+        ]
+    )
+    def test_a_first_summary_builds_a_list_only_for_a_person_without_one(
+        self, _name: str, payload: dict[str, Any], has_candidates: bool, has_list: bool, triggers: list[str]
+    ) -> None:
+        cache.delete(build_lock_key(self.team.id, self.user.id))
+        self.organization.is_ai_data_processing_approved = True
+        self.organization.save()
+        if has_candidates:
+            self._setup_reports()
+        self.create.return_value = _response(_answer([]))
+        if has_list:
+            self._build()
+        queued: list[tuple[Any, ...]] = []
+
+        with (
+            patch("products.signals.backend.personal_ranking.config.read_flag_payload", return_value=payload),
+            patch(
+                "products.signals.backend.personal_ranking_tasks.build_personal_inbox_ranking.delay",
+                side_effect=lambda *args: queued.append(args),
+            ),
+        ):
+            start_first_summary_build(self.team.id, self.user.id)
+            for args in queued:
+                build_personal_inbox_ranking(*args)
+
+        captured = [call.kwargs["properties"]["trigger"] for call in self.capture.call_args_list]
+        assert captured == triggers
+        assert PersonalInboxRanking.objects.for_team(self.team.id).filter(user_id=self.user.id).count() == len(triggers)
+        assert cache.get(build_lock_key(self.team.id, self.user.id)) is None
+
 
 class TestForMeEndpoint(APIBaseTest):
     def setUp(self) -> None: