From 81c2836bbf45c037c0c4334bd56040a3f52e9483 Mon Sep 17 00:00:00 2001
From: yashpapa6969 <yashgupta.yg856@gmail.com>
Date: Thu, 26 Dec 2024 13:24:14 +0530
Subject: [PATCH 1/9] Enhancement: Addfeature to manually retry failed task

---
 flower/api/tasks.py        | 48 +++++++++++++++++++++++++++++++++++
 flower/static/js/flower.js | 33 ++++++++++++++++++++++++
 flower/templates/task.html |  7 ++++-
 flower/utils/tasks.py      | 52 ++++++++++++++++++++++++++++++++++++++
 4 files changed, 139 insertions(+), 1 deletion(-)

diff --git a/flower/api/tasks.py b/flower/api/tasks.py
index 730c290e4..6a9c5dd73 100644
--- a/flower/api/tasks.py
+++ b/flower/api/tasks.py
@@ -14,6 +14,7 @@
 
 from ..utils import tasks
 from ..utils.broker import Broker
+from ..utils.tasks import parse_args, parse_kwargs, make_json_serializable
 from . import BaseApiHandler
 
 logger = logging.getLogger(__name__)
@@ -636,3 +637,50 @@ def get(self, taskid):
             response['worker'] = task.worker.hostname
 
         self.write(response)
+
+class TaskReapply(BaseTaskHandler):
+    @web.authenticated
+    async def post(self, taskid):
+        """
+        Get task info and reapply the task with the same arguments.
+
+        :param taskid: ID of the task to reapply.
+        """
+        # Get original task info
+        task = tasks.get_task_by_id(self.application.events, taskid)
+        if not task:
+            raise HTTPError(404, f"Unknown task '{taskid}'")
+
+        # Get task name
+        taskname = task.name
+        if not taskname:
+            raise HTTPError(400, "Cannot reapply task with no name")
+
+        try:
+            # Get the task object from registered tasks
+            task_obj = self.capp.tasks[taskname]
+        except KeyError as exc:
+            raise HTTPError(404, f"Unknown task '{taskname}'") from exc
+
+        # Parse args and kwargs from the original task
+        try:
+            args = parse_args(task.args)
+            kwargs = parse_kwargs(task.kwargs)
+        except Exception as exc:
+            logger.error("Error parsing task arguments: %s", exc)
+            raise HTTPError(400, f"Invalid task arguments: {str(exc)}") from exc
+
+        # Apply the task with original arguments
+        try:
+            # Ensure args and kwargs are JSON serializable
+            args = make_json_serializable(args)
+            kwargs = make_json_serializable(kwargs)
+
+            result = task_obj.apply_async(args=args, kwargs=kwargs)
+            response = {'task-id': result.task_id}
+            if self.backend_configured(result):
+                response.update(state=result.state)
+            self.write(response)
+        except Exception as exc:
+            logger.error("Error reapplying task with args=%s, kwargs=%s: %s", args, kwargs, str(exc))
+            raise HTTPError(500, f"Error reapplying task: {str(exc)}") from exc
diff --git a/flower/static/js/flower.js b/flower/static/js/flower.js
index 6edde42bf..b1370673a 100644
--- a/flower/static/js/flower.js
+++ b/flower/static/js/flower.js
@@ -690,4 +690,37 @@ var flower = (function () {
 
     });
 
+    $('#task-retry').click(function () {
+        const $button = $(this);
+        const $spinner = $button.find('.spinner-border');
+        const taskId = $('#taskid').text();
+    
+        if (!taskId) {
+            show_alert('Task ID is missing. Cannot proceed.', 'danger');
+            return;
+        }
+    
+        // Show loading state
+        $button.prop('disabled', true);
+        $spinner.removeClass('d-none');
+    
+        // Reapply the task using the reapply endpoint
+        $.ajax({
+            type: 'POST',
+            url: url_prefix() + '/api/task/reapply/' + taskId,
+            success: function (response) {
+                show_alert(`Task ${taskId} has been retried (new task ID: ${response['task-id']})`, 'success');
+                // Optionally reload the page after success
+                setTimeout(() => location.reload(), 1500);
+            },
+            error: function (response) {
+                show_alert(response.responseText || 'Failed to retry task', 'danger');
+                // Reset button state on error
+                $button.prop('disabled', false);
+                $spinner.addClass('d-none');
+            }
+        });
+    });
+
+
 }(jQuery));
diff --git a/flower/templates/task.html b/flower/templates/task.html
index 8d66bb08b..5a8c3732b 100644
--- a/flower/templates/task.html
+++ b/flower/templates/task.html
@@ -16,6 +16,11 @@ <h2>{{ getattr(task, 'name', None) }}
                 <button class="btn btn-danger float-end" id="task-terminate">Terminate</button>
             {% elif task.state == "RECEIVED" or task.state == "RETRY" %}
                 <button class="btn btn-danger float-end" id="task-revoke">Revoke</button>
+                {% elif task.state == "FAILURE" %}
+                <button class="btn btn-warning float-end" id="task-retry" data-bs-toggle="button">
+                    <span class="spinner-border spinner-border-sm d-none" role="status" aria-hidden="true"></span>
+                    Retry
+                </button>
             {% end %}
           </h2>
         </div>
@@ -89,4 +94,4 @@ <h2>{{ getattr(task, 'name', None) }}
       </div>
     </div>
   </div>
-{% end %}
+{% end %}
\ No newline at end of file
diff --git a/flower/utils/tasks.py b/flower/utils/tasks.py
index 4abcd82f6..b6c82e644 100644
--- a/flower/utils/tasks.py
+++ b/flower/utils/tasks.py
@@ -1,5 +1,6 @@
 import datetime
 import time
+import json
 
 from .search import parse_search_terms, satisfies_search_terms
 
@@ -68,3 +69,54 @@ def get_task_by_id(events, task_id):
 
 def as_dict(task):
     return task.as_dict()
+
+def parse_args(args):
+    """
+    Parse and process the `args` of the task.
+    """
+    if not args:
+        return []
+    try:
+        # Attempt to parse JSON
+        parsed_args = json.loads(args)
+        if isinstance(parsed_args, str) and parsed_args.startswith('(') and parsed_args.endswith(')'):
+            return eval(parsed_args)  # Handle stringified tuples
+        return parsed_args
+    except (json.JSONDecodeError, SyntaxError):
+        # Fallback for stringified tuples or ellipsis
+        if args == '...':
+            return [...]
+        if args.startswith('(') and args.endswith(')'):
+            return eval(args)
+        return [args]
+
+def parse_kwargs(kwargs):
+    """
+    Parse and process the `kwargs` of the task.
+    """
+    if not kwargs:
+        return {}
+    try:
+        # Attempt to parse JSON
+        return json.loads(kwargs)
+    except json.JSONDecodeError:
+        try:
+            # Fallback for stringified dictionaries
+            import ast
+            if kwargs.startswith('{') and kwargs.endswith('}'):
+                return ast.literal_eval(kwargs)
+        except (ValueError, SyntaxError):
+            return {}
+    return {}
+
+def make_json_serializable(obj):
+    """
+    Recursively replace non-serializable types with JSON-serializable alternatives.
+    """
+    if isinstance(obj, list):
+        return [make_json_serializable(item) for item in obj]
+    elif isinstance(obj, dict):
+        return {key: make_json_serializable(value) for key, value in obj.items()}
+    elif obj is Ellipsis:
+        return None  # Replace `...` with `null`
+    return obj  # Return the object if it's already serializable
\ No newline at end of file

