feat(tasks): route slack follow-ups through the queue with per-message actors

posthog/temporal/ai/slack_app/activities/task_creation.py

@@ -82,10 +82,11 @@ def _canvas_file_delivery_available(integration: Integration) -> bool:
 
 
 def _slack_actor_state_updates(*, user_id: int, slack_user_id: str) -> dict[str, Any]:
-    return {
-        "slack_actor_user_id": user_id,
-        "slack_actor_slack_user_id": slack_user_id,
-    }
+    from products.tasks.backend.facade import (
+        api as tasks_facade,  # noqa: PLC0415 — keep tasks deps off the slack_app import path
+    )
+
+    return tasks_facade.slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)
 
 
 def _strip_context_tag(text: str) -> str:
@@ -847,26 +848,12 @@ def forward_posthog_code_followup_activity(
     ):
         return True
 
-    # Record the live actor so async reply paths tag them instead of the
-    # thread's original mentioner. Concurrent follow-ups can race here; see PR.
+    # Reply-tag fallback for turns with no per-turn actor (boot prompt,
+    # pre-rollout runs); the actor stamped at delivery normally wins.
     if slack_user_id != mapping.latest_actor_slack_user_id:
         mapping.latest_actor_slack_user_id = slack_user_id
         mapping.save(update_fields=["latest_actor_slack_user_id", "updated_at"])
 
