From 4febdf9d4e8eecee17999a65abc9420e43ddc1fd Mon Sep 17 00:00:00 2001 From: Vojta Bartos Date: Tue, 14 Jul 2026 19:00:54 +0200 Subject: [PATCH] feat(tasks): route slack follow-ups through the workflow follow-up queue Slack thread follow-ups delivered straight to the sandbox via the facade, bypassing the task workflow's follow-up queue: no ordering with the web path, and a busy agent needed a 504 special case. The forward now signals send_followup_message like the web UI, so the workflow delivers one message at a time and each delivery blocks until the turn completes. The deterministic slack message id travels with the signal, keeping redelivery idempotent across retries. Delivery failures surface through the run error path; the forward still replies inline when a message cannot be queued. The facade send_user_message wrapper loses its last caller and is removed. --- .../ai/slack_app/activities/task_creation.py | 55 ++---- .../backend/tests/test_followup_forwarding.py | 181 +++++------------- products/tasks/backend/facade/api.py | 56 ++---- .../tasks/backend/presentation/views/api.py | 13 +- products/tasks/backend/temporal/client.py | 9 +- .../backend/temporal/process_task/workflow.py | 12 +- .../backend/temporal/tests/test_client.py | 36 ++++ products/tasks/backend/tests/test_api.py | 6 +- 8 files changed, 145 insertions(+), 223 deletions(-) create mode 100644 products/tasks/backend/temporal/tests/test_client.py diff --git a/posthog/temporal/ai/slack_app/activities/task_creation.py b/posthog/temporal/ai/slack_app/activities/task_creation.py index 1c5d0ed16f77..e65ee064457e 100644 --- a/posthog/temporal/ai/slack_app/activities/task_creation.py +++ b/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, diff --git a/products/slack_app/backend/tests/test_followup_forwarding.py b/products/slack_app/backend/tests/test_followup_forwarding.py index 884d6e79b173..af7cd3470864 100644 --- a/products/slack_app/backend/tests/test_followup_forwarding.py +++ b/products/slack_app/backend/tests/test_followup_forwarding.py @@ -1,7 +1,7 @@ from types import SimpleNamespace from unittest import TestCase as UnitTestCase -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import MagicMock, patch from django.apps import apps from django.test import TestCase @@ -48,12 +48,6 @@ def _make_slack_file(**overrides: object) -> dict[str, object]: return file -def _command_result(**kwargs): - defaults = {"success": False, "status_code": 0, "error": None, "retryable": False, "data": None} - defaults.update(kwargs) - return SimpleNamespace(**defaults) - - def _assert_quota_denial_posted(mock_slack_instance: MagicMock, channel: str, thread_ts: str) -> None: denial_calls = [ call @@ -922,25 +916,18 @@ def test_unauthorized_actor_returns_true_with_resolver_feedback(self, mock_slack mock_resolve.assert_called_once_with(mock_slack_instance, self.integration, "U_BOB", "C123", "1234.5678") mock_slack_instance.client.chat_postMessage.assert_not_called() - @patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ) - @patch("products.tasks.backend.logic.services.agent_command.send_user_message") + @patch("products.tasks.backend.facade.api.signal_task_run_user_message", return_value=True) @patch("products.slack_app.backend.api.resolve_slack_user") @patch("posthog.models.integration.SlackIntegration") - def test_cross_user_followup_authorized_prefixes_actor_name( - self, mock_slack_cls, mock_resolve, mock_send, mock_token - ): + def test_cross_user_followup_authorized_prefixes_actor_name(self, mock_slack_cls, mock_resolve, mock_signal): # A second user in the same PostHog org and team should be allowed to chip in - # on the thread; their message is forwarded under their own sandbox identity + # on the thread; their message is queued under their own PostHog identity # and their name is prepended so the agent knows who actually spoke. self._create_mapping(mentioning_user="U_ALICE") bob = User.objects.create(email="bob@test.com", first_name="Bob") mock_slack_instance = MagicMock() mock_slack_cls.return_value = mock_slack_instance mock_resolve.return_value = SlackUserContext(user=bob, slack_email="bob@test.com") - mock_send.return_value = _command_result(success=True, status_code=200) inputs = _make_inputs(self.integration.id) result = forward_posthog_code_followup_activity( @@ -948,10 +935,11 @@ def test_cross_user_followup_authorized_prefixes_actor_name( ) assert result is True - assert mock_token.call_args.args[1] == bob.id - mock_send.assert_called_once_with( - self.task_run, "Bob: please retry the build", auth_token="jwt-token", timeout=90, message_id=ANY - ) + mock_signal.assert_called_once() + 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["message_id"] is not None # No "Only the person who started" denial; the message went through. post_calls = [ call @@ -960,29 +948,22 @@ def test_cross_user_followup_authorized_prefixes_actor_name( ] assert not post_calls - @patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ) - @patch("products.tasks.backend.logic.services.agent_command.send_user_message") + @patch("products.tasks.backend.facade.api.signal_task_run_user_message", return_value=True) @patch("products.slack_app.backend.api.resolve_slack_user") @patch("posthog.models.integration.SlackIntegration") - def test_cross_user_followup_falls_back_to_email_when_no_full_name( - self, mock_slack_cls, mock_resolve, mock_send, mock_token - ): + def test_cross_user_followup_falls_back_to_email_when_no_full_name(self, mock_slack_cls, mock_resolve, mock_signal): self._create_mapping(mentioning_user="U_ALICE") bob = User.objects.create(email="bob@test.com") # no full name mock_slack_cls.return_value = MagicMock() mock_resolve.return_value = SlackUserContext(user=bob, slack_email="bob@test.com") - mock_send.return_value = _command_result(success=True, status_code=200) inputs = _make_inputs(self.integration.id) forward_posthog_code_followup_activity(inputs, "C123", "1234.5678", "U_BOB", "<@BOT> ping", "1234.5679") - assert mock_token.call_args.args[1] == bob.id - mock_send.assert_called_once_with( - self.task_run, "bob@test.com: ping", auth_token="jwt-token", timeout=90, message_id=ANY - ) + mock_signal.assert_called_once() + signal_kwargs = mock_signal.call_args.kwargs + assert signal_kwargs["content"] == "bob@test.com: ping" + assert signal_kwargs["message_id"] is not None @patch("products.slack_app.backend.api.resolve_slack_user", return_value=None) @patch("posthog.models.integration.SlackIntegration") @@ -1049,21 +1030,12 @@ def test_sandbox_not_ready_returns_true_with_message(self, mock_slack_cls): call_kwargs = mock_slack_instance.client.chat_postMessage.call_args.kwargs assert "still starting up" in call_kwargs["text"] - @patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ) - @patch("products.tasks.backend.logic.services.agent_command.send_user_message") + @patch("products.tasks.backend.facade.api.signal_task_run_user_message", return_value=True) @patch("posthog.models.integration.SlackIntegration") - def test_successful_forwarding(self, mock_slack_cls, mock_send, mock_token): + def test_successful_forwarding(self, mock_slack_cls, mock_signal): mapping = self._create_mapping() mock_slack_instance = MagicMock() mock_slack_cls.return_value = mock_slack_instance - mock_send.return_value = _command_result( - success=True, - status_code=200, - data={"result": {"assistant_message": "thanks"}}, - ) inputs = _make_inputs(self.integration.id) result = forward_posthog_code_followup_activity( @@ -1071,12 +1043,14 @@ def test_successful_forwarding(self, mock_slack_cls, mock_send, mock_token): ) assert result is True - mock_token.assert_called_once() - mock_send.assert_called_once_with( - self.task_run, "do something", auth_token="jwt-token", timeout=90, message_id=ANY - ) - # Agent is now working on the message, so the :eyes: reaction stays up — it is - # not swapped to :hedgehog: until the task genuinely completes. + mock_signal.assert_called_once() + 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"] == "do something" + assert signal_kwargs["artifact_ids"] == [] + 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. mock_slack_instance.client.reactions_add.assert_called_once_with( channel="C123", timestamp="1234.5679", name="eyes" ) @@ -1109,11 +1083,7 @@ def test_attachment_only_followup_uploads_and_forwards_to_sandbox(self) -> None: with ( patch("posthog.models.integration.SlackIntegration") as mock_slack_cls, - patch("products.tasks.backend.logic.services.agent_command.send_user_message") as mock_send, - patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ), + patch("products.tasks.backend.facade.api.signal_task_run_user_message", return_value=True) as mock_signal, patch("posthog.temporal.ai.slack_app.attachments._download_slack_file", return_value=b"log bytes"), patch("products.slack_app.backend.services.slack_messages.collect_thread_messages", return_value=[]), patch("posthog.storage.object_storage.write") as mock_write, @@ -1123,7 +1093,6 @@ def test_attachment_only_followup_uploads_and_forwards_to_sandbox(self) -> None: mock_slack_instance.client.token = "xoxb-test" mock_slack_instance.client.auth_test.return_value = {"bot_id": "B123"} mock_slack_cls.return_value = mock_slack_instance - mock_send.return_value = _command_result(success=True, status_code=200) result = forward_posthog_code_followup_activity(inputs, "C123", "1234.5678", "U_ALICE", "", "1234.5679") # Temporal retries re-run the whole activity body; the re-upload must @@ -1131,19 +1100,22 @@ def test_attachment_only_followup_uploads_and_forwards_to_sandbox(self) -> None: forward_posthog_code_followup_activity(inputs, "C123", "1234.5678", "U_ALICE", "", "1234.5679") assert result is True - sent_run, sent_message = mock_send.call_args.args - assert sent_run.id == self.task_run.id - assert sent_message.startswith("Attached Slack file(s).") - assert "Slack attachment(s) available to the agent as task files: only-log.txt." in sent_message - sent_artifacts = mock_send.call_args.kwargs["artifacts"] - assert len(sent_artifacts) == 1 - assert sent_artifacts[0]["name"] == "only-log.txt" - assert sent_artifacts[0]["source"] == "slack_user_attachment" - assert mock_send.call_args.kwargs["auth_token"] == "jwt-token" + assert mock_signal.call_count == 2 + first_call, second_call = mock_signal.call_args_list + assert first_call.args == (self.task_run.id, self.task.id, self.team.id) + 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 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() assert len(self.task_run.artifacts) == 1 + assert first_call.kwargs["artifact_ids"] == [str(self.task_run.artifacts[0]["id"])] + assert second_call.kwargs["artifact_ids"] == first_call.kwargs["artifact_ids"] + # Redelivery must reuse the deterministic idempotency key or the agent + # applies the message twice. + assert first_call.kwargs["message_id"] is not None + assert first_call.kwargs["message_id"] == second_call.kwargs["message_id"] @parameterized.expand( [ @@ -1192,39 +1164,12 @@ def test_followup_with_only_rejected_attachments_posts_notice_without_waking_age assert "no attachment was accepted" in notice assert "installer.exe" in notice - @patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ) - @patch("products.tasks.backend.logic.services.agent_command.send_user_message") - @patch("posthog.models.integration.SlackIntegration") - def test_forwarding_failure_posts_error(self, mock_slack_cls, mock_send, mock_token): - self._create_mapping() - mock_slack_instance = MagicMock() - mock_slack_cls.return_value = mock_slack_instance - mock_send.return_value = _command_result(success=False, status_code=401, error="Unauthorized", retryable=False) - - inputs = _make_inputs(self.integration.id) - result = forward_posthog_code_followup_activity( - inputs, "C123", "1234.5678", "U_ALICE", "<@BOT> do something", "1234.5679" - ) - assert result is True - call_kwargs = mock_slack_instance.client.chat_postMessage.call_args.kwargs - assert "couldn't deliver" in call_kwargs["text"] - - @patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ) - @patch("products.tasks.backend.logic.services.agent_command.send_user_message") + @patch("products.tasks.backend.facade.api.signal_task_run_user_message", return_value=False) @patch("posthog.models.integration.SlackIntegration") - def test_timeout_delegates_to_relay_without_posting(self, mock_slack_cls, mock_send, mock_token): + def test_signal_failure_posts_error_reply(self, mock_slack_cls, mock_signal): self._create_mapping() mock_slack_instance = MagicMock() mock_slack_cls.return_value = mock_slack_instance - mock_send.return_value = _command_result( - success=False, status_code=504, error="Sandbox request timed out", retryable=True - ) inputs = _make_inputs(self.integration.id) result = forward_posthog_code_followup_activity( @@ -1232,48 +1177,14 @@ def test_timeout_delegates_to_relay_without_posting(self, mock_slack_cls, mock_s ) assert result is True - mock_send.assert_called_once() - # Agent is still processing — relayAgentResponse delivers the response. - mock_slack_instance.client.chat_postMessage.assert_not_called() - # The :eyes: reaction stays up while the agent works — no swap to :hedgehog:. - mock_slack_instance.client.reactions_add.assert_called_once_with( - channel="C123", timestamp="1234.5679", name="eyes" - ) - mock_slack_instance.client.reactions_remove.assert_not_called() - - @patch( - "products.tasks.backend.logic.services.connection_token.create_sandbox_connection_token", - return_value="jwt-token", - ) - @patch("products.tasks.backend.logic.services.agent_command.send_user_message") - @patch("posthog.models.integration.SlackIntegration") - def test_connection_error_retries_and_succeeds(self, mock_slack_cls, mock_send, mock_token): - self._create_mapping() - mock_slack_instance = MagicMock() - mock_slack_cls.return_value = mock_slack_instance - mock_send.side_effect = [ - _command_result(success=False, status_code=502, error="Connection to sandbox failed", retryable=True), - _command_result(success=True, status_code=200), - ] - - inputs = _make_inputs(self.integration.id) - result = forward_posthog_code_followup_activity( - inputs, "C123", "1234.5678", "U_ALICE", "<@BOT> do something", "1234.5679" - ) - - assert result is True - assert mock_send.call_count == 2 - # Redelivery must reuse the idempotency key or the agent applies the message twice. - first_id = mock_send.call_args_list[0].kwargs["message_id"] - second_id = mock_send.call_args_list[1].kwargs["message_id"] - assert first_id and first_id == second_id - # The :eyes: reaction stays up while the agent works — no swap to :hedgehog:. - mock_slack_instance.client.reactions_add.assert_called_once_with( + mock_signal.assert_called_once() + call_kwargs = mock_slack_instance.client.chat_postMessage.call_args.kwargs + assert "couldn't deliver your message" in call_kwargs["text"] + # The :eyes: reaction is swapped to :x: so the thread sees the delivery failed. + mock_slack_instance.client.reactions_remove.assert_called_once_with( channel="C123", timestamp="1234.5679", name="eyes" ) - mock_slack_instance.client.reactions_remove.assert_not_called() - # Response is delivered by relayAgentResponse, not by this activity. - mock_slack_instance.client.chat_postMessage.assert_not_called() + mock_slack_instance.client.reactions_add.assert_any_call(channel="C123", timestamp="1234.5679", name="x") class TestEnforcePostHogCodeBillingQuotaActivity(TestCase): diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index 735ddc2033af..f848b799b533 100644 --- a/products/tasks/backend/facade/api.py +++ b/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 diff --git a/products/tasks/backend/presentation/views/api.py b/products/tasks/backend/presentation/views/api.py index e7e733edbb2f..b7d39b40daab 100644 --- a/products/tasks/backend/presentation/views/api.py +++ b/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: diff --git a/products/tasks/backend/temporal/client.py b/products/tasks/backend/temporal/client.py index de9776f9a281..4b583069701a 100644 --- a/products/tasks/backend/temporal/client.py +++ b/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: diff --git a/products/tasks/backend/temporal/process_task/workflow.py b/products/tasks/backend/temporal/process_task/workflow.py index 00014478b5c4..668659587337 100644 --- a/products/tasks/backend/temporal/process_task/workflow.py +++ b/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 diff --git a/products/tasks/backend/temporal/tests/test_client.py b/products/tasks/backend/temporal/tests/test_client.py new file mode 100644 index 000000000000..eb035dc12968 --- /dev/null +++ b/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"]) diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py index 74f91d274337..151c440f5c95 100644 --- a/products/tasks/backend/tests/test_api.py +++ b/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):