From cf3b79024660470c9e0657067948435b5ff35ab0 Mon Sep 17 00:00:00 2001
From: yashpapa6969 <yashgupta.yg856@gmail.com>
Date: Sat, 25 Jan 2025 22:01:15 +0530
Subject: [PATCH 2/9] Replace eval with ast.literal_eval in parse_args function
 for improved security

---
 flower/utils/tasks.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/flower/utils/tasks.py b/flower/utils/tasks.py
index b6c82e644..c4ca013ad 100644
--- a/flower/utils/tasks.py
+++ b/flower/utils/tasks.py
@@ -1,6 +1,7 @@
 import datetime
 import time
 import json
+import ast
 
 from .search import parse_search_terms, satisfies_search_terms
 
@@ -80,14 +81,14 @@ def parse_args(args):
         # Attempt to parse JSON
         parsed_args = json.loads(args)
         if isinstance(parsed_args, str) and parsed_args.startswith('(') and parsed_args.endswith(')'):
-            return eval(parsed_args)  # Handle stringified tuples
+            return ast.literal_eval(parsed_args)  # Handle stringified tuples safely
         return parsed_args
     except (json.JSONDecodeError, SyntaxError):
         # Fallback for stringified tuples or ellipsis
         if args == '...':
             return [...]
         if args.startswith('(') and args.endswith(')'):
-            return eval(args)
+            return ast.literal_eval(args)  
         return [args]
 
 def parse_kwargs(kwargs):
@@ -102,7 +103,6 @@ def parse_kwargs(kwargs):
     except json.JSONDecodeError:
         try:
             # Fallback for stringified dictionaries
-            import ast
             if kwargs.startswith('{') and kwargs.endswith('}'):
                 return ast.literal_eval(kwargs)
         except (ValueError, SyntaxError):

From 012e958edcbfc7841fe465543308f923c6cdfaac Mon Sep 17 00:00:00 2001
From: yashpapa6969 <yashgupta.yg856@gmail.com>
Date: Sun, 26 Jan 2025 19:33:57 +0530
Subject: [PATCH 3/9] Security: refactored code

---
 flower/utils/tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flower/utils/tasks.py b/flower/utils/tasks.py
index c4ca013ad..a6479956d 100644
--- a/flower/utils/tasks.py
+++ b/flower/utils/tasks.py
@@ -88,7 +88,7 @@ def parse_args(args):
         if args == '...':
             return [...]
         if args.startswith('(') and args.endswith(')'):
-            return ast.literal_eval(args)  
+            return ast.literal_eval(args)
         return [args]
 
 def parse_kwargs(kwargs):

From bc0b8e3468a41cb06f2b51eb52b7faa48fa3a36c Mon Sep 17 00:00:00 2001
From: yashpapa6969 <yashgupta.yg856@gmail.com>
Date: Fri, 14 Feb 2025 23:16:48 +0530
Subject: [PATCH 4/9] fix adding route for the controller TaskReapply

---
 flower/urls.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/flower/urls.py b/flower/urls.py
index d3ea5df85..1de9d8c99 100644
--- a/flower/urls.py
+++ b/flower/urls.py
@@ -51,6 +51,7 @@
     (r"/api/task/timeout/(.+)", control.TaskTimout),
     (r"/api/task/rate-limit/(.+)", control.TaskRateLimit),
     (r"/api/task/revoke/(.+)", control.TaskRevoke),
+    (r"/api/task/reapply/(.+)", tasks.TaskReapply),
     # Metrics
     (r"/metrics", monitor.Metrics),
     (r"/healthcheck", monitor.Healthcheck),

From 323474e7238f1d396f233c3da15d70276001d863 Mon Sep 17 00:00:00 2001
From: yashDriveX <yash.gupta@drivex.in>
Date: Mon, 22 Dec 2025 13:08:31 +0530
Subject: [PATCH 5/9] Add unit tests for task reapplication functionality

---
 tests/unit/api/test_tasks.py | 200 +++++++++++++++++++++++++++++++++++
 1 file changed, 200 insertions(+)

diff --git a/tests/unit/api/test_tasks.py b/tests/unit/api/test_tasks.py
index 551957d7e..96d3c6618 100644
--- a/tests/unit/api/test_tasks.py
+++ b/tests/unit/api/test_tasks.py
@@ -6,6 +6,7 @@
 
 import celery.states as states
 from celery.events import Event
+from celery.events.state import Task
 from celery.result import AsyncResult
 
 from flower.events import EventsState
@@ -96,6 +97,205 @@ def get_task_by_id(events, task_id):
         return Task()
 
 
+class TaskReapplyTests(BaseApiTestCase):
+    def test_reapply_success(self):
+        """Test successfully reapplying a task"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '[1, 2]'
+        mock_task.kwargs = '{"multiply": 2}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        body = json.loads(r.body.decode('utf-8'))
+        self.assertIn('task-id', body)
+        task.apply_async.assert_called_once_with(
+            args=[1, 2], kwargs={"multiply": 2}
+        )
+
+    def test_reapply_task_not_found(self):
+        """Test reapplying a non-existent task returns 404"""
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=None):
+            r = self.post('/api/task/reapply/nonexistent', body='')
+
+        self.assertEqual(404, r.code)
+
+    def test_reapply_task_no_name(self):
+        """Test reapplying a task with no name returns 400"""
+        mock_task = Task()
+        mock_task.name = None
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(400, r.code)
+
+    def test_reapply_unknown_task_name(self):
+        """Test reapplying a task that is not registered returns 404"""
+        mock_task = Task()
+        mock_task.name = 'unknown.task'
+        mock_task.args = '[]'
+        mock_task.kwargs = '{}'
+
+        if 'unknown.task' in self._app.capp.tasks:
+            del self._app.capp.tasks['unknown.task']
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(404, r.code)
+
+    def test_reapply_invalid_args(self):
+        """Test reapplying a task with invalid args returns 400"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = 'invalid json'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            self._app.capp.tasks['tasks.add'] = Mock()
+            with patch('flower.api.tasks.parse_args', side_effect=ValueError("Invalid args")):
+                r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(400, r.code)
+
+    def test_reapply_apply_async_error(self):
+        """Test handling error during apply_async returns 500"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '[1, 2]'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            task.apply_async = Mock(side_effect=Exception("Connection error"))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(500, r.code)
+
+    def test_reapply_with_empty_args(self):
+        """Test reapplying a task with empty args"""
+        mock_task = Task()
+        mock_task.name = 'tasks.simple'
+        mock_task.args = ''
+        mock_task.kwargs = ''
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.simple'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        task.apply_async.assert_called_once_with(args=[], kwargs={})
+
+    def test_reapply_with_ellipsis_args(self):
+        """Test reapplying a task with ellipsis in args"""
+        mock_task = Task()
+        mock_task.name = 'tasks.test'
+        mock_task.args = '...'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.test'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        task.apply_async.assert_called_once_with(args=[None], kwargs={})
+
+    def test_reapply_with_nested_json_args(self):
+        """Test reapplying task with nested JSON structures in args"""
+        mock_task = Task()
+        mock_task.name = 'tasks.process'
+        mock_task.args = '[{"user_id": 123, "items": [1, 2, 3]}, "action"]'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.process'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        task.apply_async.assert_called_once_with(
+            args=[{"user_id": 123, "items": [1, 2, 3]}, "action"],
+            kwargs={}
+        )
+
+    def test_reapply_with_complex_kwargs(self):
+        """Test reapplying task with complex JSON in kwargs"""
+        mock_task = Task()
+        mock_task.name = 'tasks.configure'
+        mock_task.args = '[]'
+        mock_task.kwargs = '{"retry": true, "timeout": 30, "options": {"key": "value"}}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.configure'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        task.apply_async.assert_called_once_with(
+            args=[],
+            kwargs={"retry": True, "timeout": 30, "options": {"key": "value"}}
+        )
+
+    def test_reapply_with_python_tuple_args(self):
+        """Test reapplying task with Python tuple string in args"""
+        mock_task = Task()
+        mock_task.name = 'tasks.tuple_task'
+        mock_task.args = '(1, 2, 3)'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.tuple_task'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        task.apply_async.assert_called_once_with(args=(1, 2, 3), kwargs={})
+
+    def test_reapply_with_python_dict_kwargs(self):
+        """Test reapplying task with Python dict string in kwargs"""
+        mock_task = Task()
+        mock_task.name = 'tasks.dict_task'
+        mock_task.args = '[]'
+        mock_task.kwargs = "{'count': 5, 'enabled': True}"
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.dict_task'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        task.apply_async.assert_called_once_with(
+            args=[],
+            kwargs={'count': 5, 'enabled': True}
+        )
+
+    def test_reapply_json_serialization_in_response(self):
+        """Test that response is properly JSON serialized"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '[1, 2]'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            task.apply_async = Mock(return_value=AsyncResult('test-task-123'))
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(200, r.code)
+        body = json.loads(r.body.decode('utf-8'))
+        self.assertIn('task-id', body)
+        self.assertEqual(body['task-id'], 'test-task-123')
+
+        self.assertIsInstance(body, dict)
+
+
 class TaskTests(BaseApiTestCase):
     def setUp(self):
         self.app = super().get_app()

