feat(tasks): route slack follow-ups through the workflow follow-up queue

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

@@ -948,13 +948,6 @@ def forward_posthog_code_followup_activity(
     if user_message_ts:
         safe_react(slack.client, channel, user_message_ts, "eyes")
 
-    auth_token = None
-    if actor_user and actor_user.id:
-        distinct_id = actor_user.distinct_id or f"user_{actor_user.id}"
-        auth_token = tasks_facade.create_sandbox_connection_token(
-            task_run.id, user_id=actor_user.id, distinct_id=distinct_id
-        )
-
     uploaded_attachments, attachment_skips = _upload_prepared_slack_attachments(
         tasks_facade,
         task_run_id=task_run.id,
@@ -973,42 +966,24 @@ def forward_posthog_code_followup_activity(
         or user_text
     )
 
-    send_kwargs: dict[str, Any] = {
-        "auth_token": auth_token,
-        "timeout": 90,
-        # Deterministic across activity retries: a retry after a partial failure
-        # (or the in-line resend below) redelivers with the same id, and the
-        # agent-server drops the duplicate instead of applying the message twice.
-        "message_id": _slack_followup_message_id(channel, user_message_ts, thread_ts),
-    }
-    if uploaded_attachments:
-        send_kwargs["artifacts"] = uploaded_attachments
-
-    result = tasks_facade.send_user_message(task_run.id, user_text, **send_kwargs)
-    if not result.success and result.retryable and result.status_code != 504:
-        result = tasks_facade.send_user_message(task_run.id, user_text, **send_kwargs)
-
-    if not result.success:
+    # Queue on the workflow so delivery is ordered with the web path. The
+    # deterministic message id keeps redelivery idempotent.
+    signal_result = tasks_facade.signal_task_run_user_message(
+        task_run.id,
+        mapping.task_id,
+        task_run.team_id,
+        content=user_text,
+        artifact_ids=_uploaded_attachment_ids(uploaded_attachments),
+        message_id=_slack_followup_message_id(channel, user_message_ts, thread_ts),
+    )
+    if signal_result is not True:
         logger.warning(
-            "posthog_code_followup_forwarding_failed",
+            "slack_app_followup_signal_failed",
             channel=channel,
             thread_ts=thread_ts,
-            error=result.error,
-            status_code=result.status_code,
+            task_run_id=str(task_run.id),
+            signal_result=signal_result,
         )
-        if result.retryable and result.status_code == 504:
-            # Agent is still processing — leave the :eyes: reaction up so the thread
-            # reads as in-progress. relayAgentResponse fires when it finishes,
-            # delivering the correct response to Slack.
-            _delete_followup_progress(
-                integration_id=inputs.integration_id,
-                channel=channel,
-                thread_ts=thread_ts,
-                user_message_ts=user_message_ts,
-                mentioning_slack_user_id=mapping.mentioning_slack_user_id,
-            )
-            return True
-
         _set_followup_done_reaction(slack, channel, user_message_ts, "x")
         slack.client.chat_postMessage(
             channel=channel,
@@ -1017,7 +992,7 @@ def forward_posthog_code_followup_activity(
         )
         return True
 
-    # Message delivered; the agent is now working on it, so leave the :eyes: reaction
+    # Message queued; the agent picks it up next, so leave the :eyes: reaction
     # up. relayAgentResponse posts the agent's response once it finishes.
     _delete_followup_progress(
         integration_id=inputs.integration_id,

products/tasks/backend/facade/api.py

@@ -198,7 +198,6 @@
     "run_task_automation_now",
     "save_code_workflow_bindings",
     "send_cancel",
-    "send_user_message",
     "select_repository_for_message",
     "set_task_run_output",
     "set_task_title",
@@ -2546,12 +2545,23 @@ def validate_task_run_artifact_ids(
 
 
 def signal_task_run_user_message(
-    run_id: str | UUID, task_id: str | UUID, team_id: int, *, content: str | None, artifact_ids: list[str]
+    run_id: str | UUID,
+    task_id: str | UUID,
+    team_id: int,
+    *,
+    content: str | None,
+    artifact_ids: list[str],
+    message_id: str | None = None,
 ) -> bool | None:
     """Queue a user_message follow-up signal on the run's workflow.
 
-    Returns ``True`` on success, ``False`` if signalling failed, ``None`` if the run isn't found.
+    Returns ``True`` on success, ``False`` when the target workflow is gone
+    (completed or evicted — a terminal outcome), ``None`` when the run isn't
+    found. Transient signalling failures propagate so a calling Temporal
+    activity retries rather than reporting a dead end to the user.
     """
+    from temporalio.service import RPCError, RPCStatusCode  # noqa: PLC0415 — keep temporalio off the api import path
+
     from products.tasks.backend.temporal.client import (  # noqa: PLC0415 — keep temporalio off the api import path
         signal_task_followup_message,
     )
@@ -2560,10 +2570,12 @@ def signal_task_run_user_message(
     if run is None:
         return None
     try:
-        signal_task_followup_message(run.workflow_id, content, artifact_ids)
-    except Exception:
-        logger.exception("Failed to signal follow-up message for task run %s", run.id)
-        return False
+        signal_task_followup_message(run.workflow_id, content, artifact_ids, message_id)
+    except RPCError as e:
+        if e.status == RPCStatusCode.NOT_FOUND:
+            logger.warning("Follow-up signal target workflow gone for task run %s", run.id)
+            return False
+        raise
     return True
 
 
@@ -4723,36 +4735,6 @@ def create_sandbox_connection_token(run_id: str | UUID, user_id: int, distinct_i
     return _create(run, user_id, distinct_id)
 
 
-def send_user_message(
-    run_id: str | UUID,
-    message: str | None = None,
-    *,
-    artifacts: list[dict] | None = None,
-    auth_token: str | None = None,
-    timeout: int | None = None,
-    message_id: str | None = None,
-):
-    """Push a follow-up user message (and/or artifacts) into a run's live sandbox.
-
-    ``message_id`` is the agent-server idempotency key — pass a deterministic id when the
-    caller may retry delivery so a redelivered message isn't applied twice.
-    """
-    from products.tasks.backend.logic.services.agent_command import (  # noqa: PLC0415 — keep sandbox deps off the api import path
-        send_user_message as _send,
-    )
-
-    run = TaskRun.objects.select_related("task").get(id=run_id)
-    # Forward only explicitly-provided optionals so the underlying call shape is unchanged.
-    extra: dict = {}
-    if artifacts is not None:
-        extra["artifacts"] = artifacts
-    if timeout is not None:
-        extra["timeout"] = timeout
-    if message_id is not None:
-        extra["message_id"] = message_id
-    return _send(run, message, auth_token=auth_token, **extra)
-
-
 def send_cancel(run_id: str | UUID, *, auth_token: str | None = None):
     """Cancel the agent running in a run's live sandbox."""
     from products.tasks.backend.logic.services.agent_command import (  # noqa: PLC0415 — keep sandbox deps off the api import path

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

@@ -1532,9 +1532,16 @@ def command(self, request, pk=None, **kwargs):
                         status=status.HTTP_400_BAD_REQUEST,
                     )
 
-            signal_result = tasks_facade.signal_task_run_user_message(
-                pk, task_id, self.team_id, content=command_params.get("content"), artifact_ids=artifact_ids
-            )
+            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
+                )
+            except Exception:
+                # A synchronous web request can't retry the way the Temporal
+                # follow-up path does, so a transient signalling failure surfaces
+                # as the same gateway error as a terminal one below.
+                logger.warning("Failed to queue user message for task run %s", pk)
+                signal_result = False
             if signal_result is None:
                 raise NotFound()
             if signal_result is False:

products/tasks/backend/temporal/client.py

@@ -465,10 +465,15 @@ def execute_build_sandbox_image_workflow(image_id: str, team_id: int, *, refresh
     )
 
 
-def signal_task_followup_message(workflow_id: str, message: str | None, artifact_ids: list[str]) -> None:
+def signal_task_followup_message(
+    workflow_id: str,
+    message: str | None,
+    artifact_ids: list[str],
+    message_id: str | None = None,
+) -> None:
     client = sync_connect()
     handle = client.get_workflow_handle(workflow_id)
-    asyncio.run(handle.signal("send_followup_message", args=[message, artifact_ids]))
+    asyncio.run(handle.signal("send_followup_message", args=[message, artifact_ids, message_id]))
 
 
 def signal_agent_text_delta(workflow_id: str, text: str) -> None:

products/tasks/backend/temporal/process_task/workflow.py

@@ -138,6 +138,9 @@ class ProcessTaskInput:
 class PendingFollowup:
     message: str | None
     artifact_ids: list[str]
+    # Sender-supplied idempotency key (stable across the sender's retries);
+    # None falls back to a workflow-generated id.
+    message_id: str | None = None
 
 
 @dataclass
@@ -690,6 +693,7 @@ async def run(self, input: ProcessTaskInput) -> ProcessTaskOutput:
                             await self._send_followup_to_sandbox(
                                 message=message,
                                 artifact_ids=artifact_ids,
+                                message_id=pending_followup.message_id,
                             )
                             continue
 
@@ -1832,7 +1836,7 @@ async def send_followup_message(
                 "artifact_count": len(artifact_ids or []),
             },
         )
-        pending_followup = PendingFollowup(message=message, artifact_ids=artifact_ids or [])
+        pending_followup = PendingFollowup(message=message, artifact_ids=artifact_ids or [], message_id=message_id)
         # Always queue. `deprecate_patch` accepts existing non-deprecated
         # markers from workflows that ran the prior `workflow.patched(...)`
         # gate, so this is safe to deploy alongside in-flight workflows. The
@@ -1887,7 +1891,9 @@ async def send_permission_response(self, response: dict[str, Any]) -> None:
             },
         )
 
-    async def _send_followup_to_sandbox(self, message: str | None, artifact_ids: list[str]) -> None:
+    async def _send_followup_to_sandbox(
+        self, message: str | None, artifact_ids: list[str], message_id: str | None = None
+    ) -> None:
         workflow.logger.info(
             "send_followup_dispatch_begin",
             extra={
@@ -1904,7 +1910,7 @@ async def _send_followup_to_sandbox(self, message: str | None, artifact_ids: lis
                     message=message,
                     posthog_mcp_scopes=self._posthog_mcp_scopes,
                     artifact_ids=artifact_ids,
-                    message_id=str(workflow.uuid4()),
+                    message_id=message_id or str(workflow.uuid4()),
                 ),
                 start_to_close_timeout=timedelta(minutes=35),
                 # The activity heartbeats while blocked on the sync delivery

products/tasks/backend/temporal/tests/test_client.py

@@ -0,0 +1,36 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from products.tasks.backend.temporal.client import (
+    execute_posthog_code_agent_relay_workflow,
+    signal_task_followup_message,
+)
+from products.tasks.backend.temporal.slack_relay.activities import RelaySlackMessageInput
+
+
+@patch("products.tasks.backend.temporal.client.sync_connect")
+def test_relay_enqueue_constructs_workflow_input(mock_connect: MagicMock) -> None:
+    # Guards against the client kwargs drifting from the RelaySlackMessageInput
+    # fields — that mismatch raises TypeError at enqueue time and every Slack
+    # relay surfaces as a 503 while the sandbox swallows the error silently.
+    mock_client = MagicMock(start_workflow=AsyncMock())
+    mock_connect.return_value = mock_client
+
+    relay_id = execute_posthog_code_agent_relay_workflow(
+        run_id="run-1", text="hello", relay_id="relay-1", user_message_ts="123.456"
+    )
+
+    assert relay_id == "relay-1"
+    workflow_input = mock_client.start_workflow.call_args.args[1]
+    assert isinstance(workflow_input, RelaySlackMessageInput)
+    assert workflow_input.text == "hello"
+    assert workflow_input.run_id == "run-1"
+
+
+@patch("products.tasks.backend.temporal.client.sync_connect")
+def test_followup_signal_sends_expected_args(mock_connect: MagicMock) -> None:
+    handle = MagicMock(signal=AsyncMock())
+    mock_connect.return_value = MagicMock(get_workflow_handle=MagicMock(return_value=handle))
+
+    signal_task_followup_message("wf-1", "hi", ["artifact-1"], message_id="msg-1")
+
+    handle.signal.assert_awaited_once_with("send_followup_message", args=["hi", ["artifact-1"], "msg-1"])

products/tasks/backend/tests/test_api.py

@@ -7829,7 +7829,7 @@ def test_command_signals_user_message(self, mock_signal_followup):
         self.assertEqual(data["jsonrpc"], "2.0")
         self.assertTrue(data["result"]["queued"])
 
-        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [])
+        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [], None)
 
     @patch("products.tasks.backend.temporal.client.signal_task_followup_message")
     def test_command_user_message_requires_code_access(self, mock_signal_followup):
@@ -7865,7 +7865,7 @@ def test_command_signals_user_message_without_active_sandbox(self, mock_signal_f
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertTrue(response.json()["result"]["queued"])
-        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [])
+        mock_signal_followup.assert_called_once_with(run.workflow_id, "Hello agent", [], None)
 
     @patch("products.tasks.backend.temporal.client.signal_task_followup_message")
     def test_command_signals_user_message_artifact_ids(self, mock_signal_followup):
@@ -7897,7 +7897,7 @@ def test_command_signals_user_message_artifact_ids(self, mock_signal_followup):
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertTrue(response.json()["result"]["queued"])
-        mock_signal_followup.assert_called_once_with(run.workflow_id, "See attached", ["artifact-123"])
+        mock_signal_followup.assert_called_once_with(run.workflow_id, "See attached", ["artifact-123"], None)
 
     @patch("products.tasks.backend.temporal.client.signal_task_followup_message")
     def test_command_returns_502_when_user_message_signal_fails(self, mock_signal_followup):