fix(tasks): acknowledge cloud task messages asynchronously

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

@@ -285,6 +285,7 @@ def send_user_message(
     timeout: int = COMMAND_TIMEOUT_SECONDS,
     message_id: str | None = None,
     steer: bool = False,
+    wait_for_completion: bool = True,
 ) -> CommandResult:
     """Send a user_message command to the sandbox agent.
 
@@ -301,6 +302,8 @@ def send_user_message(
         params["messageId"] = message_id
     if steer:
         params["steer"] = True
+    if not wait_for_completion:
+        params["waitForCompletion"] = False
     return send_agent_command(
         task_run,
         method="user_message",

products/tasks/backend/temporal/process_task/activities/forward_pending_message.py

@@ -128,6 +128,7 @@ def _ensure_pending_message_id(state: dict[str, Any]) -> None:
             auth_token=auth_token,
             timeout=PENDING_MESSAGE_TIMEOUT_SECONDS,
             message_id=pending_message_id,
+            wait_for_completion=False,
         )
         logger.info(
             "forward_pending_message_attempted",

products/tasks/backend/temporal/process_task/tests/test_forward_pending_message.py

@@ -101,6 +101,7 @@ def test_pending_message_delivered_successfully(self, mock_send, mock_token):
         forward_pending_user_message(str(run.id))
 
         mock_send.assert_called_once()
+        assert mock_send.call_args.kwargs["wait_for_completion"] is False
         assert mock_send.call_args[0][1] == "fix the tests"
         assert mock_send.call_args.kwargs["message_id"]
         run.refresh_from_db()