From a550f13e37eb9d1631de4bc489a0aed98f8ecb1d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Asif=20Saif=20Uddin=20=7B=22Auvi=22=3A=22=E0=A6=85?=
 =?UTF-8?q?=E0=A6=AD=E0=A6=BF=22=7D?= <auvipy@gmail.com>
Date: Wed, 7 Jan 2026 18:27:46 +0600
Subject: [PATCH 6/9] Update flower/utils/tasks.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
---
 flower/utils/tasks.py | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/flower/utils/tasks.py b/flower/utils/tasks.py
index a6479956d..a846cabb8 100644
--- a/flower/utils/tasks.py
+++ b/flower/utils/tasks.py
@@ -100,14 +100,14 @@ def parse_kwargs(kwargs):
     try:
         # Attempt to parse JSON
         return json.loads(kwargs)
-    except json.JSONDecodeError:
-        try:
-            # Fallback for stringified dictionaries
-            if kwargs.startswith('{') and kwargs.endswith('}'):
+    except json.JSONDecodeError as json_err:
+        # Fallback for stringified dictionaries
+        if kwargs.startswith('{') and kwargs.endswith('}'):
+            try:
                 return ast.literal_eval(kwargs)
-        except (ValueError, SyntaxError):
-            return {}
-    return {}
+            except (ValueError, SyntaxError) as literal_err:
+                raise ValueError(f"Could not parse kwargs: {kwargs!r}") from literal_err
+        raise ValueError(f"Could not parse kwargs: {kwargs!r}") from json_err
 
 def make_json_serializable(obj):
     """

From 598c274bf867a392dc07610fccc6e190acd235f0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Asif=20Saif=20Uddin=20=7B=22Auvi=22=3A=22=E0=A6=85?=
 =?UTF-8?q?=E0=A6=AD=E0=A6=BF=22=7D?= <auvipy@gmail.com>
Date: Wed, 7 Jan 2026 18:59:44 +0600
Subject: [PATCH 7/9] Update flower/api/tasks.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
---
 flower/api/tasks.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flower/api/tasks.py b/flower/api/tasks.py
index 6a9c5dd73..332f2e22b 100644
--- a/flower/api/tasks.py
+++ b/flower/api/tasks.py
@@ -666,7 +666,7 @@ async def post(self, taskid):
         try:
             args = parse_args(task.args)
             kwargs = parse_kwargs(task.kwargs)
-        except Exception as exc:
+        except (ValueError, json.JSONDecodeError, SyntaxError, TypeError) as exc:
             logger.error("Error parsing task arguments: %s", exc)
             raise HTTPError(400, f"Invalid task arguments: {str(exc)}") from exc
 

From 3bc8db2e52bd16bf62e74aeb27d466cf42c628ce Mon Sep 17 00:00:00 2001
From: yashpapa6969 <yashgupta.yg856@gmail.com>
Date: Thu, 9 Jul 2026 03:48:53 +0530
Subject: [PATCH 8/9] Address review comments on task reapply feature

- parse_args/parse_kwargs: single literal parser (json.loads then
  ast.literal_eval only - no code execution), input size cap, and strict
  result-type validation; truncated reprs (including '...') now raise
  ValueError instead of silently reapplying wrong arguments
- remove unreachable stringified-tuple branch inside the JSON parse path
- make_json_serializable: convert tuples/sets to lists and dates to
  isoformat; raise TypeError for values with no JSON equivalent
- TaskReapply: full API docstring matching other endpoints, narrow
  exception handling for argument parsing (400), apply_async isolated in
  its own try block (500)
- flower.js: guard missing task-id in reapply response; show a generic
  error alert and log details to the console instead of echoing raw
  server responses
- task.html: fix elif indentation on the Retry button block
- tests: cover truncated/non-list args, non-dict kwargs and
  non-serializable values; add direct unit tests for parse_args,
  parse_kwargs and make_json_serializable

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 flower/api/tasks.py            |  66 ++++++++++++--------
 flower/static/js/flower.js     |  34 +++++++----
 flower/templates/task.html     |   6 +-
 flower/utils/tasks.py          |  98 +++++++++++++++++++-----------
 tests/unit/api/test_tasks.py   |  73 ++++++++++++++++++++---
 tests/unit/utils/test_tasks.py | 106 +++++++++++++++++++++++++++++++++
 6 files changed, 300 insertions(+), 83 deletions(-)
 create mode 100644 tests/unit/utils/test_tasks.py

diff --git a/flower/api/tasks.py b/flower/api/tasks.py
index 332f2e22b..43ad3f0e3 100644
--- a/flower/api/tasks.py
+++ b/flower/api/tasks.py
@@ -14,7 +14,6 @@
 
 from ..utils import tasks
 from ..utils.broker import Broker
-from ..utils.tasks import parse_args, parse_kwargs, make_json_serializable
 from . import BaseApiHandler
 
 logger = logging.getLogger(__name__)
@@ -638,49 +637,68 @@ def get(self, taskid):
 
         self.write(response)
 
+
 class TaskReapply(BaseTaskHandler):
     @web.authenticated
     async def post(self, taskid):
         """
-        Get task info and reapply the task with the same arguments.
+Reapply a task with its original arguments
+
+**Example request**:
+
+.. sourcecode:: http
+
+  POST /api/task/reapply/91396550-c228-4111-9da4-9d88cfd5ddc6 HTTP/1.1
+  Content-Length: 0
+  Host: localhost:5555
+
+**Example response**:
+
+.. sourcecode:: http
+
+  HTTP/1.1 200 OK
+  Content-Length: 51
+  Content-Type: application/json; charset=UTF-8
 