-    if actor_user and actor_user.id:
-        try:
-            tasks_facade.update_task_run_state(
-                task_run.id,
-                updates=_slack_actor_state_updates(user_id=actor_user.id, slack_user_id=slack_user_id),
-            )
-        except Exception:
-            logger.exception(
-                "posthog_code_followup_actor_state_update_failed",
-                channel=channel,
-                thread_ts=thread_ts,
-                actor_user_id=actor_user.id,
-            )
-
     if task_run.is_terminal:
         return _resume_task_with_new_run(
             mapping,
@@ -974,7 +961,9 @@ def forward_posthog_code_followup_activity(
         task_run.team_id,
         content=user_text,
         artifact_ids=_uploaded_attachment_ids(uploaded_attachments),
+        actor_user_id=actor_user.id if actor_user and actor_user.id else None,
         message_id=_slack_followup_message_id(channel, user_message_ts, thread_ts),
+        actor_slack_user_id=slack_user_id,
     )
     if signal_result is not True:
         logger.warning(

products/slack_app/backend/models.py

@@ -27,6 +27,9 @@ class SlackThreadTaskMapping(UUIDModel):
         related_name="slack_thread_mappings",
     )
     mentioning_slack_user_id = models.CharField(max_length=64)
+    # Reply-tag fallback for runs started before per-turn actor capture
+    # (tasks slack_relay); drop the column and its stamp in task_creation
+    # once those runs drain.
     latest_actor_slack_user_id = models.CharField(max_length=64, null=True, blank=True)
     # Slack `ts` of the most recent message we've already shown to the agent (either
     # in the original `<slack_thread_context>` block at task creation, or in a follow-up

products/slack_app/backend/tests/test_followup_forwarding.py

@@ -939,6 +939,7 @@ def test_cross_user_followup_authorized_prefixes_actor_name(self, mock_slack_cls
         assert mock_signal.call_args.args == (self.task_run.id, self.task.id, self.team.id)
         signal_kwargs = mock_signal.call_args.kwargs
         assert signal_kwargs["content"] == "Bob: please retry the build"
+        assert signal_kwargs["actor_user_id"] == bob.id
         assert signal_kwargs["message_id"] is not None
         # No "Only the person who started" denial; the message went through.
         post_calls = [
@@ -963,6 +964,7 @@ def test_cross_user_followup_falls_back_to_email_when_no_full_name(self, mock_sl
         mock_signal.assert_called_once()
         signal_kwargs = mock_signal.call_args.kwargs
         assert signal_kwargs["content"] == "bob@test.com: ping"
+        assert signal_kwargs["actor_user_id"] == bob.id
         assert signal_kwargs["message_id"] is not None
 
     @patch("products.slack_app.backend.api.resolve_slack_user", return_value=None)
@@ -1048,6 +1050,7 @@ def test_successful_forwarding(self, mock_slack_cls, mock_signal):
         signal_kwargs = mock_signal.call_args.kwargs
         assert signal_kwargs["content"] == "do something"
         assert signal_kwargs["artifact_ids"] == []
+        assert signal_kwargs["actor_user_id"] == self.user.id
         assert signal_kwargs["message_id"] is not None
         # The message is queued on the workflow, so the :eyes: reaction stays up — it
         # is not swapped to :hedgehog: until the task genuinely completes.
@@ -1106,6 +1109,7 @@ def test_attachment_only_followup_uploads_and_forwards_to_sandbox(self) -> None:
         content = first_call.kwargs["content"]
         assert content.startswith("Attached Slack file(s).")
         assert "Slack attachment(s) available to the agent as task files: only-log.txt." in content
+        assert first_call.kwargs["actor_user_id"] == self.user.id
         assert mock_write.call_count == 2
         assert mock_write.call_args_list[0].args[0] == mock_write.call_args_list[1].args[0]
         self.task_run.refresh_from_db()

products/tasks/backend/facade/api.py

@@ -201,6 +201,7 @@
     "select_repository_for_message",
     "set_task_run_output",
     "set_task_title",
+    "slack_actor_state_updates",
     "signal_report_queryset",
     "signal_task_run_user_message",
     "signal_workflow_completion",
@@ -929,6 +930,19 @@ def update_task_run_state(
     return TaskRun.update_state_atomic(run_id, updates=updates, remove_keys=remove_keys)
 
 
+def slack_actor_state_updates(*, user_id: int, slack_user_id: str | None = None) -> dict[str, Any]:
+    """Run-state updates recording the Slack user currently steering a run.
+
+    Credential resolution and reply tagging read the keys this builds, so every
+    writer must go through here rather than assembling the dict inline.
+    """
+    from products.tasks.backend.logic.services.run_actor import (  # noqa: PLC0415 — keep tasks internals off the api import path
+        slack_actor_state_updates as _slack_actor_state_updates,
+    )
+
+    return _slack_actor_state_updates(user_id=user_id, slack_user_id=slack_user_id)
+
+
 def set_task_run_created_at_for_seeding(
     run_id: str | UUID, task_id: str | UUID, team_id: int, *, created_at: datetime
 ) -> None:
@@ -2551,7 +2565,9 @@ def signal_task_run_user_message(
     *,
     content: str | None,
     artifact_ids: list[str],
+    actor_user_id: int | None = None,
     message_id: str | None = None,
+    actor_slack_user_id: str | None = None,
 ) -> bool | None:
     """Queue a user_message follow-up signal on the run's workflow.
 
@@ -2570,7 +2586,8 @@ def signal_task_run_user_message(
     if run is None:
         return None
     try:
-        signal_task_followup_message(run.workflow_id, content, artifact_ids, message_id)
+        context = {"actor_slack_user_id": actor_slack_user_id} if actor_slack_user_id else None
+        signal_task_followup_message(run.workflow_id, content, artifact_ids, message_id, actor_user_id, context)
     except RPCError as e:
         if e.status == RPCStatusCode.NOT_FOUND:
             logger.warning("Follow-up signal target workflow gone for task run %s", run.id)
@@ -2689,6 +2706,7 @@ def relay_task_run_message(
     *,
     text: str,
     text_parts: list[str] | None = None,
+    message_id: str | None = None,
 ) -> tuple[str, str | None]:
     """Queue a Slack relay workflow for a run message, or under the agent-design
     flag signal the running task workflow to stream the text inline.
@@ -2732,7 +2750,12 @@ def relay_task_run_message(
         return "skipped", None
 
     try:
-        relay_id = execute_posthog_code_agent_relay_workflow(run_id=str(run.id), text=trimmed, delete_progress=True)
+        relay_id = execute_posthog_code_agent_relay_workflow(
+            run_id=str(run.id),
+            text=trimmed,
+            delete_progress=True,
+            message_id=message_id,
+        )
     except Exception:
         logger.exception("task_run_relay_message_enqueue_failed", extra={"run_id": str(run.id)})
         return "failed", None

products/tasks/backend/logic/services/run_actor.py

@@ -92,3 +92,16 @@ def get_task_run_credential_user(task: Task, state: dict[str, Any] | None = None
 
 def get_actor_distinct_id(actor: User) -> str:
     return actor.distinct_id or f"user_{actor.id}"
+
+
+def slack_actor_state_updates(*, user_id: int, slack_user_id: str | None = None) -> dict[str, Any]:
+    """Run-state updates recording the Slack user currently steering a run.
+
+    The keys are load-bearing: credential resolution reads
+    ``slack_actor_user_id`` and reply tagging reads
+    ``slack_actor_slack_user_id`` — every writer must build them here.
+    """
+    updates: dict[str, Any] = {"slack_actor_user_id": user_id}
+    if slack_user_id:
+        updates["slack_actor_slack_user_id"] = slack_user_id
+    return updates

products/tasks/backend/presentation/serializers.py

@@ -713,6 +713,12 @@ class TaskRunRelayMessageRequestSerializer(serializers.Serializer):
         max_length=10000,
         help_text="Joined message body. Used when text_parts is absent.",
     )
+    message_id = serializers.CharField(
+        max_length=128,
+        required=False,
+        allow_null=True,
+        help_text="Id of the user message this turn answers, when the agent-server echoes it.",
+    )
     # Kept optional for forward/backward compatibility during rollout; will be aligned once deployed.
     text_parts = serializers.ListField(
         child=serializers.CharField(max_length=10000, allow_blank=True),

products/tasks/backend/presentation/views/api.py

@@ -1175,6 +1175,7 @@ def relay_message(self, request, pk=None, **kwargs):
             self.team_id,
             text=request.validated_data["text"],
             text_parts=request.validated_data.get("text_parts"),
+            message_id=request.validated_data.get("message_id"),
         )
         if relay_status == "failed":
             return Response(
@@ -1534,7 +1535,12 @@ def command(self, request, pk=None, **kwargs):
 
             try:
                 signal_result = tasks_facade.signal_task_run_user_message(
-                    pk, task_id, self.team_id, content=command_params.get("content"), artifact_ids=artifact_ids
+                    pk,
+                    task_id,
+                    self.team_id,
+                    content=command_params.get("content"),
+                    artifact_ids=artifact_ids,
+                    actor_user_id=request.user.id,
                 )
             except Exception:
                 # A synchronous web request can't retry the way the Temporal

products/tasks/backend/temporal/client.py

@@ -470,10 +470,16 @@ def signal_task_followup_message(
     message: str | None,
     artifact_ids: list[str],
     message_id: str | None = None,
+    actor_user_id: int | None = None,
+    context: dict[str, Any] | None = None,
 ) -> None:
+    """New per-message fields go in ``context`` — the positional signal args
+    are frozen for worker deploy compat."""
     client = sync_connect()
     handle = client.get_workflow_handle(workflow_id)
-    asyncio.run(handle.signal("send_followup_message", args=[message, artifact_ids, message_id]))
+    asyncio.run(
+        handle.signal("send_followup_message", args=[message, artifact_ids, message_id, actor_user_id, context])
+    )
 
 
 def signal_agent_text_delta(workflow_id: str, text: str) -> None:
@@ -490,6 +496,7 @@ def execute_posthog_code_agent_relay_workflow(
     user_message_ts: str | None = None,
     delete_progress: bool = True,
     reaction_emoji: str | None = None,
+    message_id: str | None = None,
 ) -> str:
     relay_id = relay_id or str(uuid.uuid4())
     workflow_id = f"posthog-code-agent-relay-{run_id}-{relay_id}"
@@ -505,6 +512,7 @@ def execute_posthog_code_agent_relay_workflow(
                 user_message_ts=user_message_ts,
                 delete_progress=delete_progress,
                 reaction_emoji=reaction_emoji,
+                message_id=message_id,
             ),
             id=workflow_id,
             id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,