-        :param taskid: ID of the task to reapply.
+  {
+      "task-id": "aef8138a-4b2d-42fa-9c34-4ea23b1ffa9e"
+  }
+
+:reqheader Authorization: optional OAuth token to authenticate
+:statuscode 200: no error
+:statuscode 400: task has no name or its original arguments cannot be restored
+:statuscode 401: unauthorized request
+:statuscode 404: unknown task
+:statuscode 500: failed to reapply the task
         """
-        # Get original task info
         task = tasks.get_task_by_id(self.application.events, taskid)
         if not task:
             raise HTTPError(404, f"Unknown task '{taskid}'")
 
-        # Get task name
         taskname = task.name
         if not taskname:
             raise HTTPError(400, "Cannot reapply task with no name")
 
         try:
-            # Get the task object from registered tasks
             task_obj = self.capp.tasks[taskname]
         except KeyError as exc:
             raise HTTPError(404, f"Unknown task '{taskname}'") from exc
 
-        # Parse args and kwargs from the original task
         try:
-            args = parse_args(task.args)
-            kwargs = parse_kwargs(task.kwargs)
-        except (ValueError, json.JSONDecodeError, SyntaxError, TypeError) as exc:
-            logger.error("Error parsing task arguments: %s", exc)
-            raise HTTPError(400, f"Invalid task arguments: {str(exc)}") from exc
+            args = tasks.make_json_serializable(tasks.parse_args(task.args))
+            kwargs = tasks.make_json_serializable(tasks.parse_kwargs(task.kwargs))
+        except (ValueError, TypeError) as exc:
+            logger.error("Cannot reapply task '%s': %s", taskid, exc)
+            raise HTTPError(400, f"Invalid task arguments: {exc}") from exc
 
-        # Apply the task with original arguments
         try:
-            # Ensure args and kwargs are JSON serializable
-            args = make_json_serializable(args)
-            kwargs = make_json_serializable(kwargs)
-
             result = task_obj.apply_async(args=args, kwargs=kwargs)
-            response = {'task-id': result.task_id}
-            if self.backend_configured(result):
-                response.update(state=result.state)
-            self.write(response)
-        except Exception as exc:
-            logger.error("Error reapplying task with args=%s, kwargs=%s: %s", args, kwargs, str(exc))
-            raise HTTPError(500, f"Error reapplying task: {str(exc)}") from exc
+        except Exception as exc:  # broker and serializer failures are heterogeneous
+            logger.exception("Error reapplying task '%s' with args=%s, kwargs=%s",
+                             taskid, args, kwargs)
+            raise HTTPError(500, f"Error reapplying task: {exc}") from exc
+
+        response = {'task-id': result.task_id}
+        if self.backend_configured(result):
+            response.update(state=result.state)
+        self.write(response)
diff --git a/flower/static/js/flower.js b/flower/static/js/flower.js
index b1370673a..dbdbba78d 100644
--- a/flower/static/js/flower.js
+++ b/flower/static/js/flower.js
@@ -694,33 +694,41 @@ var flower = (function () {
         const $button = $(this);
         const $spinner = $button.find('.spinner-border');
         const taskId = $('#taskid').text();
-    
+
         if (!taskId) {
             show_alert('Task ID is missing. Cannot proceed.', 'danger');
             return;
         }
-    
-        // Show loading state
+
+        function resetButton() {
+            $button.prop('disabled', false);
+            $spinner.addClass('d-none');
+        }
+
+        // Show loading state while the task is reapplied
         $button.prop('disabled', true);
         $spinner.removeClass('d-none');
-    
-        // Reapply the task using the reapply endpoint
+
         $.ajax({
             type: 'POST',
             url: url_prefix() + '/api/task/reapply/' + taskId,
+            dataType: 'json',
             success: function (response) {
-                show_alert(`Task ${taskId} has been retried (new task ID: ${response['task-id']})`, 'success');
-                // Optionally reload the page after success
+                const newTaskId = response && response['task-id'];
+                if (!newTaskId) {
+                    show_alert('Task retry did not return a new task ID.', 'danger');
+                    resetButton();
+                    return;
+                }
+                show_alert(`Task ${taskId} has been retried (new task ID: ${newTaskId})`, 'success');
                 setTimeout(() => location.reload(), 1500);
             },
-            error: function (response) {
-                show_alert(response.responseText || 'Failed to retry task', 'danger');
-                // Reset button state on error
-                $button.prop('disabled', false);
-                $spinner.addClass('d-none');
+            error: function (data) {
+                console.error('Failed to retry task', taskId, data.status, data.responseText);
+                show_alert('Failed to retry task. Check the flower logs for details.', 'danger');
+                resetButton();
             }
         });
     });
 
-
 }(jQuery));
diff --git a/flower/templates/task.html b/flower/templates/task.html
index 5a8c3732b..b052bed26 100644
--- a/flower/templates/task.html
+++ b/flower/templates/task.html
@@ -16,8 +16,8 @@ <h2>{{ getattr(task, 'name', None) }}
                 <button class="btn btn-danger float-end" id="task-terminate">Terminate</button>
             {% elif task.state == "RECEIVED" or task.state == "RETRY" %}
                 <button class="btn btn-danger float-end" id="task-revoke">Revoke</button>
-                {% elif task.state == "FAILURE" %}
-                <button class="btn btn-warning float-end" id="task-retry" data-bs-toggle="button">
+            {% elif task.state == "FAILURE" %}
+                <button class="btn btn-warning float-end" id="task-retry">
                     <span class="spinner-border spinner-border-sm d-none" role="status" aria-hidden="true"></span>
                     Retry
                 </button>
@@ -94,4 +94,4 @@ <h2>{{ getattr(task, 'name', None) }}
       </div>
     </div>
   </div>
-{% end %}
\ No newline at end of file
+{% end %}
diff --git a/flower/utils/tasks.py b/flower/utils/tasks.py
index a846cabb8..5899529df 100644
--- a/flower/utils/tasks.py
+++ b/flower/utils/tasks.py
@@ -1,7 +1,7 @@
+import ast
 import datetime
-import time
 import json
-import ast
+import time
 
 from .search import parse_search_terms, satisfies_search_terms
 
@@ -71,52 +71,82 @@ def get_task_by_id(events, task_id):
 def as_dict(task):
     return task.as_dict()
 
+
+# Upper bound on the size of a stored args/kwargs string we are willing to
+# parse. Celery truncates event args far below this; anything larger is not
+# a legitimate task representation and is rejected before parsing.
+MAX_ARG_LENGTH = 65536
+
+
+def _parse_literal(value, kind):
+    """
+    Parse a string holding a JSON value or a Python literal (the two formats
+    celery events use for task args/kwargs) into a Python value.
+
+    Only ``json.loads`` and ``ast.literal_eval`` are used, so no code is ever
+    executed. Raises ValueError for anything that is not a plain literal,
+    including truncated reprs such as ``'(1, 2, ...'``.
+    """
+    if len(value) > MAX_ARG_LENGTH:
+        raise ValueError(f"Task {kind} too long to parse ({len(value)} bytes)")
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        pass
+    try:
+        return ast.literal_eval(value)
+    except (ValueError, TypeError, SyntaxError, MemoryError, RecursionError) as exc:
+        raise ValueError(f"Could not parse task {kind}: {value!r}") from exc
+
+
 def parse_args(args):
     """
-    Parse and process the `args` of the task.
+    Parse the string representation of a task's positional arguments
+    into a list.
+
+    Raises ValueError if the string cannot be restored to the original
+    arguments (e.g. it was truncated by celery), so callers never reapply
+    a task with wrong arguments.
     """
     if not args:
         return []
-    try:
-        # Attempt to parse JSON
-        parsed_args = json.loads(args)
-        if isinstance(parsed_args, str) and parsed_args.startswith('(') and parsed_args.endswith(')'):
-            return ast.literal_eval(parsed_args)  # Handle stringified tuples safely
-        return parsed_args
-    except (json.JSONDecodeError, SyntaxError):
-        # Fallback for stringified tuples or ellipsis
-        if args == '...':
-            return [...]
-        if args.startswith('(') and args.endswith(')'):
-            return ast.literal_eval(args)
-        return [args]
+    parsed = _parse_literal(args, 'args')
+    if isinstance(parsed, tuple):
+        parsed = list(parsed)
+    if not isinstance(parsed, list):
+        raise ValueError(f"Task args must be a list or tuple: {args!r}")
+    return parsed
+
 
 def parse_kwargs(kwargs):
     """
-    Parse and process the `kwargs` of the task.
+    Parse the string representation of a task's keyword arguments
+    into a dict.
+
+    Raises ValueError if the string cannot be restored to the original
+    keyword arguments.
     """
     if not kwargs:
         return {}
-    try:
-        # Attempt to parse JSON
-        return json.loads(kwargs)
-    except json.JSONDecodeError as json_err:
-        # Fallback for stringified dictionaries
-        if kwargs.startswith('{') and kwargs.endswith('}'):
-            try:
-                return ast.literal_eval(kwargs)
-            except (ValueError, SyntaxError) as literal_err:
-                raise ValueError(f"Could not parse kwargs: {kwargs!r}") from literal_err
-        raise ValueError(f"Could not parse kwargs: {kwargs!r}") from json_err
+    parsed = _parse_literal(kwargs, 'kwargs')
+    if not isinstance(parsed, dict):
+        raise ValueError(f"Task kwargs must be a dict: {kwargs!r}")
+    return parsed
+
 
 def make_json_serializable(obj):
     """
-    Recursively replace non-serializable types with JSON-serializable alternatives.
+    Recursively convert parsed argument values to JSON-serializable types.
+
+    Raises TypeError for values with no JSON equivalent so callers fail
+    loudly instead of reapplying a task with corrupted arguments.
     """
-    if isinstance(obj, list):
+    if obj is None or isinstance(obj, (str, int, float, bool)):
+        return obj
+    if isinstance(obj, (list, tuple, set)):
         return [make_json_serializable(item) for item in obj]
-    elif isinstance(obj, dict):
+    if isinstance(obj, dict):
         return {key: make_json_serializable(value) for key, value in obj.items()}
-    elif obj is Ellipsis:
-        return None  # Replace `...` with `null`
-    return obj  # Return the object if it's already serializable
\ No newline at end of file
+    if isinstance(obj, (datetime.datetime, datetime.date)):
+        return obj.isoformat()
+    raise TypeError(f"Value of type {type(obj).__name__} is not JSON serializable: {obj!r}")
diff --git a/tests/unit/api/test_tasks.py b/tests/unit/api/test_tasks.py
index 96d3c6618..22ef5fa18 100644
--- a/tests/unit/api/test_tasks.py
+++ b/tests/unit/api/test_tasks.py
@@ -150,18 +150,74 @@ def test_reapply_unknown_task_name(self):
         self.assertEqual(404, r.code)
 
     def test_reapply_invalid_args(self):
-        """Test reapplying a task with invalid args returns 400"""
+        """Test reapplying a task with unparseable args returns 400"""
         mock_task = Task()
         mock_task.name = 'tasks.add'
         mock_task.args = 'invalid json'
         mock_task.kwargs = '{}'
 
         with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            self._app.capp.tasks['tasks.add'] = Mock()
-            with patch('flower.api.tasks.parse_args', side_effect=ValueError("Invalid args")):
-                r = self.post('/api/task/reapply/123', body='')
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(400, r.code)
+        task.apply_async.assert_not_called()
+
+    def test_reapply_truncated_args(self):
+        """Test reapplying a task whose args repr was truncated returns 400"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = "(1, 2, 'long-value..."
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            r = self.post('/api/task/reapply/123', body='')
 
         self.assertEqual(400, r.code)
+        task.apply_async.assert_not_called()
+
+    def test_reapply_non_list_args(self):
+        """Test reapplying a task whose args are not a list/tuple returns 400"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '"just a string"'
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(400, r.code)
+        task.apply_async.assert_not_called()
+
+    def test_reapply_non_dict_kwargs(self):
+        """Test reapplying a task whose kwargs are not a dict returns 400"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '[]'
+        mock_task.kwargs = '[1, 2]'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(400, r.code)
+        task.apply_async.assert_not_called()
+
+    def test_reapply_non_serializable_args(self):
+        """Test reapplying a task with non-JSON-serializable args returns 400"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = "(b'raw-bytes',)"
+        mock_task.kwargs = '{}'
+
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            task = self._app.capp.tasks['tasks.add'] = Mock()
+            r = self.post('/api/task/reapply/123', body='')
+
+        self.assertEqual(400, r.code)
+        task.apply_async.assert_not_called()
 
     def test_reapply_apply_async_error(self):
         """Test handling error during apply_async returns 500"""
@@ -193,7 +249,7 @@ def test_reapply_with_empty_args(self):
         task.apply_async.assert_called_once_with(args=[], kwargs={})
 
     def test_reapply_with_ellipsis_args(self):
-        """Test reapplying a task with ellipsis in args"""
+        """Test reapplying a task with truncated ('...') args returns 400"""
         mock_task = Task()
         mock_task.name = 'tasks.test'
         mock_task.args = '...'
@@ -201,11 +257,10 @@ def test_reapply_with_ellipsis_args(self):
 
         with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
             task = self._app.capp.tasks['tasks.test'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
             r = self.post('/api/task/reapply/123', body='')
 
-        self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(args=[None], kwargs={})
+        self.assertEqual(400, r.code)
+        task.apply_async.assert_not_called()
 
     def test_reapply_with_nested_json_args(self):
         """Test reapplying task with nested JSON structures in args"""
@@ -256,7 +311,7 @@ def test_reapply_with_python_tuple_args(self):
             r = self.post('/api/task/reapply/123', body='')
 
         self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(args=(1, 2, 3), kwargs={})
+        task.apply_async.assert_called_once_with(args=[1, 2, 3], kwargs={})
 
     def test_reapply_with_python_dict_kwargs(self):
         """Test reapplying task with Python dict string in kwargs"""
diff --git a/tests/unit/utils/test_tasks.py b/tests/unit/utils/test_tasks.py
new file mode 100644
index 000000000..4951b5094
--- /dev/null
+++ b/tests/unit/utils/test_tasks.py
@@ -0,0 +1,106 @@
+import datetime
+import unittest
+
+from flower.utils.tasks import (MAX_ARG_LENGTH, make_json_serializable,
+                                parse_args, parse_kwargs)
+
+
+class TestParseArgs(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual([], parse_args(None))
+        self.assertEqual([], parse_args(''))
+
+    def test_json_list(self):
+        self.assertEqual([1, 2], parse_args('[1, 2]'))
+        self.assertEqual([{'a': [1]}, 'b'], parse_args('[{"a": [1]}, "b"]'))
+
+    def test_python_tuple_repr(self):
+        self.assertEqual([1, 2, 3], parse_args('(1, 2, 3)'))
+        self.assertEqual(['x'], parse_args("('x',)"))
+        self.assertEqual([], parse_args('()'))
+
+    def test_python_list_repr(self):
+        self.assertEqual([1, None, True], parse_args('[1, None, True]'))
+
+    def test_truncated_repr_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_args("(1, 2, 'long-va...")
+        with self.assertRaises(ValueError):
+            parse_args('...')
+
+    def test_non_sequence_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_args('"a string"')
+        with self.assertRaises(ValueError):
+            parse_args('42')
+        with self.assertRaises(ValueError):
+            parse_args('{"a": 1}')
+
+    def test_garbage_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_args('not a literal')
+
+    def test_code_not_executed(self):
+        with self.assertRaises(ValueError):
+            parse_args('__import__("os").system("true")')
+
+    def test_oversized_input_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_args('[' + '1,' * MAX_ARG_LENGTH + ']')
+
+
+class TestParseKwargs(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual({}, parse_kwargs(None))
+        self.assertEqual({}, parse_kwargs(''))
+
+    def test_json_object(self):
+        self.assertEqual({'retry': True, 'n': 3},
+                         parse_kwargs('{"retry": true, "n": 3}'))
+
+    def test_python_dict_repr(self):
+        self.assertEqual({'count': 5, 'enabled': True},
+                         parse_kwargs("{'count': 5, 'enabled': True}"))
+        self.assertEqual({}, parse_kwargs('{}'))
+
+    def test_non_dict_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_kwargs('[1, 2]')
+        with self.assertRaises(ValueError):
+            parse_kwargs("{'a'}")  # a set literal, not a dict
+
+    def test_garbage_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_kwargs('invalid json')
+        with self.assertRaises(ValueError):
+            parse_kwargs("{'a': 1, ...")
+
+
+class TestMakeJsonSerializable(unittest.TestCase):
+    def test_scalars_pass_through(self):
+        for value in (None, 'a', 1, 1.5, True):
+            self.assertEqual(value, make_json_serializable(value))
+
+    def test_containers_converted(self):
+        self.assertEqual([1, 2], make_json_serializable((1, 2)))
+        self.assertEqual([1], make_json_serializable({1}))
+        self.assertEqual({'a': [1, 2]}, make_json_serializable({'a': (1, 2)}))
+        self.assertEqual([[1], {'b': 2}],
+                         make_json_serializable([(1,), {'b': 2}]))
+
+    def test_datetime_converted(self):
+        dt = datetime.datetime(2026, 7, 9, 12, 0, 0)
+        self.assertEqual('2026-07-09T12:00:00', make_json_serializable(dt))
+        self.assertEqual('2026-07-09', make_json_serializable(datetime.date(2026, 7, 9)))
+
+    def test_non_serializable_rejected(self):
+        with self.assertRaises(TypeError):
+            make_json_serializable(b'bytes')
+        with self.assertRaises(TypeError):
+            make_json_serializable(...)
+        with self.assertRaises(TypeError):
+            make_json_serializable([1, object()])
+
+
+if __name__ == '__main__':
+    unittest.main()

From 301cf65c453019299913eb25fa5a0e376d9926b6 Mon Sep 17 00:00:00 2001
From: yashpapa6969 <yashgupta.yg856@gmail.com>
Date: Thu, 9 Jul 2026 04:14:46 +0530
Subject: [PATCH 9/9] Harden task reapply against truncated and unrestorable
 arguments

Follow-up hardening found by adversarial verification of the review
fixes:

- reject parsed strings ending in '...': celery's saferepr truncates
  long strings inside their quotes (e.g. "(7, 'Bearer...')" for a long
  token), which parses cleanly but would reapply the task with a
  corrupted argument
- catch RecursionError/MemoryError from json.loads too, so
  pathologically nested input returns 400 instead of an unhandled 500
- make_json_serializable: raise TypeError for sets and non-string dict
  keys instead of silently changing the type the retried task receives
- TaskReapply: use capp.send_task instead of the local task registry,
  so reapply works when flower runs without the application's task
  modules importable; preserve the original routing_key/exchange from
  the event so the retry lands on the same queue

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 flower/api/tasks.py            |  17 ++-
 flower/utils/tasks.py          |  43 +++++++-
 tests/unit/api/test_tasks.py   | 193 ++++++++++++++++++++-------------
 tests/unit/utils/test_tasks.py |  28 ++++-
 4 files changed, 193 insertions(+), 88 deletions(-)

diff --git a/flower/api/tasks.py b/flower/api/tasks.py
index 43ad3f0e3..a0a5cca06 100644
--- a/flower/api/tasks.py
+++ b/flower/api/tasks.py
@@ -679,11 +679,6 @@ async def post(self, taskid):
         if not taskname:
             raise HTTPError(400, "Cannot reapply task with no name")
 
-        try:
-            task_obj = self.capp.tasks[taskname]
-        except KeyError as exc:
-            raise HTTPError(404, f"Unknown task '{taskname}'") from exc
-
         try:
             args = tasks.make_json_serializable(tasks.parse_args(task.args))
             kwargs = tasks.make_json_serializable(tasks.parse_kwargs(task.kwargs))
@@ -691,8 +686,18 @@ async def post(self, taskid):
             logger.error("Cannot reapply task '%s': %s", taskid, exc)
             raise HTTPError(400, f"Invalid task arguments: {exc}") from exc
 
+        # Preserve the original routing when the event carries it, so the
+        # retry runs on the same queue as the original task.
+        options = {}
+        if getattr(task, 'routing_key', None):
+            options['routing_key'] = task.routing_key
+        if getattr(task, 'exchange', None):
+            options['exchange'] = task.exchange
+
         try:
-            result = task_obj.apply_async(args=args, kwargs=kwargs)
+            # send_task works even when the task module is not importable
+            # by flower (same approach as TaskSendTask above)
+            result = self.capp.send_task(taskname, args=args, kwargs=kwargs, **options)
         except Exception as exc:  # broker and serializer failures are heterogeneous
             logger.exception("Error reapplying task '%s' with args=%s, kwargs=%s",
                              taskid, args, kwargs)
diff --git a/flower/utils/tasks.py b/flower/utils/tasks.py
index 5899529df..1294c5549 100644
--- a/flower/utils/tasks.py
+++ b/flower/utils/tasks.py
@@ -93,12 +93,37 @@ def _parse_literal(value, kind):
         return json.loads(value)
     except json.JSONDecodeError:
         pass
+    except (MemoryError, RecursionError) as exc:
+        raise ValueError(f"Could not parse task {kind}: {value!r}") from exc
     try:
         return ast.literal_eval(value)
     except (ValueError, TypeError, SyntaxError, MemoryError, RecursionError) as exc:
         raise ValueError(f"Could not parse task {kind}: {value!r}") from exc
 
 
+def _reject_truncated(obj, kind):
+    """
+    Reject values truncated by celery.
+
+    Celery stores event args/kwargs via saferepr, which truncates long
+    strings by cutting them and appending '...' inside the quotes, so a
+    truncated repr like "(7, 'Bearer...')" still parses cleanly. Refuse
+    such values rather than reapply a task with corrupted arguments.
+    A legitimate string ending in '...' is indistinguishable from a
+    truncated one, so it is rejected too.
+    """
+    if isinstance(obj, str):
+        if obj.endswith('...'):
+            raise ValueError(f"Task {kind} appear truncated by celery: {obj!r}")
+    elif isinstance(obj, (list, tuple, set)):
+        for item in obj:
+            _reject_truncated(item, kind)
+    elif isinstance(obj, dict):
+        for key, value in obj.items():
+            _reject_truncated(key, kind)
+            _reject_truncated(value, kind)
+
+
 def parse_args(args):
     """
     Parse the string representation of a task's positional arguments
@@ -115,6 +140,7 @@ def parse_args(args):
         parsed = list(parsed)
     if not isinstance(parsed, list):
         raise ValueError(f"Task args must be a list or tuple: {args!r}")
+    _reject_truncated(parsed, 'args')
     return parsed
 
 
@@ -131,6 +157,7 @@ def parse_kwargs(kwargs):
     parsed = _parse_literal(kwargs, 'kwargs')
     if not isinstance(parsed, dict):
         raise ValueError(f"Task kwargs must be a dict: {kwargs!r}")
+    _reject_truncated(parsed, 'kwargs')
     return parsed
 
 
@@ -138,15 +165,23 @@ def make_json_serializable(obj):
     """
     Recursively convert parsed argument values to JSON-serializable types.
 
-    Raises TypeError for values with no JSON equivalent so callers fail
-    loudly instead of reapplying a task with corrupted arguments.
+    Raises TypeError for values with no faithful JSON equivalent (sets,
+    bytes, non-string dict keys, ...) so callers fail loudly instead of
+    reapplying a task with arguments of the wrong type. Tuples become
+    lists because that is what the original task received from the JSON
+    serializer anyway.
     """
     if obj is None or isinstance(obj, (str, int, float, bool)):
         return obj
-    if isinstance(obj, (list, tuple, set)):
+    if isinstance(obj, (list, tuple)):
         return [make_json_serializable(item) for item in obj]
     if isinstance(obj, dict):
-        return {key: make_json_serializable(value) for key, value in obj.items()}
+        result = {}
+        for key, value in obj.items():
+            if not isinstance(key, str):
+                raise TypeError(f"Dict key {key!r} is not JSON serializable")
+            result[key] = make_json_serializable(value)
+        return result
     if isinstance(obj, (datetime.datetime, datetime.date)):
         return obj.isoformat()
     raise TypeError(f"Value of type {type(obj).__name__} is not JSON serializable: {obj!r}")
diff --git a/tests/unit/api/test_tasks.py b/tests/unit/api/test_tasks.py
index 22ef5fa18..7dea89385 100644
--- a/tests/unit/api/test_tasks.py
+++ b/tests/unit/api/test_tasks.py
@@ -98,6 +98,15 @@ def get_task_by_id(events, task_id):
 
 
 class TaskReapplyTests(BaseApiTestCase):
+    def reapply(self, mock_task, taskid='123',
+                send_task_result=AsyncResult('new-task-id')):
+        """POST /api/task/reapply with mocked event state and send_task."""
+        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
+            with patch.object(self._app.capp, 'send_task',
+                              return_value=send_task_result) as send_task:
+                r = self.post(f'/api/task/reapply/{taskid}', body='')
+        return r, send_task
+
     def test_reapply_success(self):
         """Test successfully reapplying a task"""
         mock_task = Task()
@@ -105,49 +114,65 @@ def test_reapply_success(self):
         mock_task.args = '[1, 2]'
         mock_task.kwargs = '{"multiply": 2}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(200, r.code)
         body = json.loads(r.body.decode('utf-8'))
         self.assertIn('task-id', body)
-        task.apply_async.assert_called_once_with(
-            args=[1, 2], kwargs={"multiply": 2}
+        send_task.assert_called_once_with(
+            'tasks.add', args=[1, 2], kwargs={"multiply": 2}
         )
 
     def test_reapply_task_not_found(self):
         """Test reapplying a non-existent task returns 404"""
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=None):
-            r = self.post('/api/task/reapply/nonexistent', body='')
+        r, send_task = self.reapply(None, taskid='nonexistent')
 
         self.assertEqual(404, r.code)
+        send_task.assert_not_called()
 
     def test_reapply_task_no_name(self):
         """Test reapplying a task with no name returns 400"""
         mock_task = Task()
         mock_task.name = None
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
+        send_task.assert_not_called()
+
+    def test_reapply_unregistered_task(self):
+        """Test a task not registered in flower's app is still reapplied
 
-    def test_reapply_unknown_task_name(self):
-        """Test reapplying a task that is not registered returns 404"""
+        Flower often runs without the application's task modules being
+        importable; send_task does not require local registration.
+        """
         mock_task = Task()
         mock_task.name = 'unknown.task'
         mock_task.args = '[]'
         mock_task.kwargs = '{}'
 
-        if 'unknown.task' in self._app.capp.tasks:
-            del self._app.capp.tasks['unknown.task']
+        self.assertNotIn('unknown.task', self._app.capp.tasks)
+        r, send_task = self.reapply(mock_task)
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            r = self.post('/api/task/reapply/123', body='')
+        self.assertEqual(200, r.code)
+        send_task.assert_called_once_with('unknown.task', args=[], kwargs={})
 
-        self.assertEqual(404, r.code)
+    def test_reapply_preserves_routing(self):
+        """Test the original routing_key/exchange are reused when present"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '[1, 2]'
+        mock_task.kwargs = '{}'
+        mock_task.routing_key = 'high.priority'
+        mock_task.exchange = 'tasks'
+
+        r, send_task = self.reapply(mock_task)
+
+        self.assertEqual(200, r.code)
+        send_task.assert_called_once_with(
+            'tasks.add', args=[1, 2], kwargs={},
+            routing_key='high.priority', exchange='tasks'
+        )
 
     def test_reapply_invalid_args(self):
         """Test reapplying a task with unparseable args returns 400"""
@@ -156,12 +181,10 @@ def test_reapply_invalid_args(self):
         mock_task.args = 'invalid json'
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
-        task.apply_async.assert_not_called()
+        send_task.assert_not_called()
 
     def test_reapply_truncated_args(self):
         """Test reapplying a task whose args repr was truncated returns 400"""
@@ -170,12 +193,38 @@ def test_reapply_truncated_args(self):
         mock_task.args = "(1, 2, 'long-value..."
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
+
+        self.assertEqual(400, r.code)
+        send_task.assert_not_called()
+
+    def test_reapply_truncated_string_arg(self):
+        """Test a string arg truncated inside its quotes by celery returns 400
+
+        saferepr can truncate long strings while keeping the repr parseable,
+        e.g. "(7, 'Bearer...')" for a long token.
+        """
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = "(7, 'Bearer...')"
+        mock_task.kwargs = '{}'
+
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
-        task.apply_async.assert_not_called()
+        send_task.assert_not_called()
+
+    def test_reapply_deeply_nested_args(self):
+        """Test pathologically nested args return 400, not a 500"""
+        mock_task = Task()
+        mock_task.name = 'tasks.add'
+        mock_task.args = '[' * 3000
+        mock_task.kwargs = '{}'
+
+        r, send_task = self.reapply(mock_task)
+
+        self.assertEqual(400, r.code)
+        send_task.assert_not_called()
 
     def test_reapply_non_list_args(self):
         """Test reapplying a task whose args are not a list/tuple returns 400"""
@@ -184,12 +233,10 @@ def test_reapply_non_list_args(self):
         mock_task.args = '"just a string"'
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
-        task.apply_async.assert_not_called()
+        send_task.assert_not_called()
 
     def test_reapply_non_dict_kwargs(self):
         """Test reapplying a task whose kwargs are not a dict returns 400"""
@@ -198,38 +245,47 @@ def test_reapply_non_dict_kwargs(self):
         mock_task.args = '[]'
         mock_task.kwargs = '[1, 2]'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
-        task.apply_async.assert_not_called()
+        send_task.assert_not_called()
 
     def test_reapply_non_serializable_args(self):
         """Test reapplying a task with non-JSON-serializable args returns 400"""
+        for bad_args in ("(b'raw-bytes',)", '({1, 2},)'):
+            mock_task = Task()
+            mock_task.name = 'tasks.add'
+            mock_task.args = bad_args
+            mock_task.kwargs = '{}'
+
+            r, send_task = self.reapply(mock_task)
+
+            self.assertEqual(400, r.code, bad_args)
+            send_task.assert_not_called()
+
+    def test_reapply_non_string_dict_keys(self):
+        """Test kwargs with non-string dict keys return 400"""
         mock_task = Task()
         mock_task.name = 'tasks.add'
-        mock_task.args = "(b'raw-bytes',)"
-        mock_task.kwargs = '{}'
+        mock_task.args = '[]'
+        mock_task.kwargs = "{'mapping': {1: 'a'}}"
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
-        task.apply_async.assert_not_called()
+        send_task.assert_not_called()
 
-    def test_reapply_apply_async_error(self):
-        """Test handling error during apply_async returns 500"""
+    def test_reapply_send_task_error(self):
+        """Test handling error during send_task returns 500"""
         mock_task = Task()
         mock_task.name = 'tasks.add'
         mock_task.args = '[1, 2]'
         mock_task.kwargs = '{}'
 
         with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            task.apply_async = Mock(side_effect=Exception("Connection error"))
-            r = self.post('/api/task/reapply/123', body='')
+            with patch.object(self._app.capp, 'send_task',
+                              side_effect=Exception("Connection error")):
+                r = self.post('/api/task/reapply/123', body='')
 
         self.assertEqual(500, r.code)
 
@@ -240,13 +296,10 @@ def test_reapply_with_empty_args(self):
         mock_task.args = ''
         mock_task.kwargs = ''
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.simple'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(args=[], kwargs={})
+        send_task.assert_called_once_with('tasks.simple', args=[], kwargs={})
 
     def test_reapply_with_ellipsis_args(self):
         """Test reapplying a task with truncated ('...') args returns 400"""
@@ -255,12 +308,10 @@ def test_reapply_with_ellipsis_args(self):
         mock_task.args = '...'
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.test'] = Mock()
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(400, r.code)
-        task.apply_async.assert_not_called()
+        send_task.assert_not_called()
 
     def test_reapply_with_nested_json_args(self):
         """Test reapplying task with nested JSON structures in args"""
@@ -269,13 +320,11 @@ def test_reapply_with_nested_json_args(self):
         mock_task.args = '[{"user_id": 123, "items": [1, 2, 3]}, "action"]'
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.process'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(
+        send_task.assert_called_once_with(
+            'tasks.process',
             args=[{"user_id": 123, "items": [1, 2, 3]}, "action"],
             kwargs={}
         )
@@ -287,13 +336,11 @@ def test_reapply_with_complex_kwargs(self):
         mock_task.args = '[]'
         mock_task.kwargs = '{"retry": true, "timeout": 30, "options": {"key": "value"}}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.configure'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(
+        send_task.assert_called_once_with(
+            'tasks.configure',
             args=[],
             kwargs={"retry": True, "timeout": 30, "options": {"key": "value"}}
         )
@@ -305,13 +352,10 @@ def test_reapply_with_python_tuple_args(self):
         mock_task.args = '(1, 2, 3)'
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.tuple_task'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(args=[1, 2, 3], kwargs={})
+        send_task.assert_called_once_with('tasks.tuple_task', args=[1, 2, 3], kwargs={})
 
     def test_reapply_with_python_dict_kwargs(self):
         """Test reapplying task with Python dict string in kwargs"""
@@ -320,13 +364,11 @@ def test_reapply_with_python_dict_kwargs(self):
         mock_task.args = '[]'
         mock_task.kwargs = "{'count': 5, 'enabled': True}"
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.dict_task'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('new-task-id'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, send_task = self.reapply(mock_task)
 
         self.assertEqual(200, r.code)
-        task.apply_async.assert_called_once_with(
+        send_task.assert_called_once_with(
+            'tasks.dict_task',
             args=[],
             kwargs={'count': 5, 'enabled': True}
         )
@@ -338,10 +380,7 @@ def test_reapply_json_serialization_in_response(self):
         mock_task.args = '[1, 2]'
         mock_task.kwargs = '{}'
 
-        with patch('flower.api.tasks.tasks.get_task_by_id', return_value=mock_task):
-            task = self._app.capp.tasks['tasks.add'] = Mock()
-            task.apply_async = Mock(return_value=AsyncResult('test-task-123'))
-            r = self.post('/api/task/reapply/123', body='')
+        r, _ = self.reapply(mock_task, send_task_result=AsyncResult('test-task-123'))
 
         self.assertEqual(200, r.code)
         body = json.loads(r.body.decode('utf-8'))
diff --git a/tests/unit/utils/test_tasks.py b/tests/unit/utils/test_tasks.py
index 4951b5094..64a0e0d2b 100644
--- a/tests/unit/utils/test_tasks.py
+++ b/tests/unit/utils/test_tasks.py
@@ -48,6 +48,19 @@ def test_oversized_input_rejected(self):
         with self.assertRaises(ValueError):
             parse_args('[' + '1,' * MAX_ARG_LENGTH + ']')
 
+    def test_deeply_nested_input_rejected(self):
+        # json.loads raises RecursionError on this; must surface as ValueError
+        with self.assertRaises(ValueError):
+            parse_args('[' * 3000)
+
+    def test_string_truncated_by_saferepr_rejected(self):
+        # celery's saferepr can truncate a long string inside its quotes,
+        # leaving a parseable repr with corrupted content
+        with self.assertRaises(ValueError):
+            parse_args("(7, 'Bearer...')")
+        with self.assertRaises(ValueError):
+            parse_args('[{"token": "Bearer..."}]')
+
 
 class TestParseKwargs(unittest.TestCase):
     def test_empty(self):
@@ -75,6 +88,10 @@ def test_garbage_rejected(self):
         with self.assertRaises(ValueError):
             parse_kwargs("{'a': 1, ...")
 
+    def test_truncated_value_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_kwargs("{'token': 'Bearer...'}")
+
 
 class TestMakeJsonSerializable(unittest.TestCase):
     def test_scalars_pass_through(self):
@@ -83,7 +100,6 @@ def test_scalars_pass_through(self):
 
     def test_containers_converted(self):
         self.assertEqual([1, 2], make_json_serializable((1, 2)))
-        self.assertEqual([1], make_json_serializable({1}))
         self.assertEqual({'a': [1, 2]}, make_json_serializable({'a': (1, 2)}))
         self.assertEqual([[1], {'b': 2}],
                          make_json_serializable([(1,), {'b': 2}]))
@@ -100,6 +116,16 @@ def test_non_serializable_rejected(self):
             make_json_serializable(...)
         with self.assertRaises(TypeError):
             make_json_serializable([1, object()])
+        # sets have no faithful JSON equivalent; converting to a list would
+        # change the type the retried task receives
+        with self.assertRaises(TypeError):
+            make_json_serializable({1, 2})
+
+    def test_non_string_dict_keys_rejected(self):
+        with self.assertRaises(TypeError):
+            make_json_serializable({1: 'a'})
+        with self.assertRaises(TypeError):
+            make_json_serializable({'m': {(1, 2): 'b'}})
 
 
 if __name__ == '__main__':