celery-celery-10408-10404

total 332
drwxr-xr-x 1 root root   4096 Sep 30 20:09 .
drwxr-xr-x 1 root root   4096 Sep 30 20:08 ..
-rw-r--r-- 1 root root    343 Sep 30 20:07 .bumpversion.cfg
-rw-r--r-- 1 root root    306 Sep 30 20:07 .dockerignore
-rw-r--r-- 1 root root    227 Sep 30 20:07 .editorconfig
drwxr-xr-x 1 root root   4096 Sep 30 20:08 .git
drwxr-xr-x 1 root root   4096 Sep 30 20:07 .github
-rw-r--r-- 1 root root    437 Sep 30 20:07 .gitignore
-rw-r--r-- 1 root root   1073 Sep 30 20:07 .pre-commit-config.yaml
-rw-r--r-- 1 root root    616 Sep 30 20:07 .readthedocs.yaml
-rw-r--r-- 1 root root  44762 Sep 30 20:07 CONTRIBUTING.rst

-rw-r--r-- 1 root root   8619 Sep 30 20:07 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128887 Sep 30 20:07 Changelog.rst
-rw-r--r-- 1 root root   2717 Sep 30 20:07 LICENSE
-rw-r--r-- 1 root root    709 Sep 30 20:07 MANIFEST.in
-rw-r--r-- 1 root root   5866 Sep 30 20:07 Makefile
-rw-r-
…[6638 chars omitted]…
, result_from_tuple(x == self.app)
./t/unit/tasks/test_result.py:1071:        second_result = result_from_tuple(result.as_tuple(), self.app)
./t/unit/tasks/test_result.py:1075:    def test_GroupResult_as_tuple(self):
./t/unit/tasks/test_result.py:1083:        (result_id, parent_tuple), group_results = result.as_tuple()
./t/unit/tasks/test_result.py:1085:        assert parent_tuple == parent.as_tuple()

celery/canvas.py:356:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:370:    def apply_async(self, args=None, kwargs=None, route_name=None, **options):
celery/canvas.py:935:class _chain(Signature):
celery/canvas.py:1036:    def apply_async(self, args=None, kwargs=None, **options):
celery/canvas.py:1294:    def apply(self, args=None, kwargs=None, **options):
celery/canvas.py:1322:class chain(_chain):
celery/canvas.py:1398:    def apply_async(self, args=None, kwargs=None, **opts):
celery/canvas.py:1455:    def apply_async(self, args=None, kwargs=None, **opts):
celery/canvas.py:1471:    def apply_chunks(cls, task, it, n, app=None):
celery/canvas.py:1492:class group(Signature):
celery/canvas.py:1616:    def apply_async(self, args=None, kwargs=None, add_to_parent=True,
celery/canvas.py:1633:        results = list(self._apply_tasks(tasks, producer, app, p,

celery/ca
…[8630 chars omitted]…
    res, nodes = r
        id, parent = res if isinstance(res, (list, tuple)) else (res, None)
        if parent:
            parent = result_from_tuple(parent, app)

        if nodes is not None:
            return app.GroupResult(
                id, [result_from_tuple(child, app) for child in nodes],
                parent=parent,
            )

        return Result(id, parent=parent)
    return r

else:
            # Also supports using task class/instance instead of string name.
            try:
                task_name = task.name
            except AttributeError:
                task_name = task
            else:
                self._type = task

            super().__init__(
                task=task_name, args=tuple(args or ()),
                kwargs=kwargs or {},

                options=dict(options or {}, **ex),
                subtask_type=subtask_type,
                immutable=immutable,
            )

    def __call__(self, *partial_args, **partial_kwargs):
        """Call the task directly (in the current process)."""
        args, kwargs, _ = self._merge(partial_args, partial_kwargs, None)
        return self.type(*args, **kwargs)

    def delay(self, *partial_args, **partial_kwargs):
        """Shortcut to :meth:`apply_async` using star arguments."""
 
…[8630 chars omitted]…
race.py:109:IGNORED = states.IGNORED
celery/app/trace.py:110:REJECTED = states.REJECTED
celery/app/task.py:951:        return EagerResult(task_id, retval, state, traceback=tb, name=self.name)
celery/app/trace.py:109:IGNORED = states.IGNORED
celery/app/trace.py:110:REJECTED = states.REJECTED
celery/events/state.py:75:    'rejected': states.REJECTED,
celery/result.py:1026:class EagerResult(AsyncResult):

elif 'chain' not in options:
        return ChainMap({'chain': tasks}, options)
    elif tasks is not None:
        # chain option may already be set, resulting in
        # "multiple values for keyword argument 'chain'" error.
        # Issue #3379.
        # If a chain already exists, we need to extend it with the next
        # tasks in the chain.
        # Issue #5354.
        # WARNING: Be careful not to mutate `options['chain']`.
        return ChainMap({'chain': options['chain'] + tasks},
                        options)

@Signature.register_type(name='chain')
class _chain(Signature):
    tasks = getitem_property('kwargs.tasks', 'Tasks in chain.')

    @classmethod
    def from_dict(cls, d, app=None):
        tasks = d['kwargs']['tasks']
        if tasks:
            if isinstance(tasks, tuple):  # aaaargh
                tasks = d['kwargs']['tasks'] = list(tasks)
        
…[8630 chars omitted]…
      starkwargs.setdefault('app', self.app)
t/unit/tasks/test_canvas.py:532:        c._use_link = True
t/unit/tasks/test_canvas.py:546:        c2._use_link = True
t/unit/tasks/test_canvas.py:571:        c2._use_link = False
t/unit/tasks/test_canvas.py:532:        c._use_link = True
t/unit/tasks/test_canvas.py:546:        c2._use_link = True
t/unit/tasks/test_canvas.py:571:        c2._use_link = False

The tasks list would be cloned from the chain's tasks.
        All of the chain callbacks would be added to the last task in the (cloned) chain.
        All of the tasks would be linked to the same error callback
        as the chain itself, to ensure that the correct error callback is called
        if any of the (cloned) tasks of the chain fail.
        """
        # Clone chain's tasks assigning signatures from link_error
        # to each task and adding the chain's links to the last task.
        tasks = [t.clone() for t in self.tasks]
        for sig in maybe_list(self.options.get('link')) or []:
            tasks[-1].link(sig)
        for sig in maybe_list(self.options.get('link_error')) or []:

            for task in tasks:
                task.link_error(sig)
        return tasks

    def apply_async(self, args=None, kwargs=None, **options):
        # python is best a
…[8629 chars omitted]…
nce(new_chain.tasks[0].body, _chain)

    def test_chain_of_chord_upgrade_on_chaining(self):
        c = chord([signature('header')], group(signature('body')))
        c = chain(c)
        t = signature('t')
        new_chain = c | t  # t should be chained with the body of c[0] and create a new chord
        assert isinstance(new_chain, _chain)
        assert isinstance(new_chain.tasks[0].body, chord)

The responsibility of this method is to ensure that the chain is
        correctly unpacked, and then the correct callbacks are set up along the way.

        Arguments:
            args (Tuple): Partial args to be prepended to the existing args.
            kwargs (Dict): Partial kwargs to be merged with existing kwargs.
            tasks (List[Signature]): The tasks of the chain.
            root_id (str): The id of the root task.
            parent_id (str): The id of the parent task.
            link_error (Union[List[Signature], Signature]): The error callback.
                will be set for all tasks in the chain.
            app (Celery): The Celery app instance.

            last_task_id (str): The id of the last task in the chain.
            group_id (str): The id of the group that the chain is a part of.
            chord_body (Signature): The body of the chord, use
…[8629 chars omitted]…
True
t/unit/tasks/test_canvas.py:1216:        self.app.conf.task_always_eager = True
t/unit/tasks/test_canvas.py:1224:        self.app.conf.task_always_eager = True
t/unit/tasks/test_canvas.py:1230:        self.app.conf.task_always_eager = True
t/unit/tasks/test_canvas.py:1437:        self.app.conf.task_always_eager = True
t/unit/tasks/test_canvas.py:1759:        self.app.conf.task_always_eager = True

prev_task = None
                    prev_res = None

            if is_last_task:
                # chain(task_id=id) means task id is set for the last task
                # in the chain.  If the chord is part of a chord/group
                # then that chord/group must synchronize based on the
                # last task in the chain, so we only set the group_id and
                # chord callback for the last task.
                res = task.freeze(
                    last_task_id,
                    root_id=root_id, group_id=group_id, chord=chord_body,

                    group_index=group_index,
                )
            else:
                res = task.freeze(root_id=root_id)

            i += 1

            if prev_task:
                if use_link:
                    # link previous task to this task.
                    task.link(prev_task)

    
…[8075 chars omitted]…
f.add.s(1), self.add.s(x=1)])
        res = self.helper_test_get_delay(x.delay(y=1))
        assert res == [2, 2]

    def test_apply_from_generator(self):
        child_count = 42
        child_sig = self.add.si(0, 0)
        child_sigs_gen = (child_sig for _ in range(child_count))
        group_sig = group(child_sigs_gen)
        with patch("celery.canvas.Signature.apply_async") as mock_apply_async:

# Use the last task's actual ID, not the input parameter.
        self.id = results[0].id if results else last_task_id
        return tasks, results

    def apply(self, args=None, kwargs=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        last, (fargs, fkwargs) = None, (args, kwargs)
        for task in self.tasks:
            res = task.clone(fargs, fkwargs).apply(
                last and (last.get(),), **dict(self.options, **options))
            res.parent, last, (fargs, fkwargs) = last, res, (None, None)

        return last

    @property
    def app(self):
        app = self._app
        if app is None:
            try:
                app = self.tasks[0]._app
            except LookupError:
                pass
        return app or current_app

    def __repr__(self):
        if not self.tasks:
            return f'<
…[3549 chars omitted]…
p or (
            cls.app if not isinstance(cls.app, property) else current_app
        )
        backend = backend or app.backend
        return backend.restore_group(id)


@Thenable.register
class EagerResult(AsyncResult):
    """Result that we know has already been executed."""

    def __init__(self, id, ret_value, state, traceback=None, name=None):
        # pylint: disable=super-init-not-called

36:.. state:: PROPAGATE_STATES
38:PROPAGATE_STATES
57:    'REVOKED', 'RETRY', 'IGNORED', 'READY_STATES', 'UNREADY_STATES',
58:    'EXCEPTION_STATES', 'PROPAGATE_STATES', 'precedence', 'state',
71:    'REJECTED',
139:REJECTED = 'REJECTED'
142:IGNORED = 'IGNORED'
145:UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})
147:PROPAGATE_STATES = frozenset({FAILURE, REVOKED})
<class 'celery.result.EagerResult'> SUCCESS 2 ['a', 'b']

<class 'celery.exceptions.Ignore'> IGNORED 
<class 'celery.exceptions.Reject'> REJECTED (None, False)
#: Task was started by a worker (:setting:`task_track_started`).
STARTED = 'STARTED'
#: Task succeeded
SUCCESS = 'SUCCESS'
#: Task failed
FAILURE = 'FAILURE'
#: Task was revoked.
REVOKED = 'REVOKED'
#: Task was rejected (only used in events).
REJECTED = 'REJECTED'

#: Task is waiting for retry.
RETRY = 'RETRY'
IGNORED = 'IGNORED'

READY_STATES = frozenset({SUCCESS, FAILURE, REVOKED})
UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})
EXCEPTION_STATES = frozenset({RETRY, FAILURE, REVOKED})
PROPAGATE_STATES = frozenset({FAILURE, REVOKED})

ALL_STATES = frozenset({
    PENDING, RECEIVED, STARTED, SUCCESS, FAILURE, RETRY, REVOKED,
})
1616:    def apply_async(self, args=None, kwargs=None, add_to_parent=True,
1635:        result = self.app.GroupResult(group_id, results
…[1555 chars omitted]…
meError('Recursive result parents')
2120:                if node.parent is None:
2121:                    node.parent = header_result
2123:                node = node.parent
2248:            AsyncResult: The result of the body (with the result of the header in the parent of the body).
2283:            header_result = self.app.GroupResult(*header_result_args)
2285:        bodyres.parent = header_result

<class 'TypeError'> can only concatenate list (not "int") to list
<class 'TypeError'> can only concatenate list (not "int") to list

err Starting chords requires a result backend to be configured.

Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.
err Starting chords requires a result backend to be configured.

Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.

88818cd base
grep: celery/app/__pycache__/trace.cpython-312.pyc: binary file matches
docs/history/changelog-1.0.rst:747:  :setting:`CELERY_STORE_ERRORS_EVEN_IF_IGNORED` to `True`.
docs/history/changelog-3.0.rst:1494:    - :setting:`CElERY_STORE_ERRORS_EVEN_IF_IGNORED`
docs/userguide/tasks.rst:1677:    │                       (skipped for RETRY/REJECTED/IGNORED)    │
docs/userguide/tasks.rst:1688:     It does not run for ``RETRY``, ``REJECTED``, or ``IGNORED``. If you need
docs/userguide/tasks.rst:1774:        It is not executed for ``RETRY``, ``REJECTED``, or ``IGNORED`` states.
celery/app/trace.py:73:LOG_IGNORED = """\
celery/app/trace.py:93:log_policy_ignore = log_policy_t(LOG_IGNORED, 'ignored', logging.INFO, 0, 0)
celery/app/trace.py:109:IGNORED = states.IGNORED
celery/app/trace.py:114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
celery/app/trace.py:521:                        info(LOG_IGNORED, {

p 1 EagerResult [3, 4] <EagerResult: 35b859b1-31f4-4834-9bae-6f295f0eed6a>
(('35b859b1-31f4-4834-9bae-6f295f0eed6a', (('2d70949f-1497-4e56-8b08-592f43435457', (('344c3ed0-773a-40fe-a4fc-bdb34918e902', None), None)), [(('ca339292-749f-458d-a954-aabc046e3553', None), None), (('f4af4069-2acb-40fd-88a5-b86bcd027ff8', None), None)])), None)
rest AsyncResult <AsyncResult: 35b859b1-31f4-4834-9bae-6f295f0eed6a>
p 2 EagerResult [3, 4] <EagerResult: bd195c67-6c35-46c5-aa78-8e0bc048e5fa>
(('bd195c67-6c35-46c5-aa78-8e0bc048e5fa', (('76b9e487-90e3-4e8e-92dd-c8016ad0db12', (('497da2b8-0ae2-45c8-b8ca-06b3923cf1c0', None), None)), [(('be8942ea-80e1-4a02-afe7-7cf22fcd6414', None), None), (('30c99676-35c9-4691-b3f3-b8b68d7a4700', None), None)])), None)
rest AsyncResult <AsyncResult: bd195c67-6c35-46c5-aa78-8e0bc048e5fa>

IGNORED None
REJECTED None
                # seems sensible enough. If that's somehow not the case we'll
                # end up messing up chord counts and there are all sorts of
                # awful race conditions to think about. We'll hope it's not!
                sig, res, group_id = current_task
                chord_obj = chord if chord is not None else sig.options.get("chord")
                # We need to check the chord size of each contributing task so
                # that when we get to the final one, we can correctly set the
                # size in the backend and the chord can be sensible completed.
                chord_size += _chord._descend(sig)
                if chord_obj is not None and next_task is None:

                    # Per above, sanity check that we only saw one group
                    app.backend.set_chord_size(group_id, chord_size)
           
…[7837 chars omitted]…
r('Recursive result parents')
                seen.add(node.id)
                if node.parent is None:
                    node.parent = header_result
                    break
                node = node.parent

        return body_result

    def stamp(self, visitor=None, append_stamps=False, **headers):
        tasks = self.tasks
        if isinstance(tasks, group):
            tasks = tasks.tasks

File "<stdin>", line 6
    def x():pass
IndentationError: unexpected unindent

1 ['_chord', 'Signature'] 2 [('AsyncResult', 'GroupResult'), ('AsyncResult', None)]
(('744400c5-9a90-442b-b98d-a318d6ca0933', (('0feff2e0-012f-47a1-80c8-5be42e2abeca', (('3185e504-9945-4874-aaa1-fb9d2bd340dd', (('e16a1949-ad50-4cd8-b9ed-ae82c7085db6', None), None)), [(('cb0564a8-b94a-466f-9b32-5ad8c2c026d1', None), None), (('c11e788a-d2e3-4a11-b09b-0278173c02ff', None), None)])), [(('a9e04647-ffd6-467c-aa31-e31933468e0a', None), None), (('8b626af1-36f4-446d-b866-05dff9462046', None), None)])), None)
2 ['_chord', 'Signature'] 2 [('AsyncResult', 'GroupResult'), ('AsyncResult', None)]
(('c37acb0f-b495-478c-878c-1aa06822dfa3', (('b59aea76-19ca-4bb1-bf6d-0f51c1e3fb89', (('fd3bec43-a5f6-4625-b909-1287c890ead5', (('48c2f7d8-274b-4547-a696-1c1ecad0eedb', None), None)), [(('94aa075d-96f5-4d52-ac79-54e548994428', None), None), (('5ca5d7af-6df1-4d81-a1bb-923d68d0207f', None), None)])), [(('a217f96a-4b78-4e51-8418-8105479d375a', None), None), (('62b0a0a1-1faa-47f0-b678-fe74ca4a2476', None), None)])), None)

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[1179 chars omitted]…
r
    without exhausting it."""

    task.stamp(visitor, append_stamps, **headers)
    return task


def _merge_dictionaries(d1, d2, aggregate_duplicates=True):
    """Merge two dictionaries recursively into the first one.

    Example:
    >>> d1 = {'dict': {'a': 1}, 'list': [1, 2], 'tuple': (1, 2)}
    >>> d2 = {'dict': {'b': 2}, 'list': [3, 4], 'set': {'a', 'b'}}
    >>> _merge_dictionaries(d1, d2)

t/unit/app/test_exceptions.py:4:from celery.exceptions import Reject, Retry
t/unit/app/test_exceptions.py:21:class test_Reject:
t/unit/app/test_exceptions.py:24:        x = Reject('foo', requeue=True)
t/unit/app/test_exceptions.py:29:        assert repr(Reject('foo', True))
t/unit/tasks/test_trace.py:16:from celery.exceptions import BackendGetMetaError, Ignore, Reject, Retry
t/unit/tasks/test_trace.py:80:        assert get_log_policy(self.add, einfo, Reject()) is log_policy_reject
t/unit/tasks/test_trace.py:293:        assert info.state == states.IGNORED
t/unit/tasks/test_trace.py:297:    def test_when_Reject(self, mock_traceback_clear):
t/unit/tasks/test_trace.py:300:            raise Reject()
t/unit/tasks/test_trace.py:303:        assert info.state == states.REJECTED
t/unit/tasks/test_trace.py:711:                with pytest.raises(Reject):
t/unit/tasks/test_trace.py:910:           

…[1405 chars omitted]…
_Reject_marks_as_failure_when_already_acked(self):
t/unit/worker/test_request.py:385:            raise Reject(requeue=False)
t/unit/worker/test_request.py:386:        except Reject:
t/unit/worker/test_request.py:395:    def test_on_failure_Reject_rejects_with_requeue(self):
t/unit/worker/test_request.py:398:            raise Reject(requeue=True)
t/unit/worker/test_request.py:399:        except Reject:

Traceback (most recent call last):
  File "<stdin>", line 9, in <module>
  File "/repo/celery/canvas.py", line 1044, in apply_async
    return self.apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1300, in apply
    last and (last.get(),), **dict(self.options, **options))
              ^^^^^^^^^^
  File "/repo/celery/result.py", line 735, in get
    return (self.join_native if self.supports_native_join else self.join)(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join

    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: a() takes from 0 to 1 positional arguments but 2 were given

Traceback (most recent call last):
  File "<stdin>", line 9, in <module>
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: a() takes 1 positional argument but 2 were given

Ignore IGNORED None ['a']
Reject REJECTED None ['a']
............u.......................................F
=================================== FAILURES ===================================
________________________ test_chain.test_chord_to_group ________________________

self = <t.unit.tasks.test_canvas.test_chain object at 0x7327cb24faa0>

    def test_chord_to_group(self):
        c = (
            chord([self.add.s('x0', 'y0'), self.add.s('x1', 'y1')],
                  self.add.s(['foo'])) |

            group([self.add.s(['y']), self.add.s(['z'])])
        )
        assert isinstance(c, _chain)
>       assert c.apply().get() == [
               ^^^^^^^^^
            ['x0y0', 'x1y1', 'foo', 'y'],
            ['x0y0', 'x1y1', 'foo', 'z']
        ]

t/unit/tasks/test_canvas.py:591: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = %t.unit.tasks.t
…[894 chars omitted]…

               ^^^^^^^^^
E           AttributeError: 'GroupResult' object has no attribute 'state'. Did you mean: 'save'?

celery/canvas.py:1307: AttributeError
=========================== short test summary info ============================
FAILED t/unit/tasks/test_canvas.py::test_chain::test_chord_to_group - Attribu...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!

............u........................................................... [ 22%]
....................................................x..x................ [ 45%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 59%]
........................................s............................... [ 82%]
........................................................                 [100%]

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..683b7c3 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -24,6 +24,7 @@ from kombu.utils.uuid import uuid
 from vine import barrier
 
 from celery._state import current_app
+from celery import states
 from celery.exceptions import CPendingDeprecationWarning
 from celery.result import GroupResult, allow_join_result
 from celery.utils import abstract

@@ -1299,6 +1300,12 @@ class _chain(Signature):
             res = task.clone(fargs, fkwargs).apply(
                 last and (last.get(),), **dict(self.options, **options))
             res.parent, last, (fargs, fkwargs) = last, res, (None, None)
+            # Ignore and Reject are terminal task states.  In eager mode there
+            # is no worker message chain to stop dispatching the remaining
+            # callbacks, so stop building the chain here and return t
…[503 chars omitted]…
Result 2
GroupResult 2
AsyncResult None
2 (('dcd6dc15-f0b4-4fbc-93fd-24b3969d23dd', (('ed63c8df-91dd-4bca-a060-a27ad994b08f', (('bf316e74-6cf8-4aea-a98e-ceda3b564a91', (('bc200bc5-dbed-4e5c-8888-ed85befa7f03', None), None)), [(('38e5c46d-6054-4fed-90d1-e2fe5b9f0668', None), None), (('916bccbb-3df6-4209-9b37-7eafe71a31b2', None), None)])), [
AsyncResult None
GroupResult 2
GroupResult 2
AsyncResult None

M celery/canvas.py
trace-ok

submitted.

total 332
drwxr-xr-x 1 root root   4096 Sep 30 20:15 .
drwxr-xr-x 1 root root   4096 Sep 30 20:14 ..
-rw-r--r-- 1 root root    343 Sep 30 20:13 .bumpversion.cfg
-rw-r--r-- 1 root root    306 Sep 30 20:13 .dockerignore
-rw-r--r-- 1 root root    227 Sep 30 20:13 .editorconfig
drwxr-xr-x 1 root root   4096 Sep 30 20:14 .git
drwxr-xr-x 1 root root   4096 Sep 30 20:13 .github
-rw-r--r-- 1 root root    437 Sep 30 20:13 .gitignore
-rw-r--r-- 1 root root   1073 Sep 30 20:13 .pre-commit-config.yaml
-rw-r--r-- 1 root root    616 Sep 30 20:13 .readthedocs.yaml
-rw-r--r-- 1 root root  44762 Sep 30 20:13 CONTRIBUTING.rst

-rw-r--r-- 1 root root   8619 Sep 30 20:13 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128887 Sep 30 20:13 Changelog.rst
-rw-r--r-- 1 root root   2717 Sep 30 20:13 LICENSE
-rw-r--r-- 1 root root    709 Sep 30 20:13 MANIFEST.in
-rw-r--r-- 1 root root   5866 Sep 30 20:13 Makefile
-rw-r-
…[6605 chars omitted]…
, result_from_tuple(x == self.app)
./t/unit/tasks/test_result.py:1071:        second_result = result_from_tuple(result.as_tuple(), self.app)
./t/unit/tasks/test_result.py:1075:    def test_GroupResult_as_tuple(self):
./t/unit/tasks/test_result.py:1083:        (result_id, parent_tuple), group_results = result.as_tuple()
./t/unit/tasks/test_result.py:1085:        assert parent_tuple == parent.as_tuple()

t/unit/tasks/test_canvas.py:9:from celery.canvas import (Signature, _chain, _maybe_group, _merge_dictionaries, chain, chord, chunks, group,
t/unit/tasks/test_canvas.py:396:        self.app.conf.task_always_eager = True
t/unit/tasks/test_canvas.py:464:    def test_groups_in_chain_to_chord(self):
t/unit/tasks/test_canvas.py:470:    def test_empty_groups_are_skipped_in_chain(self):
t/unit/tasks/test_canvas.py:482:    def test_empty_group_body_is_skipped_when_chain_upgrades_to_chord(self):
t/unit/tasks/test_canvas.py:493:    def test_generator_backed_empty_group_is_not_skipped_when_chained(self):
t/unit/tasks/test_canvas.py:501:    def test_known_empty_generator_backed_group_is_skipped_in_chain(self):
t/unit/tasks/test_canvas.py:596:    def test_chain_of_chord__or__group_of_single_task(self):
t/unit/tasks/test_canvas.py:639:        ), "Chord followed by a group should be upgraded to a sin
…[8629 chars omitted]…
 add.store_eager_result = True
t/unit/tasks/test_trace.py:1089:                with pytest.raises(Reject):

t/unit/tasks/test_trace.py:1106:        self.trace(add, (2, 2), {}, request=request, eager=False)
t/unit/tasks/test_trace.py:1119:        self.trace(add, (2, 2), {}, request=request, eager=False)
t/unit/tasks/test_trace.py:1138:            self.trace(add, (2, 2), {}, request=request, eager=False)

else:
            # Also supports using task class/instance instead of string name.
            try:
                task_name = task.name
            except AttributeError:
                task_name = task
            else:
                self._type = task

            super().__init__(
                task=task_name, args=tuple(args or ()),
                kwargs=kwargs or {},

                options=dict(options or {}, **ex),
                subtask_type=subtask_type,
                immutable=immutable,
            )

    def __call__(self, *partial_args, **partial_kwargs):
        """Call the task directly (in the current process)."""
        args, kwargs, _ = self._merge(partial_args, partial_kwargs, None)
        return self.type(*args, **kwargs)

    def delay(self, *partial_args, **partial_kwargs):
        """Shortcut to :meth:`apply_async` using star arguments."""
 
…[8629 chars omitted]…
y/app/trace.py:588:                    I, R = Info(REJECTED, exc), ExceptionInfo(internal=True)
celery/events/state.py:75:    'rejected': states.REJECTED,
celery/states.py:71:    'REJECTED',
celery/states.py:139:REJECTED = 'REJECTED'
celery/states.py:145:UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})
t/unit/tasks/test_trace.py:303:        assert info.state == states.REJECTED

The tasks list would be cloned from the chain's tasks.
        All of the chain callbacks would be added to the last task in the (cloned) chain.
        All of the tasks would be linked to the same error callback
        as the chain itself, to ensure that the correct error callback is called
        if any of the (cloned) tasks of the chain fail.
        """
        # Clone chain's tasks assigning signatures from link_error
        # to each task and adding the chain's links to the last task.
        tasks = [t.clone() for t in self.tasks]
        for sig in maybe_list(self.options.get('link')) or []:
            tasks[-1].link(sig)
        for sig in maybe_list(self.options.get('link_error')) or []:

            for task in tasks:
                task.link_error(sig)
        return tasks

    def apply_async(self, args=None, kwargs=None, **options):
        # python is best a
…[8629 chars omitted]…
    else:
#: Task is waiting for retry.
RETRY = 'RETRY'
IGNORED = 'IGNORED'

READY_STATES = frozenset({SUCCESS, FAILURE, REVOKED})
UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})
EXCEPTION_STATES = frozenset({RETRY, FAILURE, REVOKED})
PROPAGATE_STATES = frozenset({FAILURE, REVOKED})

ALL_STATES = frozenset({
    PENDING, RECEIVED, STARTED, SUCCESS, FAILURE, RETRY, REVOKED,
})

869:    def apply(self, args=None, kwargs=None,
            kwargs (Dict): Keyword arguments to retry with.
            exc (Exception): Custom exception to report when the max retry
                limit has been exceeded (default:
                :exc:`~@MaxRetriesExceededError`).

                If this argument is set and retry is called while
                an exception was raised (``sys.exc_info()`` is set)
                it will attempt to re-raise the current exception.

                If no exception was raised it will raise the ``exc``
                argument provided.

            countdown (float): Time in seconds to delay the retry for.
            eta (~datetime.datetime): Explicit time and date to run the
                retry at.
            max_retries (int): If set, overrides the default retry limit for
                this execution.  Changes to this parameter d
…[8629 chars omitted]…
s)


class _basemap(Signature):
    _task_name = None
    _unpack_args = itemgetter('task', 'it')

    @classmethod
    def from_dict(cls, d, app=None):
        return cls(*cls._unpack_args(d['kwargs']), app=app, **d['options'])

    def __init__(self, task, it, **options):
        super().__init__(self._task_name, (),
                         {'task': task, 'it': regen(it)}, immutable=True, **options

def apply(self, args=None, kwargs=None,
              link=None, link_error=None,
              task_id=None, retries=None, throw=None,
              logfile=None, loglevel=None, headers=None, **options):
        """Execute this task locally, by blocking until the task returns.

        Arguments:
            args (Tuple): positional arguments passed on to the task.
            kwargs (Dict): keyword arguments passed on to the task.
            throw (bool): Re-raise task exceptions.
                Defaults to the :setting:`task_eager_propagates` setting.

        Returns:
            celery.result.EagerResult: pre-evaluated result.
        """
        # trace imports Task, so need to import inline.
        from celery.app.trace import build_tracer

        app = self._get_app()
        args = args or ()
        kwargs = kwargs or {}
        task_id = task_id or uuid()
        re
…[8432 chars omitted]…
get()
            4

        Using a generator expression:

        .. code-block:: pycon

            >>> lazy_chain = chain(add.s(i) for i in range(10))
            >>> res = lazy_chain(3)

    Arguments:
        *tasks (Signature): List of task signatures to chain.
            If only one argument is passed and that argument is
            an iterable, then that'll be used as the list of signatures

child_sig = self.add.si(1, 1)
        chain_sig = chain(child_sig)
        assert chain_sig.tasks[0] is child_sig

        with patch.object(
            # We want to get back the result of actually applying the task
            child_sig, "apply_async",
        ) as mock_apply, patch.object(
            # The child signature may be clone by `chain.prepare_steps()`
            child_sig, "clone", return_value=child_sig,
        ):
            res = chain_sig()

        # `_prepare_chain_from_options()` sets this `chain` kwarg with the
        # subsequent tasks which would be run - nothing in this case
        mock_apply.assert_called_once_with(chain=[])
        assert res is mock_apply.return_value

    @pytest.mark.usefixtures('depends_on_current_app')
    def test_chain_single_child_group_result(self):
        child_sig = self.add.si(1, 1)
        # The group will `clone()` 
…[8629 chars omitted]…
                 "task": "add",
                            "args": [
                                1,
                                2
                            ],
                            "kwargs": {},
                            "options": {},
                            "subtask_type": None,
                            "immutable": False
                        },
                        {

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[1276 chars omitted]…
k


def _merge_dictionaries(d1, d2, aggregate_duplicates=True):
    """Merge two dictionaries recursively into the first one.

    Example:
    >>> d1 = {'dict': {'a': 1}, 'list': [1, 2], 'tuple': (1, 2)}
    >>> d2 = {'dict': {'b': 2}, 'list': [3, 4], 'set': {'a', 'b'}}
    >>> _merge_dictionaries(d1, d2)
548c3dd0-04b2-48b0-be7b-27dead275368 FAILURE side() takes 0 positional arguments but 1 was given

<class 'celery.exceptions.Ignore'> IGNORED Ignore() None
<class 'celery.exceptions.Reject'> REJECTED reject requeue=False: None None
"""Built-in task states.

.. _states:

States
------

See :ref:`task-states`.

.. _statesets:

Sets
----

.. state:: READY_STATES

READY_STATES
~~~~~~~~~~~~

Set of states meaning the task result is ready (has been executed).

.. state:: UNREADY_STATES

UNREADY_STATES
~~~~~~~~~~~~~~

Set of states meaning the task result is not ready (hasn't been executed).

.. state:: EXCEPTION_STATES

EXCEPTION_STATES
~~~~~~~~~~~~~~~~

Set of states meaning the task returned an exception.

426:    def _dispatch_callbacks_and_chain(

Traceback (most recent call last):
  File "<stdin>", line 7, in <module>
  File "/repo/celery/canvas.py", line 1044, in apply_async
    return self.apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1299, in apply
    res = task.clone(fargs, fkwargs).apply(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 2187, in apply
    args=(tasks.apply(args, kwargs).get(propagate=propagate),),
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 735, in get

    return (self.join_native if self.supports_native_join else self.join)(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
  File "<stdin>", line 5, in a
TypeError: can only concatenate list (not "int") to list

proto 1 <class 'celery.result.AsyncResult'> 1b557253-7ba5-425b-8ad8-27981979aa9f
0 AsyncResult 1b557253-7ba5-425b-8ad8-27981979aa9f parent GroupResult children 0
1 GroupResult 7515e5cd-1266-44c9-8165-571ca49befbb parent GroupResult children 2
2 GroupResult e94ce3e4-5970-411b-abfd-274a3013e56a parent AsyncResult children 2
3 AsyncResult 6a6de70b-87be-4e43-9c7b-7cae2fcca745 parent None children 0
proto 2 <class 'celery.result.AsyncResult'> 1b557253-7ba5-425b-8ad8-27981979aa9f
0 AsyncResult 1b557253-7ba5-425b-8ad8-27981979aa9f parent GroupResult children 0
1 GroupResult 7515e5cd-1266-44c9-8165-571ca49befbb parent GroupResult children 2
2 GroupResult dabba1a9-3b53-43fc-abb8-700dffd58655 parent AsyncResult children 2
3 AsyncResult 6a6de70b-87be-4e43-9c7b-7cae2fcca745 parent None children 0

Traceback (most recent call last):
  File "<stdin>", line 6, in <module>
  File "/repo/celery/canvas.py", line 1100, in freeze
    _, results = self._frozen = self.prepare_steps(
                                ^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1279, in prepare_steps
    app.backend.ensure_chords_allowed()
  File "/repo/celery/backends/base.py", line 1272, in ensure_chords_allowed
    raise NotImplementedError(E_CHORD_NO_BACKEND.strip())
NotImplementedError: Starting chords requires a result backend to be configured.

Note that a group chained with a task is also upgraded to be a chord,

as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.

x = self.app.AsyncResult(uuid())
        x.parent = self.app.AsyncResult(uuid())
        y = result_from_tuple(x.as_tuple(), self.app)
        assert y == x
        assert y.parent == x.parent
        assert isinstance(y.parent, AsyncResult)

    def test_compat(self):
        uid = uuid()
        x = result_from_tuple([uid, []], app=self.app)
        assert x.id == uid

    def test_as_list(self):
        uid = uuid()
        x = self.app.AsyncResult(uid)
        assert x.id == x.as_list()[0]
        assert isinstance(x.as_list(), list)

    def test_GroupResult(self):
        x = self.app.GroupResult(
            uuid(), [self.app.AsyncResult(uuid()) for _ in range(10)],
        )
        assert x, result_from_tuple(x.as_tuple() == self.app)
        assert x, result_from_tuple(x == self.app)

    def test_GroupResult_with_parent(self):
        parent = self.app.AsyncResult(u
…[1331 chars omitted]…
      tasks2, _ = c2.prepare_steps((), {}, c2.tasks)
838:            # The child signature may be clone by `chain.prepare_steps()`
861:            # The child signature may be clone by `chain.prepare_steps()`
916:        prepare_steps set self.id = last_task_id (the input parameter,
977:        # Simulate what apply_async does: pass link_error to prepare_steps
979:        c.prepare_steps(
c87ec12 base

AsyncResult e008d953-47ed-47a9-b583-bf2d1cc5a3a7 p GroupResult
  GroupResult c8b6f5d9-a3a1-4a5f-a1c2-fcdc53a2e8ce p GroupResult
   child c2dcac32-fcee-4df5-9444-5ea4fcf9be28 p -
   child 152b5d16-6dfb-4ec1-80b3-c3dca85bb319 p -
   GroupResult 8a547be0-1f3d-4bc4-81b1-2d610eb9c966 p AsyncResult
    child 1bfca64d-96d2-444d-9a46-9099ba085176 p -
    child 9e298610-f842-46ac-9792-f6aaf9ea6d14 p -
    AsyncResult 2dc2fd0b-9928-4f65-8df1-8ebdfa32c89e p -
REST
 AsyncResult e008d953-47ed-47a9-b583-bf2d1cc5a3a7 p GroupResult
  GroupResult c8b6f5d9-a3a1-4a5f-a1c2-fcdc53a2e8ce p GroupResult
   child c2dcac32-fcee-4df5-9444-5ea4fcf9be28 p -

   child 152b5d16-6dfb-4ec1-80b3-c3dca85bb319 p -
   GroupResult 8a547be0-1f3d-4bc4-81b1-2d610eb9c966 p AsyncResult
    child 1bfca64d-96d2-444d-9a46-9099ba085176 p -
    child 9e298610-f842-46ac-9792-f6aaf9ea6d14 p -
    AsyncResult 2dc2fd0b-9928-4f65-8df1-8ebdfa32c89e p -

def graph(self):
        return self.build_graph()

    @property
    def supports_native_join(self):
        return self.backend.supports_native_join

    @property
    def children(self):
        return self._get_task_meta().get('children')

    def _maybe_set_cache(self, meta):

        if meta:
            state = meta['status']
            if state in states.READY_STATES:
                d = self._set_cache(self.backend.meta_from_decoded(meta))
                self.on_ready(self)
                return d
        return meta

    def _get_task_meta(self):
        if self._cache is None:
            return self._maybe_set_cache(self.backend.get_task_meta(self.id))
        return self._cache

    def _iter_meta(self, **kwargs):
        return iter([self._get_task_meta()])

    def _set_cache(self, d):
        children = d.get('children')
        if children:
            d['children'] = [
                result_from_tuple(child, self.app) for child in children
            ]
        self._cache = d
        return d

    @property
    def result(self):
        """Task return value.

def test_known_empty_generator_backed_group_is_skipped_in_chain(self):
        def tasks():
            yield from ()

        c = _chain(
            group([self.add.s(2, 2)], app=self.app),
            group(tasks(), app=self.app),
            app=self.app,
        )

        prepared_tasks, results = c.prepare_steps((), {}, c.tasks)

        assert len(prepared_tasks) == 1
        assert prepared_tasks[0].task == self.add.name
        assert isinstance(results[0], AsyncResult)

    def test_prepare_steps_set_last_task_id_to_chain(self):
        last_task = self.add.s(2).set(task_id='42')
        c = self.add.s(4) | last_task
        assert c.id is None
        tasks, _ = c.prepare_steps((), {}, c.tasks, last_task_id=last_task.id)
        assert c.id == last_task.id

    def test_group_to_chord(self):
        c = (
            self.add.s(5) |
            group([self.add.s(i, i)
…[5466 chars omitted]…
          visitor_headers = visitor.on_group_start(self, **headers) or {}
        headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
        self.stamp_links(visitor, append_stamps, **headers)

        if isinstance(self.tasks, _regen):
1714:    def _prepared(self, tasks, partial_args, group_id, root_id, app,
1763:    def _apply_tasks(self, tasks, producer=None, app=None, p=None,

1 EagerResult
[('EagerResult', 0), ('GroupResult', 2), ('EagerResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('AsyncResult', 0)]
2 EagerResult
[('EagerResult', 0), ('GroupResult', 2), ('EagerResult', 0)] [('AsyncResult', 0), ('GroupResult', 2), ('AsyncResult', 0)]

356:    def apply(self, args=None, kwargs=None, **options):
1294:    def apply(self, args=None, kwargs=None, **options):
1652:    def apply(self, args=None, kwargs=None, **options):
2179:    def apply(self, args=None, kwargs=None,
        else:
            stamps = headers.copy()
            for task in tasks:
                task.stamp(visitor, append_stamps, **stamps)

        if visitor is not None:
            visitor.on_chord_header_end(self, **headers)

        if visitor is not None and self.body is not None:
            visitor_headers = visitor.on_chord_body(self, **headers) or {}
            headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
            self.body.stamp(visitor, append_stamps, **headers)

    def apply_async(self, args=None, kwargs=None, task_id=None,
                    producer=None, publisher=None, connection=None,
                   
…[2905 chars omitted]…
imple signature counts as 1 completion for this chord
            return 1
        # Any other types are assumed to be iterables of simple signatures
        return len(sig_obj)

    def __length_hint__(self):
        """Return the number of tasks in this chord's header (recursively)."""
        tasks = getattr(self.tasks, "tasks", self.tasks)
        return sum(self._descend(task) for task in tasks)

139:REJECTED = 'REJECTED'
142:IGNORED = 'IGNORED'
            # there is no worker to consume the chain embedded in the
            # request, so applying it here must not execute its successors.
>           if res.state in (states.IGNORED, states.REJECTED):
               ^^^^^^^^^
E           AttributeError: 'GroupResult' object has no attribute 'state'. Did you mean: 'save'?

celery/canvas.py:1306: AttributeError
=========================== short test summary info ============================
FAILED t/unit/tasks/test_canvas.py::test_chain::test_chord_to_group - Attribu...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!

............u........................................................... [ 22%]
....................................................x..x................ [ 45%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 59%]
........................................s............................... [ 82%]
........................................................                 [100%]
312 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 6.28s

Ignore IGNORED None []
Reject REJECTED None []
diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..6614845 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -26,6 +26,7 @@ from vine import barrier
 from celery._state import current_app
 from celery.exceptions import CPendingDeprecationWarning
 from celery.result import GroupResult, allow_join_result
+from celery import states
 from celery.utils import abstract

 from celery.utils.collections import ChainMap
 from celery.utils.functional import _regen
@@ -1299,6 +1300,11 @@ class _chain(Signature):
             res = task.clone(fargs, fkwargs).apply(
                 last and (last.get(),), **dict(self.options, **options))
             res.parent, last, (fargs, fkwargs) = last, res, (None, None)
+            # Ignore and Reject are terminal task outcomes.  In eager mode
+            # there is no worker to consume the chain embedded in the
+            # request, so applying it here must not execute its successors.
+            if getattr(res, 'state', None) in (states.IGNORED, states.REJECTED):
+                break
         return last
 
     @property

def link_error(self, sig):
        # Any child task might error so we need to ensure that they are all
        # capable of calling the linked error signature. This opens the
        # possibility that the task is called more than once but that's better
        # than it not being called at all.
        #
        # We return a concretised tuple of the signatures actually applied to
        # each child task signature, of which there might be none!
        sig = maybe_signature(sig)

        return tuple(child_task.link_error(sig.clone()) for child_task in self.tasks)

    def _prepared(self, tasks, partial_args, group_id, root_id, app,
                  CallableSignature=abstract.CallableSignature,
                  from_dict=Signature.from_dict,
                  isinstance=isinstance, tuple=tuple):
        """Recursively unroll the group into a generator of its tasks.

        T
…[3018 chars omitted]…
   generator: A generator for the AsyncResult of the tasks in the group.
        """
        # pylint: disable=redefined-outer-name
        #   XXX chord is also a class in outer scope.
        app = app or self.app
        with app.producer_or_acquire(producer) as producer:
            # Iterate through tasks two at a time. If tasks is a generator,
478:    def _set_cache(self, d):
 M celery/canvas.py

celery/backends/asynchronous.py:15:from celery import states
celery/backends/cassandra.py:4:from celery import states
celery/backends/database/models.py:7:from celery import states
celery/backends/database/__init__.py:5:from celery import states
celery/backends/elasticsearch.py:7:from celery import states
celery/backends/mongodb.py:8:from celery import states
celery/backends/redis.py:13:from celery import states
celery/backends/rpc.py:13:from celery import states
celery/canvas.py:29:from celery import states
celery/contrib/testing/manager.py:12:from celery import states
Traceback (most recent call last):
  File "<stdin>", line 7, in <module>

AttributeError: 'EagerResult' object has no attribute 'results'. Did you mean: 'result'?

grep: celery/app/__pycache__/base.cpython-312.pyc: binary file matches
grep: celery/events/__pycache__/event.cpython-312.pyc: binary file matches
grep: t/unit/tasks/__pycache__/test_canvas.cpython-312-pytest-9.1.1.pyc: binary file matches
docs/faq.rst:556:    * The `header` and `fanout` exchange types
docs/faq.rst:557:        (`fanout` is supported by Redis).
docs/history/changelog-2.2.rst:482:    fanout exchanges so it's able to perform worker remote control
docs/history/changelog-3.1.rst:929:        BROKER_TRANSPORT_OPTIONS = {'fanout_patterns': True}
docs/history/changelog-4.4.rst:20:- Remove redis fanout caveats.
docs/history/whatsnew-3.1.rst:585:    BROKER_TRANSPORT_OPTIONS = {'fanout_prefix': True}
docs/history/whatsnew-4.0.rst:614:The Redis ``fanout_patterns`` and ``fanout_prefix`` transport
docs/history/whatsnew-4.0.rst:628:        'fanout_patterns': True,
docs/history/whatsne

…[1418 chars omitted]…
(json.dumps(original_group))
t/unit/tasks/test_canvas.py:1147:        deserialized_group = group_type.from_dict(serialized_group)
t/unit/tasks/test_canvas.py:1148:        assert isinstance(deserialized_group, group_type)
t/unit/tasks/test_canvas.py:1151:            for child_task in deserialized_group.tasks
t/unit/tasks/test_canvas.py:1158:        serialized_group = json.loads(json.dumps(outer_group))

EagerResult a09d797d-682c-48ba-971b-54fb623dec83 0
GroupResult 6eaacb52-505c-4438-905d-3c61be27e218 2
EagerResult 13039a0f-a84a-4cb6-826a-0131a1155036 0
rest
AsyncResult a09d797d-682c-48ba-971b-54fb623dec83 0
GroupResult 6eaacb52-505c-4438-905d-3c61be27e218 2
AsyncResult 13039a0f-a84a-4cb6-826a-0131a1155036 0
(('a09d797d-682c-48ba-971b-54fb623dec83', (('6eaacb52-505c-4438-905d-3c61be27e218', (('13039a0f-a84a-4cb6-826a-0131a1155036', None), None)), [(('5d12c7d8-e8f2-4fa1-8738-353ead59f407', None), None), (('8b5cb76b-6d3b-41b4-a774-74daf590ce90', None), None)])), None)

........................................................................ [ 37%]
........................................................................ [ 74%]
.................................................                        [100%]
193 passed, 2 warnings in 3.19s
 celery/canvas.py | 6 ++++++
 1 file changed, 6 insertions(+)

submitted.

total 332
drwxr-xr-x 1 root root   4096 Sep 30 20:17 .
drwxr-xr-x 1 root root   4096 Sep 30 20:16 ..
-rw-r--r-- 1 root root    343 Sep 30 20:15 .bumpversion.cfg
-rw-r--r-- 1 root root    306 Sep 30 20:15 .dockerignore
-rw-r--r-- 1 root root    227 Sep 30 20:15 .editorconfig
drwxr-xr-x 1 root root   4096 Sep 30 20:16 .git
drwxr-xr-x 1 root root   4096 Sep 30 20:15 .github
-rw-r--r-- 1 root root    437 Sep 30 20:15 .gitignore
-rw-r--r-- 1 root root   1073 Sep 30 20:15 .pre-commit-config.yaml
-rw-r--r-- 1 root root    616 Sep 30 20:15 .readthedocs.yaml
-rw-r--r-- 1 root root  44762 Sep 30 20:15 CONTRIBUTING.rst

-rw-r--r-- 1 root root   8619 Sep 30 20:15 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128887 Sep 30 20:15 Changelog.rst
-rw-r--r-- 1 root root   2717 Sep 30 20:15 LICENSE
-rw-r--r-- 1 root root    709 Sep 30 20:15 MANIFEST.in
-rw-r--r-- 1 root root   5866 Sep 30 20:15 Makefile
-rw-r-
…[8630 chars omitted]…
(self):
./t/unit/app/test_builtins.py:175:        self.app.conf.task_always_eager = True
./t/unit/app/test_builtins.py:180:    def test_apply_eager_with_arguments(self):
./t/unit/app/test_builtins.py:181:        self.app.conf.task_always_eager = True
./t/unit/app/test_utils.py:18:        assert self.app.conf.find_option('always_eager')
./t/unit/app/test_utils.py:27:            'always_eager') is False

616cc17 base
"""Celery error types.

Error Hierarchy
===============

- :exc:`Exception`
    - :exc:`celery.exceptions.CeleryError`
        - :exc:`~celery.exceptions.ImproperlyConfigured`
        - :exc:`~celery.exceptions.SecurityError`
        - :exc:`~celery.exceptions.TaskPredicate`
            - :exc:`~celery.exceptions.Ignore`

            - :exc:`~celery.exceptions.Reject`
            - :exc:`~celery.exceptions.Retry`
        - :exc:`~celery.exceptions.TaskError`
            - :exc:`~celery.exceptions.QueueNotFound`
            - :exc:`~celery.exceptions.IncompleteStream`
            - :exc:`~celery.exceptions.NotRegistered`
            - :exc:`~celery.exceptions.AlreadyRegistered`
            - :exc:`~celery.exceptions.TimeoutError`
            - :exc:`~celery.exceptions.MaxRetriesExceededError`
            - :exc:`~celery.exceptions.TaskRevokedError`
            - :exc:`~cele
…[4462 chars omitted]…
f.is_eager = is_eager
        self.sig = sig
        super().__init__(self, exc, when, **kwargs)

    def humanize(self):
        if isinstance(self.when, numbers.Number):
            return f'in {self.when}s'
        return f'at {self.when}'

    def __str__(self):
        if self.message:
            return self.message
        if self.excs:
            return f'Retry {self.humanize()}: {self.excs}'

>>> @Signature.register_type()
        >>> class mysig(Signature):
        >>>     pass
        """
        def _inner(subclass):
            cls.TYPES[name or subclass.__name__] = subclass
            return subclass

        return _inner

    @classmethod
    def from_dict(cls, d, app=None):

        """Create a new signature from a dict.
        Subclasses can override this method to customize how are
        they created from a dict.
        """
        typ = d.get('subtask_type')
        if typ:
            target_cls = cls.TYPES[typ]
            if target_cls is not cls:
                return target_cls.from_dict(d, app=app)
        return Signature(d, app=app)

    def __init__(self, task=None, args=None, kwargs=None, options=None,
                 type=None, subtask_type=None, immutable=False,
                 app=None, **ex):
        self._app = app

        if isins
…[8630 chars omitted]…
   (result_id, parent_tuple), group_results = result.as_tuple()
        assert result_id == result.id
        assert parent_tuple == parent.as_tuple()
        assert parent_tuple[0][0] == parent.id
        assert isinstance(group_results, list)
        expected_grp_res = [((f'async-result-{i}', None), None)
                            for i in range(2)]
        assert group_results == expected_grp_res

class _chain(Signature):
    tasks = getitem_property('kwargs.tasks', 'Tasks in chain.')

    @classmethod
    def from_dict(cls, d, app=None):
        tasks = d['kwargs']['tasks']
        if tasks:
            if isinstance(tasks, tuple):  # aaaargh
                tasks = d['kwargs']['tasks'] = list(tasks)
            tasks = [maybe_signature(task, app=app) for task in tasks]
        return cls(tasks, app=app, **d['options'])

    def __init__(self, *tasks, **options):
        tasks = (regen(tasks[0]) if len(tasks) == 1 and is_list(tasks[0])
                 else tasks)
        super().__init__('celery.chain', (), {'tasks': tasks}, **options
                         )
        self._use_link = options.pop('use_link', None)
        self.subtask_type = 'chain'
        self._frozen = None

    def __call__(self, *args, **kwargs):
        if self.tasks:
            return self.apply_asyn
…[8629 chars omitted]…
or is not None:
            visitor_headers = visitor.on_group_start(self, **headers) or {}
        headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
        self.stamp_links(visitor, append_stamps, **headers)

        if isinstance(self.tasks, _regen):
123:    def as_tuple(self):
930:class GroupResult(ResultSet):
1005:    def as_tuple(self):
1026:class EagerResult(AsyncResult):

The tasks list would be cloned from the chain's tasks.
        All of the chain callbacks would be added to the last task in the (cloned) chain.
        All of the tasks would be linked to the same error callback
        as the chain itself, to ensure that the correct error callback is called
        if any of the (cloned) tasks of the chain fail.
        """
        # Clone chain's tasks assigning signatures from link_error
        # to each task and adding the chain's links to the last task.
        tasks = [t.clone() for t in self.tasks]
        for sig in maybe_list(self.options.get('link')) or []:
            tasks[-1].link(sig)
        for sig in maybe_list(self.options.get('link_error')) or []:

            for task in tasks:
                task.link_error(sig)
        return tasks

    def apply_async(self, args=None, kwargs=None, **options):
        # python is best a
…[8629 chars omitted]…
r)

    def _get_task_meta(self):
        return self._cache

t/unit/tasks/test_trace.py:80:        assert get_log_policy(self.add, einfo, Reject()) is log_policy_reject
t/unit/tasks/test_trace.py:297:    def test_when_Reject(self, mock_traceback_clear):
t/unit/tasks/test_trace.py:300:            raise Reject()
t/unit/tasks/test_trace.py:1049:        is caught and re-raised as Reject(requeue=True)."""

For example:
        chain(signature1, chain(signature2, signature3)) --> Upgrades to chain(signature1, signature2, signature3)
        chain(group(signature1, signature2), signature3) --> Upgrades to chord([signature1, signature2], signature3)

        The responsibility of this method is to ensure that the chain is
        correctly unpacked, and then the correct callbacks are set up along the way.

        Arguments:
            args (Tuple): Partial args to be prepended to the existing args.
            kwargs (Dict): Partial kwargs to be merged with existing kwargs.
            tasks (List[Signature]): The tasks of the chain.
            root_id (str): The id of the root task.

            parent_id (str): The id of the parent task.
            link_error (Union[List[Signature], Signature]): The error callback.
                will be set for all tasks in the chain.
     
…[8382 chars omitted]…
  ``wait`` is enabled.
        """
        self.app.control.revoke_by_stamped_headers(headers, connection=connection,
                                                   terminate=terminate, signal=signal,
                                                   reply=wait, timeout=timeout)

t/unit/tasks/test_trace.py:290:            raise Ignore()
t/unit/worker/test_request.py:304:            raise Ignore()

# Use the last task's actual ID, not the input parameter.
        self.id = results[0].id if results else last_task_id
        return tasks, results

    def apply(self, args=None, kwargs=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        last, (fargs, fkwargs) = None, (args, kwargs)
        for task in self.tasks:
            res = task.clone(fargs, fkwargs).apply(
                last and (last.get(),), **dict(self.options, **options))
            res.parent, last, (fargs, fkwargs) = last, res, (None, None)

        return last

    @property
    def app(self):
        app = self._app
        if app is None:
            try:
                app = self.tasks[0]._app
            except LookupError:
                pass
        return app or current_app

    def __repr__(self):
        if not self.tasks:
            return f'<
…[4498 chars omitted]…
rguments.
728-            options (Dict): Task execution options.
729-        """
730-
731-    def signature_from_request(self, request=None, args=None, kwargs=None,
732-                               queue=None, **extra_options):
733-        request = self.request if request is None else request
734-        args = request.args if args is None else args
grep: celery/trace.py: No such file or directory

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[4448 chars omitted]…
      return qualname(args[0])

                @app.task(shadow_name=shadow_name, serializer='pickle')
                def apply_function_async(fun, *args, **kwargs):
                    return fun(*args, **kwargs)

        Arguments:
            args (Tuple): Task positional arguments.
            kwargs (Dict): Task keyword arguments.
            options (Dict): Task execution options.
        """

celery/app/trace.py:73:LOG_IGNORED = """\
celery/app/trace.py:78:LOG_REJECTED = """\
celery/app/trace.py:92:log_policy_reject = log_policy_t(LOG_REJECTED, 'rejected', logging.WARN, 1, 1)
celery/app/trace.py:93:log_policy_ignore = log_policy_t(LOG_IGNORED, 'ignored', logging.INFO, 0, 0)
celery/app/trace.py:109:IGNORED = states.IGNORED
celery/app/trace.py:110:REJECTED = states.REJECTED
celery/app/trace.py:114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
celery/app/trace.py:521:                        info(LOG_IGNORED, {
celery/app/trace.py:588:                    I, R = Info(REJECTED, exc), ExceptionInfo(internal=True)
celery/app/trace.py:594:                    I, R = Info(IGNORED, exc), ExceptionInfo(internal=True)
celery/events/state.py:75:    'rejected': states.REJECTED,
celery/states.py:57:    'REVOKED', 'RETRY', 'IGNORED', 'READY_STATES', 'UNREADY_STATES',

celery/states.p
…[4092 chars omitted]…
ckend=self.backend,
960-                                           task_name=self.name, **kwargs)
961-
962-    def signature(self, args=None, *starargs, **starkwargs):
963-        """Create signature.
964-
965-        Returns:
966-            :class:`~celery.signature`:  object for
967-                this task, wrapping arguments and execution options
968-                for a single task invocation.

083cc08f-e4c6-4098-97d8-fde67e5d1419 SUCCESS 2 2 ['a', 'b'] <class 'celery.result.EagerResult'>
Traceback (most recent call last):
  File "<stdin>", line 5, in <module>
  File "/repo/celery/canvas.py", line 1044, in apply_async
    return self.apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1299, in apply
    res = task.clone(fargs, fkwargs).apply(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 2187, in apply
    args=(tasks.apply(args, kwargs).get(propagate=propagate),),
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/repo/celery/result.py", line 735, in get
    return (self.join_native if self.supports_native_join else self.join)(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
  File "<stdin>", line 4, in t
TypeError: can only concatenate list (not "int") to list

#: Task was started by a worker (:setting:`task_track_started`).
STARTED = 'STARTED'
#: Task succeeded
SUCCESS = 'SUCCESS'
#: Task failed
FAILURE = 'FAILURE'
#: Task was revoked.
REVOKED = 'REVOKED'
#: Task was rejected (only used in events).
REJECTED = 'REJECTED'
#: Task is waiting for retry.
RETRY = 'RETRY'

IGNORED = 'IGNORED'

READY_STATES = frozenset({SUCCESS, FAILURE, REVOKED})
UNREADY_STATES = frozenset({PENDING, RECEIVED, STARTED, REJECTED, RETRY})
EXCEPTION_STATES = frozenset({RETRY, FAILURE, REVOKED})
PROPAGATE_STATES = frozenset({FAILURE, REVOKED})

ALL_STATES = frozenset({
    PENDING, RECEIVED, STARTED, SUCCESS, FAILURE, RETRY, REVOKED,
})
        body = kwargs.pop('body', None) or self.kwargs['body']
        kwargs = dict(self.kwargs['kwargs'], **kwargs)
        body = body.clone(**options)
        app = self._get_app(body)
        tasks = (self.tasks.clone() if isinstanc
…[2688 chars omitted]…
  File "/repo/celery/backends/base.py", line 1272, in ensure_chords_allowed
    raise NotImplementedError(E_CHORD_NO_BACKEND.strip())
NotImplementedError: Starting chords requires a result backend to be configured.

Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.

t/unit/tasks/test_canvas.py:9:from celery.canvas import (Signature, _chain, _maybe_group, _merge_dictionaries, chain, chord, chunks, group,
t/unit/tasks/test_canvas.py:76:        self.replaced_group = replaced_group
t/unit/tasks/test_canvas.py:82:        self.replace_with_group = replace_with_group
t/unit/tasks/test_canvas.py:105:class group_subclass(group):
t/unit/tasks/test_canvas.py:330:    @pytest.mark.parametrize('_id, group_id, chord, root_id, parent_id, group_index', [
t/unit/tasks/test_canvas.py:333:    def test_freezing_args_set_in_options(self, _id, group_id, chord, root_id, parent_id, group_index):
t/unit/tasks/test_canvas.py:337:            group_id=group_id,
t/unit/tasks/test_canvas.py:341:            group_index=group_index,
t/unit/tasks/test_canvas.py:346:        assert options['group_id'] == group_id
t/unit/tasks/test_canvas.py:350:        assert options['group_index']
…[8629 chars omitted]…
run(self, header, body, partial_args, app=None, interval=None,

2227-            countdown=1, max_retries=None, eager=False,
2228-            task_id=None, kwargs=None, **options):
2229-        """Execute the chord.
2230-
2231-        Executing the chord means executing the header and sending the
2232-        result to the body. In case of an empty header, the body is
2233-        executed immediately.

@pytest.mark.usefixtures('depends_on_current_app')
    def test_app_falls_back_to_default(self):
        from celery._state import current_app
        assert chain().app is current_app

    def test_handles_dicts(self):
        c = chain(
            self.add.s(5, 5), dict(self.add.s(8)), app=self.app,
        )
        c.freeze()
        tasks, _ = c._frozen
        assert all(isinstance(task, Signature) for task in tasks)

        assert all(task.app is self.app for task in tasks)

    def test_groups_in_chain_to_chord(self):
        g1 = group([self.add.s(2, 2), self.add.s(4, 4)])
        g2 = group([self.add.s(3, 3), self.add.s(5, 5)])
        c = g1 | g2
        assert isinstance(c, chord)

    def test_empty_groups_are_skipped_in_chain(self):
        c = chain(
            group([self.add.s(2, 2), self.add.s(4, 4)], app=self.app),
            group(app=self.app),
            
…[4198 chars omitted]…
s not None:
            results.extend(parent.as_list())
        return results

    def forget(self):
        """Forget the result of this task and its parents."""
        self._cache = None
        if self.parent:
            self.parent.forget()

        self.backend.remove_pending_result(self)
        self.backend.forget(self.id)

    def revoke(self, connection=None, terminate=False, signal=None,

P 1 <class 'celery.result.AsyncResult'> (('04ac1f25-34e9-4cad-9172-a6da25bb9e16', (('a60020c1-f6e0-4490-9046-d58b5cec1c71', (('0f155f2e-da48-4b30-a823-2a64783aec02', (('59d26db4-9015-46ad-840e-d8c70acaea32', None), None)), [(('cb1f7131-e873-45ac-a3ea-2f2c63ce1373', None), None), (('2464dbec-c1b0-451d-a598-29c8aa50e0d0', None), None)])), [(('ae3947f8-7c8a-4c7c-9738-0e5149da3d14', None), None), (('b67efae3-5b87-4ff3-8387-56c50691c1cf', None), None)])), None)
P 2 <class 'celery.result.AsyncResult'> (('ac7b21f2-b52c-436f-8caf-7b7e37a00198', (('9c326b8e-f856-4b4a-8b77-3b0fcd899b68', (('1e2cbbe4-6083-4248-aaf3-66ad36c9ea84', (('331e1876-2cd8-41b0-996b-9d73cd5d5b2e', None), None)), [(('a375cb5a-d13a-479c-9d4f-31c06845bc1a', None), None), (('279e9ae8-8f94-4b8c-84a3-761cde63d75f', None), None)])), [(('d18aa723-5b22-40a5-821e-47d9dfa29e2d', None), None), (('685b0126-eb77-4640-b69a-eb6771224361', None), None)])), None)

AsyncResult 4ce877f7-5c99-4300-836c-46ce2d9bf28f
  GroupResult 485ef537-29e0-4eb6-8a94-d813450f9e8c
   GroupResult c260bf73-4d86-4876-ae02-7a611886e762
    AsyncResult f9c6022e-f288-47d9-abf4-0d4ddc31d1a8
    child ac4a795a-ec67-40f3-a979-a57ada69c953
    child 54683854-7c2b-4a1d-8351-b4eb19b938f1
   child 5a65783a-cf45-4727-b22e-faf6b1f6a388
   child 72b5c67c-4327-4450-9db9-86934e53080e

ignore_result = get_actual_ignore_result(task, task_request)
            track_started = not eager and (task.track_started and not ignore_result)
            # #6476
            if eager and not ignore_result and task.store_eager_result:
                publish_result = True
            else:
                publish_result = not eager and not ignore_result

            redelivered = (task_request.delivery_info
                           and task_request.delivery_info.get('redelivered', False))
            if deduplicate_successful_tasks and redelivered:
                if task_request.id in successful_requests:

                    return trace_ok_t(R, I, T, Rstr)
                r = AsyncResult(task_request.id, app=app)

                try:
                    state = r.state
                except BackendGetMetaError:
                    pass
                else:
      
…[4517 chars omitted]…
  except Exception as exc:
                    I, R, state, retval = on_error(task_request, exc)
                    # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)
                    traceback_clear(exc)
                except BaseException:
                    raise
                else:
  File "<stdin>", line 6
    defx=None
IndentationError: unexpected unindent

File "<stdin>", line 7
    def bad(): pass
IndentationError: unexpected unindent

File "<stdin>", line 7
    def bad(): pass
IndentationError: unexpected unindent

Ignore IGNORED  None ['bad']
Reject REJECTED (None, False) None ['bad']
............u.......................................F
=================================== FAILURES ===================================
________________________ test_chain.test_chord_to_group ________________________

self = <t.unit.tasks.test_canvas.test_chain object at 0x7e00c4047830>

    def test_chord_to_group(self):
        c = (
            chord([self.add.s('x0', 'y0'), self.add.s('x1', 'y1')],
                  self.add.s(['foo'])) |

            group([self.add.s(['y']), self.add.s(['z'])])
        )
        assert isinstance(c, _chain)
>       assert c.apply().get() == [
               ^^^^^^^^^
            ['x0y0', 'x1y1', 'foo', 'y'],
            ['x0y0', 'x1y1', 'foo', 'z']
        ]

t/unit/tasks/test_canvas.py:591: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

sel
…[1954 chars omitted]…
r mode there is no broker callback/chain dispatch to stop
+            # the chain for us, so do it here explicitly.  In particular, do
+            # not call get() on the result: both states deliberately have a
+            # value of None and must remain visible on the returned result.
+            if res.state in ('IGNORED', 'REJECTED'):
+                break
         return last
 
     @property

............u........................................................... [ 22%]
....................................................x..x................ [ 45%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 59%]
........................................s............................... [ 82%]
........................................................                 [100%]
312 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 7.01s

File "<stdin>", line 6
    def x(a=1): return a
IndentationError: unexpected unindent

123:    def as_tuple(self):
124-        parent = self.parent
125-        return (self.id, parent.as_tuple() if parent is not None else None), None
126-
127-    def as_list(self):
128-        """Return as a list of task IDs."""
--
1005:    def as_tuple(self):
1006-        return (
1007-            (self.id, self.parent and self.parent.as_tuple()),
1008-            [r.as_tuple() for r in self.results]
1009-        )

1010-
 celery/canvas.py | 8 ++++++++
 celery/result.py | 4 ++--
 2 files changed, 10 insertions(+), 2 deletions(-)
diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..1f5800b 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -25,6 +25,7 @@ from vine import barrier
 
 from celery._state import current_app
 from celery.exceptions import CPendingDeprecationWarning
+from celery.states import IGNORED, REJECTED
 from celery.result import GroupResult, allow_join_r
…[1234 chars omitted]…
     def as_list(self):
         """Return as a list of task IDs."""
@@ -1119,7 +1119,7 @@ def result_from_tuple(r, app=None):
     if not isinstance(r, ResultBase):
         res, nodes = r
         id, parent = res if isinstance(res, (list, tuple)) else (res, None)
-        if parent:
+        if parent is not None:
             parent = result_from_tuple(parent, app)
 
         if nodes is not None:

............u........................................................... [ 22%]
....................................................x..x................ [ 45%]
............................uu.uu.......uuu.uuu.uuu.uuu.uuu.uuu.uuu.uuu. [ 59%]
........................................s............................... [ 82%]
........................................................                 [100%]
312 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 6.86s

submitted.

total 332
drwxr-xr-x 1 root root   4096 Sep 30 20:23 .
drwxr-xr-x 1 root root   4096 Sep 30 20:22 ..
-rw-r--r-- 1 root root    343 Sep 30 20:21 .bumpversion.cfg
-rw-r--r-- 1 root root    306 Sep 30 20:21 .dockerignore
-rw-r--r-- 1 root root    227 Sep 30 20:21 .editorconfig
drwxr-xr-x 1 root root   4096 Sep 30 20:22 .git
drwxr-xr-x 1 root root   4096 Sep 30 20:21 .github
-rw-r--r-- 1 root root    437 Sep 30 20:21 .gitignore
-rw-r--r-- 1 root root   1073 Sep 30 20:21 .pre-commit-config.yaml
-rw-r--r-- 1 root root    616 Sep 30 20:21 .readthedocs.yaml
-rw-r--r-- 1 root root  44762 Sep 30 20:21 CONTRIBUTING.rst

-rw-r--r-- 1 root root   8619 Sep 30 20:21 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128887 Sep 30 20:21 Changelog.rst
-rw-r--r-- 1 root root   2717 Sep 30 20:21 LICENSE
-rw-r--r-- 1 root root    709 Sep 30 20:21 MANIFEST.in
-rw-r--r-- 1 root root   5866 Sep 30 20:21 Makefile
-rw-r-
…[8629 chars omitted]…
e == parent.as_tuple()
./t/unit/tasks/test_stamping.py:9:from celery.exceptions import Ignore
./t/unit/tasks/test_stamping.py:1042:        with pytest.raises(Ignore):
./t/unit/tasks/test_tasks.py:14:from celery.exceptions import Ignore, ImproperlyConfigured, Retry
./t/unit/tasks/test_tasks.py:1175:        with pytest.raises(Ignore):
./t/unit/tasks/test_tasks.py:1183:        with pytest.raises(Ignore):

t/unit/tasks/test_canvas.py:9:from celery.canvas import (Signature, _chain, _maybe_group, _merge_dictionaries, chain, chord, chunks, group,
t/unit/tasks/test_canvas.py:464:    def test_groups_in_chain_to_chord(self):
t/unit/tasks/test_canvas.py:470:    def test_empty_groups_are_skipped_in_chain(self):
t/unit/tasks/test_canvas.py:482:    def test_empty_group_body_is_skipped_when_chain_upgrades_to_chord(self):
t/unit/tasks/test_canvas.py:493:    def test_generator_backed_empty_group_is_not_skipped_when_chained(self):
t/unit/tasks/test_canvas.py:501:    def test_known_empty_generator_backed_group_is_skipped_in_chain(self):
t/unit/tasks/test_canvas.py:639:        ), "Chord followed by a group should be upgraded to a single chord with chained body."
t/unit/tasks/test_canvas.py:894:    def test_group_in_center_of_chain(self):
t/unit/tasks/test_canvas.py:1331:        # for the group precedin
…[1067 chars omitted]…
tasks/test_canvas.py:2026:        # Create a chord with header group and body chain (no task_id set)
t/unit/tasks/test_result.py:527:    def test_join_native_with_group_chain_group(self):

t/unit/tasks/test_result.py:528:        """Test group(chain(group)) case, join_native can be run correctly.
t/unit/tasks/test_result.py:529:        In group(chain(group)) case, GroupResult has no _cache property, and

#: Error raised for timeouts.
    TimeoutError = TimeoutError

    #: The task's UUID.
    id = None

    #: The task result backend to use.
    backend = None

    def __init__(self, id, backend=None,
                 task_name=None,            # deprecated
                 app=None, parent=None):

        if id is None:
            raise ValueError(
                f'AsyncResult requires valid id, not {type(id)}')
        self.app = app_or_default(app or self.app)
        self.id = id
        self.backend = backend or self.app.backend
        self.parent = parent
        self.on_ready = promise(self._on_fulfilled, weak=True)
        self._cache = None
        self._ignored = False

    @property
    def ignored(self):
        """If True, task result retrieval is disabled."""
        if hasattr(self, '_ignored'):
            return self._ignored
        return False

    @ignored.
…[8630 chars omitted]…
_headers(visitor_headers, append_stamps, **headers)
        self.stamp_links(visitor, append_stamps, **headers)

        if isinstance(self.tasks, _regen):
            self.tasks.map(_partial(_stamp_regen_task, visitor=visitor, append_stamps=append_stamps, **headers))
        else:
            new_tasks = []
            for task in self.tasks:
                task = maybe_signature(task, app=self.app)

return ChainMap({'chain': options['chain'] + tasks},
                        options)


@Signature.register_type(name='chain')
class _chain(Signature):
    tasks = getitem_property('kwargs.tasks', 'Tasks in chain.')

    @classmethod
    def from_dict(cls, d, app=None):
        tasks = d['kwargs']['tasks']
        if tasks:

            if isinstance(tasks, tuple):  # aaaargh
                tasks = d['kwargs']['tasks'] = list(tasks)
            tasks = [maybe_signature(task, app=app) for task in tasks]
        return cls(tasks, app=app, **d['options'])

    def __init__(self, *tasks, **options):
        tasks = (regen(tasks[0]) if len(tasks) == 1 and is_list(tasks[0])
                 else tasks)
        super().__init__('celery.chain', (), {'tasks': tasks}, **options
                         )
        self._use_link = options.pop('use_link', None)
        self.subtask_type = 
…[8630 chars omitted]…
ord_error_from_stack.
                    if isinstance(task, chord) and task.body:
                        task.body.link_error(errback)

            tasks.append(task)
            results.append(res)

            prev_task, prev_res = task, res
            if isinstance(task, chord):
                app.backend.ensure_chords_allowed()
                # If the task is a chord, and the body is a chain

tasks[-1].link(sig)
        for sig in maybe_list(self.options.get('link_error')) or []:
            for task in tasks:
                task.link_error(sig)
        return tasks

    def apply_async(self, args=None, kwargs=None, **options):
        # python is best at unpacking kwargs, so .run is here to do that.
        args = args if args else ()
        kwargs = kwargs if kwargs else []
        app = self.app

        if app.conf.task_always_eager:
            with allow_join_result():
                return self.apply(args, kwargs, **options)
        return self.run(args, kwargs, app=app, **(
            dict(self.options, **options) if options else self.options))

    def run(self, args=None, kwargs=None, group_id=None, chord=None,
            task_id=None, link=None, link_error=None, publisher=None,
            producer=None, root_id=None, parent_id=None, app=None,
 
…[8629 chars omitted]…
rent = parent
        self.on_ready = promise(self._on_fulfilled, weak=True)
        self._cache = None
191:    def get(self, timeout=None, propagate=True, interval=0.5,
331:    def get_leaf(self):
505:    def state(self):
726:    def get(self, timeout=None, propagate=True, interval=0.5,
1026:class EagerResult(AsyncResult):
1059:    def get(self, timeout=None, propagate=True,
1098:    def state(self):

869:    def apply(self, args=None, kwargs=None,
                argument provided.
            countdown (float): Time in seconds to delay the retry for.
            eta (~datetime.datetime): Explicit time and date to run the
                retry at.
            max_retries (int): If set, overrides the default retry limit for
                this execution.  Changes to this parameter don't propagate to
                subsequent task retry attempts.  A value of :const:`None`,
                means "use the default", so if you want infinite retries you'd
                have to set the :attr:`max_retries` attribute of the task to
                :const:`None` first.
            time_limit (int): If set, overrides the default time limit.

            soft_time_limit (int): If set, overrides the default soft
                time limit.
            throw (bool): If this is :const:`False`,
…[8629 chars omitted]…
e_gid(options)
        tasks = self._prepared(self.tasks, [], group_id, root_id, app)
        return app.GroupResult(group_id, [
            sig.apply(args=args, kwargs=kwargs, **options) for sig, _, _ in tasks
        ])

    def set_immutable(self, immutable):
        for task in self.tasks:
            task.set_immutable(immutable)

    def stamp(self, visitor=None, append_stamps=False, **headers):

def apply(self, args=None, kwargs=None,
              link=None, link_error=None,
              task_id=None, retries=None, throw=None,
              logfile=None, loglevel=None, headers=None, **options):
        """Execute this task locally, by blocking until the task returns.

        Arguments:
            args (Tuple): positional arguments passed on to the task.
            kwargs (Dict): keyword arguments passed on to the task.
            throw (bool): Re-raise task exceptions.
                Defaults to the :setting:`task_eager_propagates` setting.

        Returns:
            celery.result.EagerResult: pre-evaluated result.
        """
        # trace imports Task, so need to import inline.
        from celery.app.trace import build_tracer

        app = self._get_app()
        args = args or ()
        kwargs = kwargs or {}
        task_id = task_id or uuid()
        re
…[4108 chars omitted]…
self):
        return True

    def get(self, timeout=None, propagate=True,
            disable_sync_subtasks=True, **kwargs):
        if disable_sync_subtasks:
            assert_will_not_block()

        if self.successful():
            return self.result
1714:    def _prepared(self, tasks, partial_args, group_id, root_id, app,
1763:    def _apply_tasks(self, tasks, producer=None, app=None, p=None,

B RUN
<class 'celery.result.EagerResult'> FAILURE unsupported operand type(s) for +: 'NoneType' and 'int'

    def link(self, sig):
        # Simply link to first task. Doing this is slightly misleading because
        # the callback may be executed before all children in the group are
        # completed and also if any children other than the first one fail.
        #
        # The callback signature is cloned and made immutable since it the
        # first task isn't actually capable of passing the return values of its
        # siblings to the callback task.
        sig = sig.clone().set(immutable=True)

        return self.tasks[0].link(sig)

    def link_error(self, sig):
        # Any child task might error so we need to ensure that they are all
        # capable of calling the linked error signature. This opens the
        # possibility that the task is called more than once bu
…[4707 chars omitted]…
f
                # awful race conditions to think about. We'll hope it's not!
                sig, res, group_id = current_task
                chord_obj = chord if chord is not None else sig.options.get("chord")
                # We need to check the chord size of each contributing task so
377:    def successful(self):
381:    def failed(self):
644:    def successful(self):
653:    def failed(self):

73:LOG_IGNORED = """\
78:LOG_REJECTED = """\
92:log_policy_reject = log_policy_t(LOG_REJECTED, 'rejected', logging.WARN, 1, 1)
93:log_policy_ignore = log_policy_t(LOG_IGNORED, 'ignored', logging.INFO, 0, 0)
109:IGNORED = states.IGNORED
110:REJECTED = states.REJECTED
114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
426:    def _dispatch_callbacks_and_chain(
427:        retval, callbacks, chain, parent_id, root_id, priority,
429:        """Dispatch callbacks and chain for a completed task.
431:        Dispatches link callbacks and then the next chain step.
435:        Note: dispatch is not atomic.  If callbacks succeed but the

436:        chain step fails (or vice-versa), a Reject + redeliver may
437:        re-dispatch the already-sent callbacks.  This is acceptable
440:        if callbacks:
441:            if len(callbacks) > 1:
443:                for sig in callbacks:
462: 
…[7689 chars omitted]…
urn self.state == states.SUCCESS

    def failed(self):
        """Return :const:`True` if the task failed."""
        return self.state == states.FAILURE

    def throw(self, *args, **kwargs):
        self.on_ready.throw(*args, **kwargs)

    def maybe_throw(self, propagate=True, callback=None):
        cache = self._get_task_meta() if self._cache is None else self._cache
        state, value, tb = (

# -*- TRACE -*-
                try:
                    if task_before_start:
                        task_before_start(uuid, args, kwargs)

                    R = retval = fun(*args, **kwargs)
                    state = SUCCESS
                except Reject as exc:
                    I, R = Info(REJECTED, exc), ExceptionInfo(internal=True)
                    state, retval = I.state, I.retval
                    I.handle_reject(task, task_request)
                    # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)

                    traceback_clear(exc)
                except Ignore as exc:
                    I, R = Info(IGNORED, exc), ExceptionInfo(internal=True)
                    state, retval = I.state, I.retval
                    I.handle_ignore(task, task_request)
                    # MEMORY LEAK FIX: Clear traceback 
…[8437 chars omitted]…
n options, group_id, options.get('root_id')

    def _freeze_group_tasks(self, _id=None, group_id=None, chord=None,
                            root_id=None, parent_id=None, group_index=None):
        """Freeze the tasks in the group.

        Note:
            If the group tasks are created from a generator, the tasks generator would
            not be exhausted, and the tasks would be frozen lazily.

370:    def apply_async(self, args=None, kwargs=None, route_name=None, **options):
1036:    def apply_async(self, args=None, kwargs=None, **options):
1398:    def apply_async(self, args=None, kwargs=None, **opts):
1455:    def apply_async(self, args=None, kwargs=None, **opts):
1616:    def apply_async(self, args=None, kwargs=None, add_to_parent=True,
2153:    def apply_async(self, args=None, kwargs=None, task_id=None,
                task=task_name, args=tuple(args or ()),
                kwargs=kwargs or {},
                options=dict(options or {}, **ex),
                subtask_type=subtask_type,
                immutable=immutable,
            )

    def __call__(self, *partial_args, **partial_kwargs):
        """Call the task directly (in the current process)."""
        args, kwargs, _ = self._merge(partial_args, partial_kwargs, None)
        return self.type(*args, **kwargs)

…[4835 chars omitted]…
                           loader_cleanup()
                        except (KeyboardInterrupt, SystemExit, MemoryError):
                            raise
                        except Exception as exc:
                            logger.error('Process cleanup failed: %r', exc,
                                         exc_info=True)
        except MemoryError:
            raise
        except Reject:

536:    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,

            queue (str, kombu.Queue): The queue to route the task to.
                This must be a key present in :setting:`task_queues`, or
                :setting:`task_create_missing_queues` must be
                enabled.  See :ref:`guide-routing` for more
                information.

            exchange (str, kombu.Exchange): Named custom exchange to send the
                task to.  Usually not used in combination with the ``queue``
                argument.

            routing_key (str): Custom routing key used to route the task to a
                worker server.  If in combination with a ``queue`` argument
                only used to specify custom routing keys to topic exchanges.

            priority (int): The task priority, a number between 0 and 9.
                The interpretat
…[4036 chars omitted]…
ptions)

        preopts = self._get_exec_options()
        options = dict(preopts, **options) if options else preopts

        options.setdefault('ignore_result', self.ignore_result)
        if self.priority:
            options.setdefault('priority', self.priority)
952:        self._use_link = options.pop('use_link', None)
1063:        use_link = self._use_link
1163:        use_link = self._use_link

options.setdefault('priority', self.priority)

        app = self._get_app()
        if app.conf.task_always_eager:
            with app.producer_or_acquire(producer) as eager_producer:
                serializer = options.get('serializer')
                if serializer is None:
                    if eager_producer.serializer:
                        serializer = eager_producer.serializer
                    else:
                        serializer = app.conf.task_serializer
                body = args, kwargs

                content_type, content_encoding, data = serialization.dumps(
                    body, serializer,
                )
                args, kwargs = serialization.loads(
                    data, content_type, content_encoding,
                    accept=[content_type]
                )
            with denied_join_result():
                return self
…[6504 chars omitted]…

                    )
                if tasks:
                    prev_task = tasks[-1]
                    prev_res = results[-1]
                else:
                    prev_task = None
                    prev_res = None

            if is_last_task:
                # chain(task_id=id) means task id is set for the last task
                # in the chain.  If the chord is part of a chord/group

206:    def handle_reject(self, task, req, **kwargs):
209:    def handle_ignore(self, task, req, **kwargs):
trace_ok_t = namedtuple('trace_ok_t', ('retval', 'info', 'runtime', 'retstr'))


def info(fmt, context):
    """Log 'fmt % context' with severity 'INFO'.

    'context' is also passed in extra with key 'data' for custom handlers.
    """
    logger.info(fmt, context, extra={'data': context})

def task_has_custom(task, attr):
    """Return true if the task overrides ``attr``."""
    return mro_lookup(task.__class__, attr, stop={BaseTask, object},
                      monkey_patched=['celery.app.task'])


def get_log_policy(task, einfo, exc):
    if isinstance(exc, Reject):
        return log_policy_reject
    elif isinstance(exc, Ignore):
        return log_policy_ignore
    elif einfo.internal:
        return log_policy_internal
    else:
        if task.throws and isinstance(
…[957 chars omitted]…
se" from "not set". We check
    # __dict__ to detect only instance-level (i.e., explicitly set) values.
    if isinstance(req, Context) and 'ignore_result' not in req.__dict__:
        actual = None

    return actual if actual is not None else task.ignore_result


class TraceInfo:
    """Information about task execution."""
  File "<stdin>", line 7
    defx=None
IndentationError: unexpected unindent

ignore_result = get_actual_ignore_result(task, req)

        if ignore_result:
            store_errors = task.store_errors_even_if_ignored
        elif eager and task.store_eager_result:
            store_errors = True
        else:
            store_errors = not eager

        return {
            RETRY: self.handle_retry,
            FAILURE: self.handle_failure,

        }[self.state](task, req,
                      store_errors=store_errors,
                      call_errbacks=call_errbacks)

    def handle_reject(self, task, req, **kwargs):
        self._log_error(task, req, ExceptionInfo())

    def handle_ignore(self, task, req, **kwargs):
        self._log_error(task, req, ExceptionInfo())

    def handle_retry(self, task, req, store_errors=True, **kwargs):
        """Handle retry exception."""
        # the exception raised is the Retry semi-predicate,
        # and it's exc' attribute is the original exception raised (if any).
        type_, _, tb = sys.exc_info()
        einfo = None
        try:
            reason = self.retval
            einfo = ExceptionInfo((type_, reason, tb))
  File "<stdin>", line 7
    def_bad = None
IndentationError: unexpected unindent

877:def _prepare_chain_from_options(options, tasks, use_link):
878-    # When we publish groups we reuse the same options dictionary for all of
879-    # the tasks in the group. See:
880-    # https://github.com/celery/celery/blob/fb37cb0b8/celery/canvas.py#L1022.
881-    # Issue #5354 reported that the following type of canvases
882-    # causes a Celery worker to hang:
883-    # group(
884-    #   add.s(1, 1),
885-    #   add.s(1, 1)
886-    # ) | tsum.s() | add.s(1) | group(add.s(1), add.s(1))
887-    # The resolution of #5354 in PR #5681 was to only set the `chain` key
888-    # in the options dictionary if it is not present.

889-    # Otherwise we extend the existing list of tasks in the chain with the new
890-    # tasks: options['chain'].extend(chain_).
891-    # Before PR #5681 we overrode the `chain` key in each iteration
892-    # of the loop which applies all the tasks in t
…[632 chars omitted]…
, even though there might not be
903-    # one to begin with. Alternatively, it causes Celery to execute more tasks
904-    # that were previously present in the previous task in the group.
905-    # The solution is to be careful and never mutate the options dictionary
906-    # to begin with.
907-    # Here is an example of a canvas which triggers this issue:
Ignore SUCCESS 2 [1]
Reject SUCCESS 2 [1]

426:    def _dispatch_callbacks_and_chain(
541:                                _dispatch_callbacks_and_chain(
619:                        _dispatch_callbacks_and_chain(
    # https://github.com/celery/celery/blob/fb37cb0b8/celery/canvas.py#L1022.
    # Issue #5354 reported that the following type of canvases
    # causes a Celery worker to hang:
    # group(
    #   add.s(1, 1),
    #   add.s(1, 1)
    # ) | tsum.s() | add.s(1) | group(add.s(1), add.s(1))
    # The resolution of #5354 in PR #5681 was to only set the `chain` key
    # in the options dictionary if it is not present.

    # Otherwise we extend the existing list of tasks in the chain with the new
    # tasks: options['chain'].extend(chain_).
    # Before PR #5681 we overrode the `chain` key in each iteration
    # of the loop which applies all the tasks in the group:
    # options['chain'] = tasks if not use_link else None
…[3645 chars omitted]…
has already been prepared, and res is
                # set to the last task in the callback chain.

                # We need to change that so that it points to the
                # group result object.
                node = res
                while node.parent:
                    node = node.parent
                prev_res = node
        # Use the last task's actual ID, not the input parameter.

ignore_result = get_actual_ignore_result(task, task_request)
            track_started = not eager and (task.track_started and not ignore_result)
            # #6476
            if eager and not ignore_result and task.store_eager_result:
                publish_result = True
            else:
                publish_result = not eager and not ignore_result

            redelivered = (task_request.delivery_info
                           and task_request.delivery_info.get('redelivered', False))
            if deduplicate_successful_tasks and redelivered:
                if task_request.id in successful_requests:

                    return trace_ok_t(R, I, T, Rstr)
                r = AsyncResult(task_request.id, app=app)

                try:
                    state = r.state
                except BackendGetMetaError:
                    pass
                else:
      
…[4525 chars omitted]…
egen(tasks2)
        else:
            new_tasks = []
            # Need to unroll subgroups early so that chord gets the
            # right result instance for chord_unlock etc.
            results = list(self._freeze_unroll(
                new_tasks, group_id, chord, root_id, parent_id,
            ))
            if isinstance(self.tasks, MutableSequence):
                self.tasks[:] = new_tasks

f02ada1 base
1 SUCCESS [1]
2 SUCCESS [1]

"""Composing task work-flows.

.. seealso:

    You should import these from :mod:`celery` and not this module.
"""

import itertools
import operator
import types
import warnings
from abc import ABCMeta, abstractmethod

from collections import deque
from collections.abc import MutableSequence
from copy import deepcopy
from functools import partial as _partial
from functools import reduce
from operator import itemgetter
from types import GeneratorType

from kombu.utils.functional import fxrange, reprcall
from kombu.utils.objects import cached_property
from kombu.utils.uuid import uuid
from vine import barrier

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
from c
…[221 chars omitted]…
rt remove_repeating_from_task, truncate

__all__ = (
    'Signature', 'chain', 'xmap', 'xstarmap', 'chunks',
    'group', 'chord', 'signature', 'maybe_signature',
)


def maybe_unroll_group(group):
    """Unroll group with only one member.
    This allows treating a group of a single task as if it
  File "<stdin>", line 4
    @app.task def t(x): return x+1
              ^^^
SyntaxError: invalid syntax

(retval,), chain=chain[:-1],
                parent_id=parent_id, root_id=root_id,
                priority=priority,
            )

    def trace_task(uuid, args, kwargs, request=None):
        # R      - is the possibly prepared return value.
        # I      - is the Info object.
        # T      - runtime
        # Rstr   - textual representation of return value
        # retval - is the always unmodified return value.
        # state  - is the resulting task state.

        # This function is very long because we've unrolled all the calls
        # for performance reasons, and because the function is so long
        # we want the main variables (I, and R) to stand out visually from the
        # the rest of the variables, so breaking PEP8 is worth it ;)
        R = I = T = Rstr = retval = state = None
        task_request = None
        time_start = monotonic()
  
…[2999 chars omitted]…
       self.tasks = new_tasks
        return gid, results

    def freeze(self, _id=None, group_id=None, chord=None,
               root_id=None, parent_id=None, group_index=None):
        return self.app.GroupResult(*self._freeze_group_tasks(
            _id=_id, group_id=group_id,
            chord=chord, root_id=root_id, parent_id=parent_id, group_index=group_index
        ))

    _freeze = freeze

try:
                    state = r.state
                except BackendGetMetaError:
                    pass
                else:
                    if state == SUCCESS:
                        info(LOG_IGNORED, {
                            'id': task_request.id,
                            'name': get_task_name(task_request, name),
                            'description': 'Task already completed successfully.'
                        })
                        _root_id = task_request.root_id or uuid

                        _priority = task_request.delivery_info.get('priority') if \
                            inherit_parent_priority else None
                        try:
                            _meta = r._get_task_meta()
                            stored_retval = _meta.get('result')
                            # Children are populated by mark_as_done on the
…[1125 chars omitted]…
                      # will requeue indefinitely.  Broker-level
                            # dead-letter / max-delivery-count policies are
                            # the intended circuit-breaker.
                            logger.error(
                                'Failed to dispatch chain/callbacks for '
  File "<stdin>", line 6
    def f(x): return x+1
IndentationError: unexpected unindent

I, R, state, retval = on_error(
                        task_request, exc, RETRY, call_errbacks=False)
                    # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)
                    traceback_clear(exc)
                except Exception as exc:
                    I, R, state, retval = on_error(task_request, exc)
                    # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)
                    traceback_clear(exc)
                except BaseException:
                    raise
                else:
                    try:

                        # callback tasks must be applied before the result is
                        # stored, so that result.children is populated.

                        # groups are called inline and will store trail
                        # separately, s
…[1487 chars omitted]…
uest, name),
                                'return_value': Rstr,
                                'runtime': T,
                                'args': task_request.get('argsrepr') or safe_repr(args),
                                'kwargs': task_request.get('kwargsrepr') or safe_repr(kwargs),
                            })

                # -* POST *-
                if state not in IGNORE_STATES:

a chain None callbacks None
B
0ae73cc7-ddda-4b21-a44a-a73b9121cb53

512:                    return trace_ok_t(R, I, T, Rstr)
561:                        return trace_ok_t(R, I, T, Rstr)
684:        return trace_ok_t(R, I, T, Rstr)
700:        return trace_ok_t(report_internal_error(task, exc), TraceInfo(FAILURE, exc), 0.0, None)
                if state not in IGNORE_STATES:
                    if task_after_return:
                        task_after_return(
                            state, retval, uuid, args, kwargs, None,
                        )
            finally:
                try:
                    if postrun_receivers:

                        send_postrun(sender=task, task_id=uuid, task=task,
                                     args=args, kwargs=kwargs,
                                     retval=retval, state=state)
                finally:
                    pop_task()
                    pop_request()
                    if not eag
…[941 chars omitted]…
                 if (_callbacks or _chain) and not _children:
541:                                _dispatch_callbacks_and_chain(
542:                                    stored_retval, _callbacks, _chain,
555:                                'Failed to dispatch chain/callbacks for '
619:                        _dispatch_callbacks_and_chain(
620:                            retval, task.request.callbacks,

Traceback (most recent call last):
  File "<stdin>", line 7, in <module>
  File "/repo/celery/canvas.py", line 1100, in freeze
    _, results = self._frozen = self.prepare_steps(
                                ^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1279, in prepare_steps
    app.backend.ensure_chords_allowed()
  File "/repo/celery/backends/base.py", line 1272, in ensure_chords_allowed
    raise NotImplementedError(E_CHORD_NO_BACKEND.strip())
NotImplementedError: Starting chords requires a result backend to be configured.

Note that a group chained with a task is also upgraded to be a chord,

as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.

81:    callbacks = None
179:            'link': self.callbacks,
916:            'callbacks': maybe_list(link),
942:        ret = tracer(task_id, args, kwargs, request)
1054:        # Ensure callbacks or errbacks from the replaced signature are retained
1058:        for callback in maybe_list(self.request.callbacks) or []:
1062:        # If the replacement signature is a chain, we need to push callbacks
            propagate=throw, app=self._get_app(),
        )
        ret = tracer(task_id, args, kwargs, request)
        retval = ret.retval
        if isinstance(retval, ExceptionInfo):

            retval, tb = retval.exception, retval.traceback
            if isinstance(retval, ExceptionWithTraceback):
                retval = retval.exc
        if isinstance(retval, Retry) and retval.sig is not None:
            return retval.sig.apply(retries=retries + 1)
        state = states.SUCC
…[395 chars omitted]…
                                        task_name=self.name, **kwargs)

    def signature(self, args=None, *starargs, **starkwargs):
        """Create signature.

        Returns:
            :class:`~celery.signature`:  object for
                this task, wrapping arguments and execution options
                for a single task invocation.
        """
        starkwargs.setdefault('app', self.app)

{'task_id': 'e48c782e-df80-40b2-843c-60241a667525', 'reply_to': '065b9d60-dc52-3528-a9e3-c85886380277'} <@task: x.b of x at 0x7ad7f24d3c50> None
{'task_id': 'c1e88987-de23-4d6f-abf7-5c9961dbd9a9', 'reply_to': '065b9d60-dc52-3528-a9e3-c85886380277'}

def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                    link=None, link_error=None, shadow=None, **options):
        """Apply tasks asynchronously by sending a message.

        Arguments:
            args (Tuple): The positional arguments to pass on to the task.

            kwargs (Dict): The keyword arguments to pass on to the task.

            countdown (float): Number of seconds into the future that the
                task should execute.  Defaults to immediate execution.

            eta (~datetime.datetime): Absolute time and date of when the task
                should be executed.  May not be specified if `countdown`
                is also supplied.

            expires (float, ~datetime.datetime): Datetime or
                seconds in the future for the task should expire.
                The task won't be executed after the exp
…[1463 chars omitted]…
 is None
                else [self.time_limit, self.soft_time_limit]
            ),
            'ignore_result': options.get('ignore_result', False),
            'delivery_info': {
                'is_eager': True,
                'exchange': options.get('exchange'),
                'routing_key': options.get('routing_key'),
                'priority': options.get('priority'),
            }
        }

Ignore SUCCESS 2 [1]
Reject SUCCESS 2 [1]

IGNORED 
684:        return trace_ok_t(R, I, T, Rstr)
685-
686-    return trace_task
--
697-        raise
698-    except Exception as exc:
699-        _signal_internal_error(task, uuid, args, kwargs, request, exc)
700:        return trace_ok_t(report_internal_error(task, exc), TraceInfo(FAILURE, exc), 0.0, None)
701-
702-

Ignore SUCCESS 2 [1]
Reject SUCCESS 2 [1]

SUCCESS [1]
 celery/app/task.py | 16 ++++++++++++++--
 celery/canvas.py   |  6 ++++++
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..7f3e426 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -25,6 +25,7 @@ from vine import barrier
 
 from celery._state import current_app
 from celery.exceptions import CPendingDeprecationWarning
+from celery import states
 from celery.result import GroupResult, allow_join_result
 from celery.utils import abstract
 from celery.utils.collections import ChainMap

@@ -1081,6 +1082,11 @@ class _chain(Signature):
             options = _prepare_chain_from_options(options, tasks, use_link)
 
             result_from_apply = first_task.apply_async(**options)
+            # In eager mode semipredicates are represented by an EagerResult
+            # rather than being raised.  They terminate a chain just as they
+            # do in the worker; do not run the remaining tasks.
+            if result_from_apply.state in (states.IGNORED, states.REJECTED):
+                return result_from_apply
             # If we only have a single task, it may be important that we pass
             # the real result object rather than the one obtained via freezing.
             # e.g. For `GroupResult`s, we need to pass back the result object

['x.b', 'x.a'] x.a
B
SUCCESS

SUCCESS [1]

Ignore IGNORED []
Reject REJECTED []

0
 celery/app/task.py | 16 ++++++++++++++--
 celery/canvas.py   | 25 ++++++++++++++++++++++++-
 2 files changed, 38 insertions(+), 3 deletions(-)

0

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..828894f 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -25,6 +25,7 @@ from vine import barrier
 
 from celery._state import current_app
 from celery.exceptions import CPendingDeprecationWarning
+from celery import states
 from celery.result import GroupResult, allow_join_result
 from celery.utils import abstract
 from celery.utils.collections import ChainMap

@@ -1040,8 +1041,17 @@ class _chain(Signature):
         app = self.app
 
         if app.conf.task_always_eager:
+            # Execute eagerly in order instead of encoding the remainder as
+            # links: links are dispatched by Task.apply before its result is
+            # returned, which would make Ignore/Reject non-terminal.
             with allow_join_result():
-                return self.apply(args, kwargs, **options)
+                result =
…[1370 chars omitted]…
n the worker; do not run the remaining tasks.
+            if result_from_apply.state in (states.IGNORED, states.REJECTED):
+                return result_from_apply
             # If we only have a single task, it may be important that we pass
             # the real result object rather than the one obtained via freezing.
             # e.g. For `GroupResult`s, we need to pass back the result object

0
 celery/canvas.py | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

submitted.

CONTRIBUTING.rst
CONTRIBUTORS.txt
Changelog.rst
LICENSE
MANIFEST.in
Makefile
README.rst
SECURITY.md
TODO
bandit.json
celery
celery.egg-info

docker
docs
examples
extra
helm-chart
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
__init__.py
__main__.py
_state.py
app
apps
backends
beat.py
bin
bootsteps.py
canvas.py
concurrency
contrib
events
exceptions.py
fixups
loaders
local.py
platforms.py
result.py
schedules.py
security
signals.py
states.py
utils
worker
0d5eed3 base

__init__.py
test_canvas.py
test_chord.py
test_context.py
test_result.py
test_stamping.py
test_states.py
test_tasks.py
test_trace.py
unit_tasks.py
/repo/t/unit/tasks/test_canvas.py
/repo/t/unit/tasks/test_chord.py

/repo/t/unit/tasks/test_result.py
/repo/t/unit/tasks/test_stamping.py
/repo/t/unit/tasks/test_tasks.py
/repo/t/unit/tasks/test_trace.py

246:      as ``Task.apply_async``:
278:        options (Dict): Additional options to :meth:`Task.apply_async`.
353:        """Shortcut to :meth:`apply_async` using star arguments."""
354:        return self.apply_async(partial_args, partial_kwargs)
359:        Same as :meth:`apply_async` but executed the task inline instead
370:    def apply_async(self, args=None, kwargs=None, route_name=None, **options):
383:            :meth:`~@Task.apply_async` and the :ref:`guide-calling` guide.
390:            _apply = self._apply_async
821:        return self.apply_async().get()
859:    def _apply_async(self):
861:            return self.type.apply_async
958:            return self.apply_async(args, kwargs)

1036:    def apply_async(self, args=None, kwargs=None, **options):
1083:            result_from_apply = first_task.apply_async(**options)
1398:    def apply_async(self, args=None, kwargs=None
…[430 chars omitted]…
  This is used by :meth:`apply_async` to run all the tasks in the group
1812:                sig.apply_async(producer=producer, add_to_parent=False,
2075:        return self.apply_async((), {'body': body} if body else {}, **options)
2153:    def apply_async(self, args=None, kwargs=None, task_id=None,
2277:            header_result = header.apply_async(partial_args, kwargs, task_id=group_id, **options)

else:
                # chain | task/chord -> chain
                # use type(self) for _chain subclasses
                return type(self)(seq_concat_item(
                    self.unchain_tasks(), other), app=self._app)
        else:
            return NotImplemented

    def clone(self, *args, **kwargs):
        to_signature = maybe_signature
        signature = super().clone(*args, **kwargs)
        signature.kwargs['tasks'] = [

            to_signature(sig, app=self._app, clone=True)
            for sig in signature.kwargs['tasks']
        ]
        return signature

    def unchain_tasks(self):
        """Return a list of tasks in the chain.

        The tasks list would be cloned from the chain's tasks.
        All of the chain callbacks would be added to the last task in the (cloned) chain.
        All of the tasks would be linked to the same error callback
      
…[8025 chars omitted]…
sted, they are unrolled - all tasks within
                # groups should be called in parallel
                task = maybe_unroll_group(task)
                if (
                    isinstance(task, group) and
                    isinstance(task.tasks, (list, tuple)) and
                    not task.tasks and
                    (steps or prev_task)
                ):
                    continue

# first task gets partial args from chain
            if clone:
                if is_first_task:
                    task = task.clone(args, kwargs)
                else:
                    task = task.clone()
            elif is_first_task:
                task.args = tuple(args) + tuple(task.args)

            if isinstance(task, _chain):
                # splice (unroll) the chain
                steps_extend(task.tasks)

                continue

            # TODO why isn't this asserting is_last_task == False?
            if isinstance(task, group) and prev_task:
                # automatically upgrade group(...) | s to chord(group, s)
                # for chords we freeze by pretending it's a normal
                # signature instead of a group.
                tasks.pop()
                results.pop()
                try:
                    task = chord(
     
…[6131 chars omitted]…
return cls(*cls._unpack_args(d['kwargs']), app=app, **d['options'])

    def __init__(self, task, it, **options):
        super().__init__(self._task_name, (),
                         {'task': task, 'it': regen(it)}, immutable=True, **options
                         )

    def apply_async(self, args=None, kwargs=None, **opts):
        # need to evaluate generators
        args = args if args else ()

13:from celery.exceptions import Ignore, ImproperlyConfigured, MaxRetriesExceededError, Reject, Retry
864:            raise Reject(exc, requeue=False)
1042:            ~@Ignore: This is always raised when called in asynchronous context.
1227:            raise Ignore('Replaced by new task')

869:    def apply(self, args=None, kwargs=None,
870-              link=None, link_error=None,
871-              task_id=None, retries=None, throw=None,
872-              logfile=None, loglevel=None, headers=None, **options):
873-        """Execute this task locally, by blocking until the task returns.
874-
875-        Arguments:
876-            args (Tuple): positional arguments passed on to the task.
877-            kwargs (Dict): keyword arguments passed on to the task.
878-            throw (bool): Re-raise task exceptions.
879-                Defaults to the :setting:`task_eager_propagates` setting.
880-

881-        Returns:
882-            celery.result.EagerResult: pre-evaluated result.
883-        """
884-        # trace imports Task, so need to import inline.
885-        from celery.app.trace import build_tracer
886-
887-        app = self._get_app()
888-        args = args or
…[1236 chars omitted]…
   else [self.time_limit, self.soft_time_limit]
922-            ),
923-            'ignore_result': options.get('ignore_result', False),
924-            'delivery_info': {
925-                'is_eager': True,
926-                'exchange': options.get('exchange'),
927-                'routing_key': options.get('routing_key'),
928-                'priority': options.get('priority'),
929-            }

}
        }
        if 'stamped_headers' in options:
            request['stamped_headers'] = maybe_list(options['stamped_headers'])
            request['stamps'] = {
                header: maybe_list(options.get(header, [])) for header in request['stamped_headers']
            }

        tb = None
        tracer = build_tracer(
            task.name, task, eager=True,
            propagate=throw, app=self._get_app(),

        )
        ret = tracer(task_id, args, kwargs, request)
        retval = ret.retval
        if isinstance(retval, ExceptionInfo):
            retval, tb = retval.exception, retval.traceback
            if isinstance(retval, ExceptionWithTraceback):
                retval = retval.exc
        if isinstance(retval, Retry) and retval.sig is not None:
            return retval.sig.apply(retries=retries + 1)
        state = states.SUCCESS if ret.info is No
…[524 chars omitted]…
 signature.

        Returns:
            :class:`~celery.signature`:  object for
                this task, wrapping arguments and execution options
                for a single task invocation.
        """
        starkwargs.setdefault('app', self.app)
        return signature(self, args, *starargs, **starkwargs)
    subtask = signature

    def s(self, *args, **kwargs):
        """Create signature.

13:from billiard.einfo import ExceptionInfo, ExceptionWithTraceback
23:from celery.exceptions import BackendGetMetaError, Ignore, InvalidTaskError, Reject, Retry
73:LOG_IGNORED = """\
78:LOG_REJECTED = """\
92:log_policy_reject = log_policy_t(LOG_REJECTED, 'rejected', logging.WARN, 1, 1)
93:log_policy_ignore = log_policy_t(LOG_IGNORED, 'ignored', logging.INFO, 0, 0)
109:IGNORED = states.IGNORED
110:REJECTED = states.REJECTED
114:IGNORE_STATES = frozenset({IGNORED, RETRY, REJECTED})
120:trace_ok_t = namedtuple('trace_ok_t', ('retval', 'info', 'runtime', 'retstr'))
123:def info(fmt, context):
128:    logger.info(fmt, context, extra={'data': context})

137:def get_log_policy(task, einfo, exc):
138:    if isinstance(exc, Reject):
140:    elif isinstance(exc, Ignore):
142:    elif einfo.internal:
216:        type_, _, tb = sys.exc_info()
217:        einfo = None
220:            einfo = Exce
…[787 chars omitted]…
 up here
277:            return einfo
284:    def _log_error(self, task, req, einfo):
285:        eobj = einfo.exception = get_pickled_exception(einfo.exception)
287:            eobj = einfo.exception = eobj.exc
288:        exception, traceback, exc_info, sargs, skwargs = (
290:            safe_str(einfo.traceback),
291:            einfo.exc_info,
295:        policy = get_log_policy(task, einfo, eobj)

1026:class EagerResult(AsyncResult):
1027-    """Result that we know has already been executed."""
1028-
1029-    def __init__(self, id, ret_value, state, traceback=None, name=None):
1030-        # pylint: disable=super-init-not-called
1031-        # XXX should really not be inheriting from AsyncResult
1032-        self.id = id
1033-        self._result = ret_value
1034-        self._state = state
1035-        self._traceback = traceback
1036-        self._name = name
1037-        self.on_ready = promise()

1038-        self.on_ready(self)
1039-
1040-    def then(self, callback, on_error=None, weak=False):
1041-        return self.on_ready.then(callback, on_error)
1042-
1043-    def _get_task_meta(self):
1044-        return self._cache
1045-
1046-    def __reduce__(self):
1047-        return self.__class__, self.__reduce_args__()
1048-
1049-    def __reduce_args__(self):
1050-        r
…[773 chars omitted]…
 # XXX Compat (remove 5.0)
1072-
1073-    def forget(self):
1074-        pass
1075-
1076-    def revoke(self, *args, **kwargs):
1077-        self._state = states.REVOKED
1078-
1079-    def __repr__(self):
1080-        return f'<EagerResult: {self.id}>'
1081-
1082-    @property
1083-    def _cache(self):
1084-        return {
1085-            'task_id': self.id,
1086-            'result': self._result,

27:    'GroupResult', 'EagerResult', 'result_from_tuple',
123:    def as_tuple(self):
125:        return (self.id, parent and parent.as_tuple()), None
482:                result_from_tuple(child, self.app) for child in children
1005:    def as_tuple(self):
1007:            (self.id, self.parent and self.parent.as_tuple()),
1008:            [r.as_tuple() for r in self.results]
1113:def result_from_tuple(r, app=None):
1123:            parent = result_from_tuple(parent, app)
1127:                id, [result_from_tuple(child, app) for child in nodes],

elif isinstance(other, str):
            return other == self.id
        return NotImplemented

    def __repr__(self):
        return f'<{type(self).__name__}: {self.id} [{", ".join(r.id for r in self.results)}]>'

    def __str__(self):
        """`str(self) -> self.id`."""
        return str(self.id)

    def __hash__(self):

        """`hash(self) -> hash(self.id)`."""
        return hash(self.id)

    def as_tuple(self):
        return (
            (self.id, self.parent and self.parent.as_tuple()),
            [r.as_tuple() for r in self.results]
        )

    @property
    def children(self):
        return self.results

    @classmethod
    def restore(cls, id, backend=None, app=None):
        """Restore previously saved group result."""
        app = app or (
            cls.app if not isinstance(cls.app, property) else current_app
        )
        backend = backend 
…[2639 chars omitted]…
    res, nodes = r
        id, parent = res if isinstance(res, (list, tuple)) else (res, None)
        if parent:
            parent = result_from_tuple(parent, app)

        if nodes is not None:
            return app.GroupResult(
                id, [result_from_tuple(child, app) for child in nodes],
                parent=parent,
            )

        return Result(id, parent=parent)
    return r

14:from celery.result import AsyncResult, EagerResult, GroupResult, ResultSet, assert_will_not_block, result_from_tuple
163:        child = [self.app.AsyncResult(uuid()).as_tuple()
235:        r1._set_cache({'children': [r2.as_tuple()]})
1036:        assert x, result_from_tuple(x.as_tuple() == self.app)
1037:        assert x, result_from_tuple(x == self.app)
1042:        y = result_from_tuple(x.as_tuple(), self.app)
1049:        x = result_from_tuple([uid, []], app=self.app)
1062:        assert x, result_from_tuple(x.as_tuple() == self.app)
1063:        assert x, result_from_tuple(x == self.app)
1071:        second_result = result_from_tuple(result.as_tuple(), self.app)
1075:    def test_GroupResult_as_tuple(self):
1083:        (result_id, parent_tuple), group_results = result.as_tuple()

1085:        assert parent_tuple == parent.as_tuple()
/repo/celery/backends/base.py:531:          
…[314 chars omitted]…
result': result.as_tuple()}), states.SUCCESS)
/repo/celery/result.py:123:    def as_tuple(self):
/repo/celery/result.py:125:        return (self.id, parent and parent.as_tuple()), None
/repo/celery/result.py:1005:    def as_tuple(self):
/repo/celery/result.py:1007:            (self.id, self.parent and self.parent.as_tuple()),
/repo/celery/result.py:1008:            [r.as_tuple() for r in self.results]

with pytest.raises(RuntimeError):
            res_subtask_async.get()
        res_subtask_async.get(disable_sync_subtasks=False)

    def test_populate_name(self):
        res = EagerResult('x', 'x', states.SUCCESS, None, 'test_task')
        assert res.name == 'test_task'

        res = EagerResult('x', 'x', states.SUCCESS, name='test_task_named_argument')
        assert res.name == 'test_task_named_argument'

class test_tuples:

    def test_AsyncResult(self):
        x = self.app.AsyncResult(uuid())
        assert x, result_from_tuple(x.as_tuple() == self.app)
        assert x, result_from_tuple(x == self.app)

    def test_with_parent(self):
        x = self.app.AsyncResult(uuid())
        x.parent = self.app.AsyncResult(uuid())
        y = result_from_tuple(x.as_tuple(), self.app)
        assert y == x
        assert y.parent == x.parent
        assert isinstance(y.paren
…[1244 chars omitted]…
   (result_id, parent_tuple), group_results = result.as_tuple()
        assert result_id == result.id
        assert parent_tuple == parent.as_tuple()
        assert parent_tuple[0][0] == parent.id
        assert isinstance(group_results, list)
        expected_grp_res = [((f'async-result-{i}', None), None)
                            for i in range(2)]
        assert group_results == expected_grp_res

Traceback (most recent call last):
  File "/tmp/repro1.py", line 25, in <module>
    print(type(exc_task).__name__, res.state, res.get())
                                              ^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
  File "/tmp/repro1.py", line 9, in add
    def add(x, y): return x + y
                          ~~^~~

TypeError: unsupported operand type(s) for +: 'NoneType' and 'int'

36:.. state:: PROPAGATE_STATES
38:PROPAGATE_STATES
58:    'EXCEPTION_STATES', 'PROPAGATE_STATES', 'precedence', 'state',
139:REJECTED = 'REJECTED'
142:IGNORED = 'IGNORED'
147:PROPAGATE_STATES = frozenset({FAILURE, REVOKED})

356:    def apply(self, args=None, kwargs=None, **options):
1294:    def apply(self, args=None, kwargs=None, **options):
1652:    def apply(self, args=None, kwargs=None, **options):
8:import itertools
9:import operator
10:import types
11:import warnings
12:from abc import ABCMeta, abstractmethod
13:from collections import deque
14:from collections.abc import MutableSequence
15:from copy import deepcopy
16:from functools import partial as _partial

17:from functools import reduce
18:from operator import itemgetter
19:from types import GeneratorType
21:from kombu.utils.functional import fxrange, reprcall
22:from kombu.utils.objects import cached_property
23:from kombu.utils.uuid import uuid
24:from vine import barrier
26:from celery._state import current_app
27:from celery.exceptions import CPendingDeprecationWarning
28:from celery.result import GroupResult, allow_join_result
29:from celery.utils import abstract

def apply(self, args=None, kwargs=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        last, (fargs, fkwargs) = None, (args, kwargs)
        for task in self.tasks:
            res = task.clone(fargs, fkwargs).apply(
                last and (last.get(),), **dict(self.options, **options))
            res.parent, last, (fargs, fkwargs) = last, res, (None, None)
        return last

1302:            if res.state in (states.IGNORED, states.REJECTED):

from celery._state import current_app
from celery.exceptions import CPendingDeprecationWarning
from celery.result import GroupResult, allow_join_result
from celery import states
from celery.utils import abstract
from celery.utils.collections import ChainMap
from celery.utils.functional import _regen
PromiseProxy IGNORED None
PromiseProxy REJECTED None

--- original ---
 AsyncResult 52ad6806-acb9-4791-94c4-bdd88bb645b5
   GroupResult 67e833a9-6a81-4953-83f0-669f11ad55c3
     AsyncResult 3e561eed-1bf4-435f-8250-ebb14f022a10
     AsyncResult d8b60ebc-56da-43f5-87e3-2e5af0ed14af
     GroupResult a03c5631-a109-4051-908e-9455e83a8bba
       AsyncResult 11de71fb-30ec-44e7-9fd6-6d186678d316
       AsyncResult 75d720de-eaed-48fb-a534-ab5113a2f7d7
       AsyncResult bc3db3cd-49dd-4595-93ae-6c9e5e93d8fe
--- restored ---
 AsyncResult 52ad6806-acb9-4791-94c4-bdd88bb645b5
   GroupResult 67e833a9-6a81-4953-83f0-669f11ad55c3

     AsyncResult 3e561eed-1bf4-435f-8250-ebb14f022a10
     AsyncResult d8b60ebc-56da-43f5-87e3-2e5af0ed14af
     GroupResult a03c5631-a109-4051-908e-9455e83a8bba
       AsyncResult 11de71fb-30ec-44e7-9fd6-6d186678d316
       AsyncResult 75d720de-eaed-48fb-a534-ab5113a2f7d7
       AsyncResult bc3db3cd-49dd-4595-93ae-6c9e5e93d8fe

--- original ---
 AsyncResult f33044b9-a51f-4500-baf9-02bab0587190
   GroupResult 9702893a-2a0f-450c-9175-e2320e07a372
     AsyncResult 3f19266c-e267-4d90-b057-5da71347aaaa
     AsyncResult ce61f0f8-6d89-411f-b167-a777a6de4a84
     GroupResult e001e5bf-aeb8-4278-a787-2e978677ce63
       AsyncResult 6b612777-873e-4fa4-a0d4-d3ebc045187b
       AsyncResult 1967abb4-354a-44d8-be0c-e43ac0328521
       AsyncResult f7c9e8e7-5eec-4cf5-a72d-6adf2dc9a389
--- restored ---
 AsyncResult f33044b9-a51f-4500-baf9-02bab0587190
   GroupResult 9702893a-2a0f-450c-9175-e2320e07a372

     AsyncResult 3f19266c-e267-4d90-b057-5da71347aaaa
     AsyncResult ce61f0f8-6d89-411f-b167-a777a6de4a84
     GroupResult e001e5bf-aeb8-4278-a787-2e978677ce63
       AsyncResult 6b612777-873e-4fa4-a0d4-d3ebc045187b
       AsyncResult 1967abb4-354a-44d8-be0c-e43ac0328521
       AsyncResult f7c9e8e7-5eec-4cf5-a72d-6adf2dc9a389

t_g_g_t OK
t_g_g OK
Traceback (most recent call last):
  File "/tmp/repro3.py", line 31, in <module>
    res = mk().apply_async()
          ^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 2182, in apply_async
    return self.run(tasks, body, args, task_id=task_id, kwargs=kwargs, **merged_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 2282, in run
    header_result = header.apply_async(partial_args, kwargs, task_id=group_id, **options)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/repo/celery/canvas.py", line 1638, in apply_async
    results = list(self._apply_tasks(tasks, producer, app, p,
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1817, in _apply_tasks
    sig.apply_async(produce
…[1194 chars omitted]…
line 1817, in _apply_tasks
    sig.apply_async(producer=producer, add_to_parent=False,
  File "/repo/celery/canvas.py", line 402, in apply_async
    return _apply(args, kwargs, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/app/task.py", line 668, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: add() missing 1 required positional argument: 'y'

t_g_g_t OK
t_g_g OK
g_g_t OK
g_g OK
t_g_t_g_t OK
t_g_g_t OK
t_g_g OK
g_g_t OK
g_g OK
t_g_t_g_t OK

t/unit/tasks/test_chord.py:92:                         GroupResult=GroupResult, result_from_tuple=Mock())
__init__.py
app
apps
backends
bin
concurrency
conftest.py
contrib
events
fixups
security

tasks
test_canvas.py
test_generics.py
test_loops.py
utils
worker

AsyncResult 7131cb62-ab79-423b-ab93-82be3ae5661c children= []
   GroupResult 85e06d5c-65f5-4327-a45b-e794ab3c155c children= ['df387028-d88f-4066-89cb-25b93ee34b03', 'fd876210-30af-4f08-b5f6-f4b4eb76e6ac']
     GroupResult 1c1a5c59-f0a8-4142-af5f-50ae6f6d82f1 children= ['2bbf0e59-dfcd-4bbf-a552-19709edbfbab', '66a022c6-ea7d-42c9-b9b3-3a3f5e7a39d7']
       AsyncResult 5658bfe1-9c22-4893-bd14-7538d04b5b5d children= []

# XXX: Without verifying `cls` is actually an exception class,
        #      an attacker could execute arbitrary python code.
        #      cls could be anything, even eval().
        try:
            if isinstance(exc_msg, (tuple, list)):
                exc = cls(*exc_msg)
            else:
                exc = cls(exc_msg)
        except Exception as err:  # noqa
            exc = Exception(f'{cls}({exc_msg})')

        return exc

    def prepare_value(self, result):
        """Prepare value for storage."""
        if self.serializer != 'pickle' and isinstance(result, ResultBase):
            return result.as_tuple()
        return result

    def encode(self, data):
        _, _, payload = self._encode(data)
        return payload

    def _encode(self, data):
        return dumps(data, serializer=self.serializer)

    def meta_from_decoded(self, meta):
        if meta
…[1717 chars omitted]…
r getattr(get_current_task(), 'request', None)
        if request:
            return [r.as_tuple() for r in getattr(request, 'children', [])]

    def __reduce__(self, args=(), kwargs=None):
        kwargs = {} if not kwargs else kwargs
        return (unpickle_backend, (self.__class__, args, kwargs))


class SyncBackendMixin:
    def iter_native(self, result, timeout=None, interval=0.5, no_ack=True,

__version__ = '5.6.2'
VERSION_BANNER = f'{__version__} ({SERIES})'
.. _changelog:

================
 Change history
================

This document contains change notes for bugfix & new features
in the main branch & 5.6.x series, please see :ref:`whatsnew-5.6` for
an overview of what's new in Celery 5.6.

.. _version-5.6.2:

5.6.2
=====

:release-date: 2026-01-04
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix recursive WorkController instantiation in DjangoWorkerFixup + AttributeError when pool_cls is a string (#10045)
- Bugfix: Revoked tasks now immediately update backend status to REVOKED (#9869)
- Prepare for release: v5.6.2 (#10049)

.. _version-5.6.1:

5.6.1
=====

:release-date: 2025-12-29
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix Redis Sentinel ACL authentication support (#10013)
- Fix: Broker heartbeats not sent during graceful shutdown (#9986)
- docs #5410 -- Document confirm_publish broker transport option (#10016)
- close DB pools only in prefork mode (#10020)

ERROR: Could not find a version that satisfies the requirement celery==5.6.3 (from versions: none)
ERROR: No matching distribution found for celery==5.6.3

WARNING: Retrying (Retry(total=1, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NameResolutionError("HTTPSConnection(host='pypi.org', port=443): Failed to resolve 'pypi.org' ([Errno -3] Temporary failure in name resolution)")': /simple/celery/
WARNING: Retrying (Retry(total=0, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NameResolutionError("HTTPSConnection(host='pypi.org', port=443): Failed to resolve 'pypi.org' ([Errno -3] Temporary failure in name resolution)")': /simple/celery/
ERROR: No matching distribution found for celery
ERROR: Could not find a version that satisfies the requirement celery (from versions: none)
ERROR: No matching distribution found for celery

1 TGT OK
1 TGGT OK
1 TGG OK
1 GGT OK
1 GG OK
1 TGGGT OK
1 TGGT2 OK
2 TGT OK
2 TGGT OK
2 TGG OK
2 GGT OK
2 GG OK

2 TGGGT OK
2 TGGT2 OK

x.backend.wait_for_pending.return_value = 84
        x.parent = EagerResult(uuid(), KeyError('foo'), states.FAILURE)
        with pytest.raises(KeyError):
            x.get(propagate=True)
        x.backend.wait_for_pending.assert_not_called()

        x.parent = EagerResult(uuid(), 42, states.SUCCESS)
        assert x.get(propagate=True) == 84
        x.backend.wait_for_pending.assert_called()

    def test_get_children(self):
        tid = uuid()

        x = self.app.AsyncResult(tid)
        child = [self.app.AsyncResult(uuid()).as_tuple()
                 for i in range(10)]
        x._cache = {'children': child}
        assert x.children
        assert len(x.children) == 10

        x._cache = {'status': states.SUCCESS}
        x.backend._cache[tid] = {'result': None}
        assert x.children is None

    def test_build_graph_get_leaf_collect(self):
        x = self.app.A
…[2054 chars omitted]…
ok_res = self.app.AsyncResult(self.task1['id'])
        nok_res = self.app.AsyncResult(self.task3['id'])
        nok_res2 = self.app.AsyncResult(self.task4['id'])

        assert ok_res.successful()
        assert not nok_res.successful()
        assert not nok_res2.successful()

        pending_res = self.app.AsyncResult(uuid())
        assert not pending_res.successful()

    def test_raising(self):

595:            'children': self.current_task_children(request),
716:    def get_children(self, task_id):
719:            return self.get_task_meta(task_id)['children']
869:    def current_task_children(self, request=None):
872:            return [r.as_tuple() for r in getattr(request, 'children', [])]
celery/app/trace.py:537:                            _children = _meta.get('children')
celery/app/trace.py:540:                            if (_callbacks or _chain) and not _children:
celery/app/trace.py:613:                        # stored, so that result.children is populated.
celery/app/task.py:78:    _children = None   # see property
celery/app/task.py:198:    def children(self):
celery/app/task.py:199:        # children must be an empty list for every thread
celery/app/task.py:200:        if self._children is None:

celery/app/task.py:201:            self._children = []
celery/app/task.py:202:        return self._children
celery/app/task.py:259:    #: (``result.children``).

if state == SUCCESS:
                        info(LOG_IGNORED, {
                            'id': task_request.id,
                            'name': get_task_name(task_request, name),
                            'description': 'Task already completed successfully.'
                        })
                        _root_id = task_request.root_id or uuid
                        _priority = task_request.delivery_info.get('priority') if \
                            inherit_parent_priority else None
                        try:
                            _meta = r._get_task_meta()
                            stored_retval = _meta.get('result')

                            # Children are populated by mark_as_done on the
                            # original execution.  If present, callbacks were
                            # already dispatched — skip to avoid dupl
…[4661 chars omitted]…

                        )
                        task.backend.mark_as_done(
                            uuid, retval, task_request, publish_result,
                        )
                    except EncodeError as exc:
                        I, R, state, retval = on_error(task_request, exc)
                        # MEMORY LEAK FIX: Clear traceback frames to prevent memory retention (Issue #8882)

if app is None:
            try:
                tasks = self.tasks.tasks  # is a group
            except AttributeError:
                tasks = self.tasks
            if tasks:
                app = tasks[0]._app
            if app is None and body is not None:
                app = body._app
        return app if app is not None else current_app

    tasks = getitem_property('kwargs.header', 'Tasks in chord header.')

    body = getitem_property('kwargs.body', 'Body task of chord.')


# Add a back-compat alias for the previous `chord` class name which conflicts
# with keyword arguments elsewhere in this file
chord = _chord


def signature(varies, *args, **kwargs):
    """Create new signature.

    - if the first argument is a signature already then it's cloned.
    - if the first argument is a dict, then a Signature version is returned.

    Returns:
        Signature: The 
…[709 chars omitted]…
oned when this flag is enabled.

    Returns:
        Optional[abstract.CallableSignature]
    """
    if d is not None:
        if isinstance(d, abstract.CallableSignature):
            if clone:
                d = d.clone()
        elif isinstance(d, dict):
            d = signature(d)

        if app is not None:
            d._app = app
    return d


maybe_subtask = maybe_signature  # XXX compat

import builtins
import inspect
import io
import logging
import os
import platform
import sys
import threading
import types
import warnings
from contextlib import contextmanager
from functools import wraps

from importlib import import_module, reload
from unittest.mock import MagicMock, Mock, patch

import pytest
from kombu import Queue

from celery.backends.cache import CacheBackend, DummyClient
# we have to import the pytest plugin fixtures here,
# in case user did not do the `python setup.py develop` yet,
# that installs the pytest plugin into the setuptools registry.
from celery.contrib.pytest import celery_app, celery_enable_logging, celery_parameters, depends_on_current_app
from celery.contrib.testing.app import TestApp, Trap
from celery.contrib.testing.mocks import TaskMessage, TaskMessage1, task_message_from_sig

# Tricks flake8 into silencing redefining fixtures warnings.
__all
…[437 chars omitted]…
SIO_init = io.StringIO.__init__

SENTINEL = object()


def noop(*args, **kwargs):
    pass


class WhateverIO(io.StringIO):

    def __init__(self, v=None, *a, **kw):
        _SIO_init(self, v.decode() if isinstance(v, bytes) else v, *a, **kw)

    def write(self, data):
        _SIO_write(self, data.decode() if isinstance(data, bytes) else data)


@pytest.fixture(scope='session')
def celery_config():

Traceback (most recent call last):
  File "/tmp/brute2.py", line 29, in <module>
    res = mk().apply()
          ^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1300, in apply
    res = task.clone(fargs, fkwargs).apply(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 2192, in apply
    args=(tasks.apply(args, kwargs).get(propagate=propagate),),
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 735, in get
    return (self.join_native if self.supports_native_join else self.join)(

           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    
…[741 chars omitted]…
GroupResult' object has no attribute 'state'. Did you mean: 'save'?
Traceback (most recent call last):
  File "/tmp/brute2.py", line 29, in <module>
    res = mk().apply()
          ^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1303, in apply
    if res.state in (states.IGNORED, states.REJECTED):
       ^^^^^^^^^
AttributeError: 'GroupResult' object has no attribute 'state'. Did you mean: 'save'?

Saved working directory and index state WIP on master: 0d5eed3 base
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: add() takes 2 positional arguments but 3 were given
On branch master
Changes not staged for commit:
  (use "git add <file>..." to update what will be committed)
  (use "git restore <file>..." to discard changes in working directory)
	modified:   celery/canvas.py

no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (fa8a6c0921c64179674695c12c968e71774dc18d)

value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: add() takes 2 positional arguments but 3 were given
PromiseProxy IGNORED None
PromiseProxy REJECTED None

Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):

Saved working directory and index state WIP on master: 0d5eed3 base
4
On branch master
Changes not staged for commit:
  (use "git add <file>..." to update what will be committed)
  (use "git restore <file>..." to discard changes in working directory)
	modified:   celery/canvas.py

no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (7f29cf8b5efb70de9dc0f45233faff76491cdc1b)

sed: -e expression #1, char 9: missing command

tasks = getattr(self.tasks, "tasks", self.tasks)
        return sum(self._descend(task) for task in tasks)

    def run(self, header, body, partial_args, app=None, interval=None,
            countdown=1, max_retries=None, eager=False,
            task_id=None, kwargs=None, **options):
        """Execute the chord.

        Executing the chord means executing the header and sending the
        result to the body. In case of an empty header, the body is
        executed immediately.

        Arguments:
            header (group): The header to execute.
            body (Signature): The body to execute.
            partial_args (tuple): Arguments to pass to the header.
            app (Celery): The Celery app instance.
            interval (float): The interval between retries.
            countdown (int): The countdown between retries.
            max_retries (int): The maximum 
…[3520 chars omitted]…
the new behavior by setting task_allow_error_cb_on_chord_header to True "
                "and report any concerns you might have in our issue tracker before we make a final decision "
                "regarding how errbacks should behave when used with chords.",
                CPendingDeprecationWarning,
                stacklevel=2,
            )

        # Edge case for nested chords in the header

# the callback may be executed before all children in the group are
        # completed and also if any children other than the first one fail.
        #
        # The callback signature is cloned and made immutable since it the
        # first task isn't actually capable of passing the return values of its
        # siblings to the callback task.
        sig = sig.clone().set(immutable=True)
        return self.tasks[0].link(sig)

    def link_error(self, sig):
        # Any child task might error so we need to ensure that they are all
        # capable of calling the linked error signature. This opens the

        # possibility that the task is called more than once but that's better
        # than it not being called at all.
        #
        # We return a concretised tuple of the signatures actually applied to
        # each child task signature, of which there might be non
…[5957 chars omitted]…
n result.

    def _freeze_gid(self, options):
        """Freeze the group id by the existing task_id or a new UUID."""
        # remove task_id and use that as the group_id,
        # if we don't remove it then every task will have the same id...
        options = {**self.options, **{
            k: v for k, v in options.items()
            if k not in self._IMMUTABLE_OPTIONS or k not in self.options

1978:class _chord(Signature):
                "immutable": False
            }
            >>> chord_sig = chord.from_dict(chord_dict)

        Iterates over the given tasks in the dictionary and convert them to signatures.
        Chord header needs to be defined in d['kwargs']['header'] as a sequence
        of tasks.
        Chord body needs to be defined in d['kwargs']['body'] as a single task.

        The tasks themselves can be dictionaries or signatures (or both).
        """

        options = d.copy()
        args, options['kwargs'] = cls._unpack_args(**options['kwargs'])
        return cls(*args, app=app, **options)

    @staticmethod
    def _unpack_args(header=None, body=None, **kwargs):
        # Python signatures are better at extracting keys from dicts
        # than manually popping things off.
        return (header, body), kwargs

    def __init__(self, header, body=
…[3368 chars omitted]…

            visitor.on_chord_header_end(self, **headers)

        if visitor is not None and self.body is not None:
            visitor_headers = visitor.on_chord_body(self, **headers) or {}
            headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
            self.body.stamp(visitor, append_stamps, **headers)

    def apply_async(self, args=None, kwargs=None, task_id=None,

0d5eed3 base
celery               5.6.2       /repo
/repo/celery/__init__.py

44:def maybe_unroll_group(group):
45-    """Unroll group with only one member.
46-    This allows treating a group of a single task as if it
47-    was a single task without pre-knowledge."""
48-    # Issue #1656
49-    try:
50-        size = len(group.tasks)
51-    except TypeError:
52-        try:
53-            size = group.tasks.__length_hint__()
54-        except (AttributeError, TypeError):
55-            return group

56-        else:
57-            return list(group.tasks)[0] if size == 1 else group
58-    else:
59-        return group.tasks[0] if size == 1 else group
44:def maybe_unroll_group(group):
775:            other = maybe_unroll_group(other)
964:            other = maybe_unroll_group(other)
1193:                task = maybe_unroll_group(task)
1879:            tasks1, tasks2 = itertools.tee(self._unroll_tasks(self.tasks))
1888:            results = list(self._freeze_unroll(
1915:    def _unroll_tasks(self, tasks):
1920:    def _freeze_unroll(self, new_tasks, group_id, chord, root_id, parent_id):
2094:            other = maybe_unroll_group(other)

1 TTGGTT OK
1 TG_GG_T OK
1 TGGTG OK
1 GTGGT OK
1 nested OK
1 chordbody OK
2 TTGGTT OK
2 TG_GG_T OK
2 TGGTG OK
2 GTGGT OK
2 nested OK
2 chordbody OK

{
 "task": "celery.chain",
 "args": [],
 "kwargs": {
  "tasks": [
   {
    "task": "t.add",
    "args": [
     1,
     2
    ],
    "kwargs": {},

    "options": {},
    "subtask_type": null,
    "immutable": false
   },
   {
    "task": "celery.group",
    "args": [],
    "kwargs": {
     "tasks": [
      {
       "task": "t.add",
       "args": [
        0,
        0
       ],
       "kwargs": {},
       "options": {},
       "subtask_type": null,
       "immutable": false
      },
      {
       "task": "t.add",
       "args": [
        1,
        1
       ],
       "kwargs": {},
       "options": {},
       "subtask_type": null,
       "immutable": false
      }
     ]
    },
    "options": {},
    "subtask_type": "group",
    "immutable": false
   },
   {
    "task": "celery.chord",
  
OK
 O (0, 'AsyncResult', '86db0f65-b948-426d-9283-18afcc44e5f1')
 O (1, 'GroupResult', '036a40cc
…[494 chars omitted]…
cc-fc78-4eaf-b51d-57085e92dc61')
 R (2, 'AsyncResult', '24482cc8-139b-418b-8890-23447a28dc06')
 R (2, 'AsyncResult', '7d26a8ee-3a7a-44a3-b200-c2ba93270763')
 R (2, 'GroupResult', '80a81dfa-c360-45e6-afeb-27d387b67522')
 R (3, 'AsyncResult', '2afd06e3-8a91-4777-9bb1-650ce916f50b')
 R (3, 'AsyncResult', '764f9587-3a13-4b06-a293-47b8d18ad1dc')
 R (3, 'AsyncResult', '9401a30f-f9f3-41ed-a575-efb504f96e1b')

426:    def _dispatch_callbacks_and_chain(
427-        retval, callbacks, chain, parent_id, root_id, priority,
428-    ):
429-        """Dispatch callbacks and chain for a completed task.
430-
431-        Dispatches link callbacks and then the next chain step.
432-        Does NOT fire task lifecycle signals (on_success, task_postrun)
433-        or call mark_as_done — callers handle those separately.
434-
435-        Note: dispatch is not atomic.  If callbacks succeed but the
436-        chain step fails (or vice-versa), a Reject + redeliver may
437-        re-dispatch the already-sent callbacks.  This is acceptable

438-        under Celery's at-least-once delivery model.
439-        """
440-        if callbacks:
441-            if len(callbacks) > 1:
442-                sigs, groups = [], []
443-                for sig in callbacks:
444-                    sig = signature(sig, app=a
…[1668 chars omitted]…
     # dead-letter / max-delivery-count policies are
553-                            # the intended circuit-breaker.
554-                            logger.error(
555-                                'Failed to dispatch chain/callbacks for '
556-                                'deduplicated task %s',
557-                                task_request.id,
558-                                exc_info=True,

474:    def freeze(self, _id=None, group_id=None, chord=None,
475-               root_id=None, parent_id=None, group_index=None):
476-        """Finalize the signature by adding a concrete task id.
477-
478-        The task won't be called and you shouldn't call the signature
479-        twice after freezing it as that'll result in two task messages
480-        using the same task id.
481-
482-        The arguments are used to override the signature's headers during
483-        freezing.
484-
485-        Arguments:

486-            _id (str): Task id to use if it didn't already have one.
487-                New UUID is generated if not provided.
488-            group_id (str): Group id to use if it didn't already have one.
489-            chord (Signature): Chord body when freezing a chord header.
490-            root_id (str): Root id to use.
491-            parent_id (str): Parent id
…[553 chars omitted]…

1101-        _, results = self._frozen = self.prepare_steps(
1102-            self.args, self.kwargs, self.tasks, root_id, parent_id, None,
1103-            self.app, _id, group_id, chord, clone=False,
1104-            group_index=group_index,
1105-        )
1106-        return results[0]
1107-
1108-    def stamp(self, visitor=None, append_stamps=False, **headers):
1109-        visitor_headers = None

1 plain OK
1 linkerr OK
1 link OK
1 unrolled OK
2 plain OK
2 linkerr OK
2 link OK
2 unrolled OK

top AsyncResult cdd0f82c children= []
   parent GroupResult 907b9de9 children= ['a1242e11', '7654ad60']
     child AsyncResult a1242e11 children= []
     child AsyncResult 7654ad60 children= []
     parent GroupResult 7c5058d5 children= ['ac93abc7', 'b17fd677']
       child AsyncResult ac93abc7 children= []
       child AsyncResult b17fd677 children= []
       parent AsyncResult 049a2a3d children= []

File "/repo/celery/canvas.py", line 1301, in apply
    last and (last.get(),), **dict(self.options, **options))
              ^^^^^^^^^^
  File "/repo/celery/result.py", line 735, in get
    return (self.join_native if self.supports_native_join else self.join)(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task

    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: add() takes 2 positional arguments but 3 were given
Traceback (most recent call last):
  File "/tmp/brute6.py", line 27, in <module>
    res = mk().apply_async()
          ^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", 
…[620 chars omitted]…
^^^^^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 816, in join
    value = result.get(
            ^^^^^^^^^^^
  File "/repo/celery/result.py", line 1068, in get
    raise self.result if isinstance(
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
TypeError: add() takes 2 positional arguments but 3 were given

tuple: A tuple of the group id, and the AsyncResult of each of the group tasks.
        """
        # pylint: disable=redefined-outer-name
        #  XXX chord is also a class in outer scope.
        opts = self.options
        try:
            gid = opts['task_id']
        except KeyError:
            gid = opts['task_id'] = group_id or uuid()
        if group_id:
            opts['group_id'] = group_id
        if chord:

            opts['chord'] = chord
        if group_index is not None:
            opts['group_index'] = group_index
        root_id = opts.setdefault('root_id', root_id)
        parent_id = opts.setdefault('parent_id', parent_id)
        if isinstance(self.tasks, _regen):
            # When the group tasks are a generator, we need to make sure we don't
            # exhaust it during the freeze process. We use two generators to do this.
            # One 
…[3674 chars omitted]…
roup_id=group_id,
                                  chord=chord, root_id=root_id,
                                  parent_id=parent_id,
                                  group_index=group_index)
                group_index += 1

    def __repr__(self):
        if self.tasks:
            return remove_repeating_from_task(
                self.tasks[0]['task'],
                f'group({self.tasks!r})')

1 plain OK
1 taskid OK
1 args OK
1 call OK
1 delay OK
2 plain OK
2 taskid OK
2 args OK
2 call OK
2 delay OK

task.stamp(visitor, append_stamps, **stamps)

        if visitor is not None:
            visitor.on_chord_header_end(self, **headers)

        if visitor is not None and self.body is not None:
            visitor_headers = visitor.on_chord_body(self, **headers) or {}
            headers = self._stamp_headers(visitor_headers, append_stamps, **headers)
            self.body.stamp(visitor, append_stamps, **headers)

    def apply_async(self, args=None, kwargs=None, task_id=None,
                    producer=None, publisher=None, connection=None,

                    router=None, result_cls=None, **options):
        args = args if args else ()
        kwargs = kwargs if kwargs else {}
        args = (tuple(args) + tuple(self.args)
                if args and not self.immutable else self.args)
        body = kwargs.pop('body', None) or self.kwargs['body']
        kwargs = d
…[986 chars omitted]…
     body = self.body if body is None else body
        tasks = (self.tasks.clone() if isinstance(self.tasks, group)
                 else group(self.tasks, app=self.app))
        return body.apply(
            args=(tasks.apply(args, kwargs).get(propagate=propagate),),
        )

    @classmethod
    def _descend(cls, sig_obj):
        """Count the number of tasks in the given signature recursively.

878:def _prepare_chain_from_options(options, tasks, use_link):
879-    # When we publish groups we reuse the same options dictionary for all of
880-    # the tasks in the group. See:
881-    # https://github.com/celery/celery/blob/fb37cb0b8/celery/canvas.py#L1022.
882-    # Issue #5354 reported that the following type of canvases
883-    # causes a Celery worker to hang:
884-    # group(
885-    #   add.s(1, 1),
886-    #   add.s(1, 1)
887-    # ) | tsum.s() | add.s(1) | group(add.s(1), add.s(1))
888-    # The resolution of #5354 in PR #5681 was to only set the `chain` key
889-    # in the options dictionary if it is not present.

890-    # Otherwise we extend the existing list of tasks in the chain with the new
891-    # tasks: options['chain'].extend(chain_).
892-    # Before PR #5681 we overrode the `chain` key in each iteration
893-    # of the loop which applies all the tasks in t
…[499 chars omitted]…
5-            # If we only have a single task, it may be important that we pass
1086-            # the real result object rather than the one obtained via freezing.
1087-            # e.g. For `GroupResult`s, we need to pass back the result object
1088-            # which will actually have its promise fulfilled by the subtasks,
1089-            # something that will never occur for the frozen result.

1 same_group_obj OK
1 same_sig OK
2 same_group_obj OK
2 same_sig OK

789:    def get_group_meta(self, group_id, cache=True):
790-        self._ensure_not_eager()
791-        if cache:
792-            try:
793-                return self._cache[group_id]
794-            except KeyError:
795-                pass
796-
797-        meta = self._ensure_retryable(self._restore_group, group_id=group_id)
798-        if cache and meta is not None:
799-            self._cache[group_id] = meta
800-        return meta

801-
802:    def restore_group(self, group_id, cache=True):
803-        """Get the result for a group."""
804-        meta = self.get_group_meta(group_id, cache=cache)
805-        if meta:
806-            return meta['result']
807-
808:    def save_group(self, group_id, result):
809-        """Store the result of an executed group."""
810-        return self._ensure_retryable(
811-            self._save_group,
812-            group_id=group_id,
813-  
…[1077 chars omitted]…
t later this
1175-        # was extended to support other serializers, so the
1176-        # structure is kind of weird.
1177-        if meta:
1178-            meta = self.decode(meta)
1179-            result = meta['result']
1180-            meta['result'] = result_from_tuple(result, self.app)
1181-            return meta
1182-
1183-    def _apply_chord_incr(self, header_result_args, body, **kwargs):

celery/canvas.py:989:                # it leads to a situation where two consecutive chords are formed.
docs/history/changelog-5.4.rst:128:- Eliminate consecutive chords generated by group | task upgrade (#8663)
docs/userguide/calling.rst:371:    On each consecutive retry this number will be added to the retry

def __or__(self, other):
        if isinstance(other, group):
            # unroll group with one member
            other = maybe_unroll_group(other)
            if not isinstance(other, group):
                return self.__or__(other)
            # chain | group() -> chain
            tasks = self.unchain_tasks()
            if not tasks:
                # If the chain is empty, return the group
                return other
            if isinstance(tasks[-1], chord):

                # CHAIN [last item is chord] | GROUP -> chain with chord body.
                tasks[-1].body = tasks[-1].body | other
                return type(self)(tasks, app=self.app)
            # use type(self) for _chain subclasses
            return type(self)(seq_concat_item(
                tasks, other), app=self._app)
        elif isinstance(other, _chain):
            # chain | chain -> chain
     
…[834 chars omitted]…
s[-2].body | sig.tasks[-1]
                    sig.tasks = sig.tasks[:-1]
                return sig
            elif self.tasks and isinstance(self.tasks[-1], chord) and not isinstance(other, chord):
                # CHAIN [last item is chord] | TASK -> chain with chord body.
                sig = self.clone()
                sig.tasks[-1].body = sig.tasks[-1].body | other
                return sig

Traceback (most recent call last):
  File "/tmp/worker_test.py", line 23, in <module>
    with start_worker(app, perform_pending_operations=True, loglevel='error'):
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 137, in __enter__
    return next(self.gen)
           ^^^^^^^^^^^^^^
  File "/repo/celery/contrib/testing/worker.py", line 121, in start_worker
    with _start_worker_thread(app,
         ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 137, in __enter__
    return next(self.gen)

           ^^^^^^^^^^^^^^
  File "/repo/celery/contrib/testing/worker.py", line 156, in _start_worker_thread
    assert 'celery.ping' in app.tasks
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError

File "/repo/celery/local.py", line 95, in __class__
    return self._get_class()
           ^^^^^^^^^^^^^^^^^
  File "/repo/celery/local.py", line 91, in _get_class
    return self._get_current_object().__class__
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/local.py", line 95, in __class__
    return self._get_class()
           ^^^^^^^^^^^^^^^^^
  File "/repo/celery/local.py", line 91, in _get_class
    return self._get_current_object().__class__
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/repo/celery/local.py", line 95, in __class__
    return self._get_class()
           ^^^^^^^^^^^^^^^^^
  File "/repo/celery/local.py", line 91, in _get_class
    return self._get_current_object().__class__
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/local.py", line 95, in __class__
    return self._get_class()
           ^^^^^^^^^^^^^^
…[497 chars omitted]…
rgs, **self.__kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/app/__init__.py", line 68, in task_by_cons
    return app.tasks[
           ^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/kombu/utils/objects.py", line 40, in __get__
    return super().__get__(instance, owner)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
RecursionError: maximum recursion depth exceeded

File "/repo/celery/app/trace.py", line 619, in trace_task
    _dispatch_callbacks_and_chain(
  File "/repo/celery/app/trace.py", line 469, in _dispatch_callbacks_and_chain
    _chsig.apply_async(
  File "/repo/celery/canvas.py", line 2184, in apply_async
    return self.run(tasks, body, args, task_id=task_id, kwargs=kwargs, **merged_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 2284, in run
    header_result = header.apply_async(partial_args, kwargs, task_id=group_id, **options)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1640, in apply_async
    results = list(self._apply_tasks(tasks, producer, app, p,

              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/celery/canvas.py", line 1819, in _
…[981 chars omitted]…
self._to_remote_traceback(tb))
  File "/repo/celery/result.py", line 386, in throw
    self.on_ready.throw(*args, **kwargs)
  File "/usr/local/lib/python3.12/site-packages/vine/promises.py", line 235, in throw
    reraise(type(exc), exc, tb)
  File "/usr/local/lib/python3.12/site-packages/vine/utils.py", line 27, in reraise
    raise value
TypeError: add() takes 2 positional arguments but 3 were given

~~^~~
TypeError: can only concatenate list (not "int") to list

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/repo/celery/backends/base.py", line 1229, in on_chord_part_return
    ret = j(
          ^^
  File "/repo/celery/result.py", line 878, in join_native
    raise value
TypeError: can only concatenate list (not "int") to list

[2026-10-02 00:00:28,860: ERROR/MainProcess] Task t.add[46f27ffd-58df-4a06-8897-4620b14ed374] raised unexpected: TypeError('can only concatenate list (not "int") to list')
Traceback (most recent call last):
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
  File "/tmp/worker_test.py", line 11, in add
    def add(x, y): return x + y
                          ~~^~~
TypeError: can onl
…[654 chars omitted]…
tb))
  File "/repo/celery/result.py", line 386, in throw
    self.on_ready.throw(*args, **kwargs)
  File "/usr/local/lib/python3.12/site-packages/vine/promises.py", line 235, in throw
    reraise(type(exc), exc, tb)
  File "/usr/local/lib/python3.12/site-packages/vine/utils.py", line 27, in reraise
    raise value
celery.exceptions.ChordError: TypeError('can only concatenate list (not "int") to list')

Traceback (most recent call last):
  File "/tmp/worker_test.py", line 26, in <module>
    with start_worker(app, perform_pending_operations=True, loglevel='error'):
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 137, in __enter__
    return next(self.gen)
           ^^^^^^^^^^^^^^
  File "/repo/celery/contrib/testing/worker.py", line 121, in start_worker
    with _start_worker_thread(app,
         ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 137, in __enter__
    return next(self.gen)

           ^^^^^^^^^^^^^^
  File "/repo/celery/contrib/testing/worker.py", line 156, in _start_worker_thread
    assert 'celery.ping' in app.tasks
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError

~~^~~
TypeError: can only concatenate list (not "int") to list

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/repo/celery/backends/base.py", line 1229, in on_chord_part_return
    ret = j(
          ^^
  File "/repo/celery/result.py", line 878, in join_native
    raise value
TypeError: can only concatenate list (not "int") to list

[2026-10-02 00:00:56,270: ERROR/MainProcess] Task t.inc[ba388840-6b6f-4640-a46b-c2788b3b870e] raised unexpected: TypeError('can only concatenate list (not "int") to list')
Traceback (most recent call last):
  File "/repo/celery/app/trace.py", line 585, in trace_task
    R = retval = fun(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^
  File "/tmp/worker_test.py", line 11, in inc
    def inc(x): return x + 1
                       ~~^~~
TypeError: can only concate
…[645 chars omitted]…
tb))
  File "/repo/celery/result.py", line 386, in throw
    self.on_ready.throw(*args, **kwargs)
  File "/usr/local/lib/python3.12/site-packages/vine/promises.py", line 235, in throw
    reraise(type(exc), exc, tb)
  File "/usr/local/lib/python3.12/site-packages/vine/utils.py", line 27, in reraise
    raise value
celery.exceptions.ChordError: TypeError('can only concatenate list (not "int") to list')

value 10
--- client structure ---
  (0, 'AsyncResult', 'c072e778')
  (1, 'GroupResult', '54ea6197')
  (2, 'AsyncResult', 'db64269b')
  (2, 'AsyncResult', '95cb2667')
  (2, 'GroupResult', '1826b9c7')
  (3, 'AsyncResult', '8527ddac')
  (3, 'AsyncResult', '8b812874')
  (3, 'AsyncResult', 'cee9c291')
roundtrip OK
  R (0, 'AsyncResult', 'c072e778')

  R (1, 'GroupResult', '54ea6197')
  R (2, 'AsyncResult', 'db64269b')
  R (2, 'AsyncResult', '95cb2667')
  R (2, 'GroupResult', '1826b9c7')
  R (3, 'AsyncResult', '8527ddac')
  R (3, 'AsyncResult', '8b812874')
  R (3, 'AsyncResult', 'cee9c291')
--- backend fetched ---
  (0, 'AsyncResult', 'c072e778')
roundtrip2 OK
  R (0, 'AsyncResult', 'c072e778')

value 10
--- client structure ---
  (0, 'AsyncResult', 'ffd14a25')
  (1, 'GroupResult', 'f146c764')
  (2, 'AsyncResult', 'c2df4cd2')
  (2, 'AsyncResult', '2d7f49bf')
  (2, 'GroupResult', '6c094d6a')
  (3, 'AsyncResult', '348eb905')
  (3, 'AsyncResult', '9e1926eb')
  (3, 'AsyncResult', '1ec863b5')
roundtrip OK
  R (0, 'AsyncResult', 'ffd14a25')

  R (1, 'GroupResult', 'f146c764')
  R (2, 'AsyncResult', 'c2df4cd2')
  R (2, 'AsyncResult', '2d7f49bf')
  R (2, 'GroupResult', '6c094d6a')
  R (3, 'AsyncResult', '348eb905')
  R (3, 'AsyncResult', '9e1926eb')
  R (3, 'AsyncResult', '1ec863b5')
--- backend fetched ---
  (0, 'AsyncResult', 'ffd14a25')
roundtrip2 OK
  R (0, 'AsyncResult', 'ffd14a25')

47:    @app.task(name='celery.chord_unlock', max_retries=None, shared=False,
48-              default_retry_delay=app.conf.result_chord_retry_interval, ignore_result=True, lazy=False, bind=True)
49-    def unlock_chord(self, group_id, callback, interval=None,
50-                     max_retries=None, result=None,
51-                     Result=app.AsyncResult, GroupResult=app.GroupResult,
52-                     result_from_tuple=result_from_tuple, **kwargs):
53-        if interval is None:
54-            interval = self.default_retry_delay
55-
56-        # check if the task group is ready, and if so apply the callback.
57-        callback = maybe_signature(callback, app)
58-        deps = GroupResult(

59-            group_id,
60-            [result_from_tuple(r, app=app) for r in result],
61-            app=app,
62-        )
63-        j = deps.join_native if deps.supports_native_joi
…[522 chars omitted]…
      timeout=app.conf.result_chord_join_timeout,
80-                    propagate=True,
81-                )
82-        except Exception as exc:  # pylint: disable=broad-except
83-            try:
84-                culprit = next(deps._failed_join_report())
85-                reason = f'Dependency {culprit.id} raised {exc!r}'
86-            except StopIteration:
87-                reason = repr(exc)

def _maybe_set_cache(self, meta):
        if meta:
            state = meta['status']
            if state in states.READY_STATES:
                d = self._set_cache(self.backend.meta_from_decoded(meta))
                self.on_ready(self)
                return d
        return meta

    def _get_task_meta(self):
        if self._cache is None:
            return self._maybe_set_cache(self.backend.get_task_meta(self.id))

        return self._cache

    def _iter_meta(self, **kwargs):
        return iter([self._get_task_meta()])

    def _set_cache(self, d):
        children = d.get('children')
        if children:
            d['children'] = [
                result_from_tuple(child, self.app) for child in children
            ]
        self._cache = d
        return d

    @property
    def result(self):
        """Task return value.

        Note:
            When the task has been executed, this contains the return value.
            If the task raised an exception, this will be the exception
            instance.
        """
        return self._get_task_meta()['result']
    info = result

    @property
    def traceback(self):
123:    def as_tuple(self):
458:    def children(self):
1005:    def as_tuple(self):
1012:    def children(self):

def graph(self):
        return self.build_graph()

    @property
    def supports_native_join(self):
        return self.backend.supports_native_join

    @property
    def children(self):
        return self._get_task_meta().get('children')

    def _maybe_set_cache(self, meta):

        if meta:

[(0, 'AsyncResult', '05105d73'), (1, 'GroupResult', '8c14265b'), (2, 'AsyncResult', 'dc6dd673'), (2, 'AsyncResult', 'fbfa9d58'), (2, 'GroupResult', 'd85bc4a0'), (3, 'AsyncResult', '06351067'), (3, 'AsyncResult', 'af911a52'), (3, 'AsyncResult', 'e211c57b')]
restored GR:
  (0, 'GroupResult', '8c14265b')
  (1, 'AsyncResult', 'dc6dd673')
  (1, 'AsyncResult', 'fbfa9d58')
  (1, 'GroupResult', 'd85bc4a0')
  (2, 'AsyncResult', '06351067')
  (2, 'AsyncResult', 'af911a52')
  (2, 'AsyncResult', 'e211c57b')
same as GR? True
[["05105d73-5d5f-481a-ae0a-13f8c761125f", [["8c14265b-c358-4f22-b351-692e9f7b0c4f", [["d85bc4a0-4e37-449a-972b-fbcf412dde63", [["e211c57b-c2a0-4d70-9ce5-072cc18dbea9", null], null]], [[["06351067-1199-4ba5-9f7d-24f8381121f0", null], null], [["af911a52-60dc-4525-9f61-c79bc00c902f", null], null]]]], [[["dc6dd673-2c11-49ff-b99d-45c1d3701662", null], null], [["fbfa9d58-6f0c-4c55-b1dd-9ed45e5e6639", null], null]]]], null]
json roundtrip True

  R (0, 'AsyncResult', '05105d73')
  R (1, 'GroupResult', '8c14265b')
  R (2, 'AsyncResult', 'dc6dd673')
  R (2, 'AsyncResult', 'fbfa9d58')
  R (2, 'GroupResult', 'd85bc4a0')
  R (3, 'AsyncResult', '06351067')
  R (3, 'AsyncResult', 'af911a52')
  R (3, 'AsyncResult', 'e211c57b')

Traceback (most recent call last):
  File "<stdin>", line 5, in <module>
AssertionError

    def _on_fulfilled(self, result):
        self.backend.remove_pending_result(self)
        return result

    def as_tuple(self):
        parent = self.parent
        return (self.id, parent and parent.as_tuple()), None

    def as_list(self):
        """Return as a list of task IDs."""
        results = []
        parent = self.parent
        results.append(self.id)
        if parent is not None:

Traceback (most recent call last):
  File "<stdin>", line 8, in <module>
AssertionError
        self.backend.remove_pending_result(self)
        return result

    def as_tuple(self):
        parent = self.parent
        return (self.id, parent and parent.as_tuple()), None

    def as_list(self):
        """Return as a list of task IDs."""

        results = []
        parent = self.parent
        results.append(self.id)
        if parent is not None:
            results.extend(parent.as_list())

self.backend.remove_pending_result(self)
        return result

    def as_tuple(self):
        parent = self.parent
        return (
            (self.id, parent and parent.as_tuple()),
            self.children and [child.as_tuple() for child in self.children],
        )

    def as_list(self):
        """Return as a list of task IDs."""

        results = []
        parent = self.parent

..................s..................................................... [ 67%]
..................................                                       [100%]
105 passed, 1 skipped in 4.77s

ERROR: file or directory not found: t/unit/test_result.py


no tests ran in 0.01s

warn(RuntimeWarning(

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
988 passed, 1 skipped, 2 xfailed, 9 warnings, 28765 subtests passed in 30.38s

Traceback (most recent call last):
  File "/tmp/repro2b.py", line 12, in <module>
    t = res.as_tuple()
        ^^^^^^^^^^^^^^
  File "/repo/celery/result.py", line 127, in as_tuple
    self.children and [child.as_tuple() for child in self.children],
                       ^^^^^^^^^^^^^^
AttributeError: 'tuple' object has no attribute 'as_tuple'

nodes: [('GroupResult', 'group-2')]
OK: fan-out preserved

PromiseProxy IGNORED None
PromiseProxy REJECTED None
1 TGT OK
1 TGGT OK
1 TGG OK
1 GGT OK
1 GG OK
1 TGGGT OK
1 TGGT2 OK
2 TGT OK
2 TGGT OK
2 TGG OK

2 GGT OK
2 GG OK
2 TGGGT OK
2 TGGT2 OK
  R (3, 'AsyncResult', 'e3a208f3')
  R (3, 'AsyncResult', '7deed113')
  R (3, 'AsyncResult', '30650e59')
ERROR t/unit/backends/test_s3.py
!!!!!!!!!!!!!!!!!!! Interrupted: 7 errors during collection !!!!!!!!!!!!!!!!!!!!
8 skipped, 7 errors in 2.30s

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
312 passed, 1 skipped, 2 xfailed, 3 warnings, 29 subtests passed in 7.49s
ERROR t/unit/backends/test_rpc.py - Failed: 'celery' not found in `markers` c...
!!!!!!!!!!!!!!!!!!! Interrupted: 5 errors during collection !!!!!!!!!!!!!!!!!!!!
5 skipped, 5 errors in 1.18s

!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
1 error in 0.69s

ERROR t/unit/security/test_key.py
ERROR t/unit/security/test_security.py
ERROR t/unit/security/test_serialization.py
ERROR t/unit/utils/test_annotations.py
!!!!!!!!!!!!!!!!!!! Interrupted: 14 errors during collection !!!!!!!!!!!!!!!!!!!
9 skipped, 14 errors in 10.60s

ERROR t/unit/bin/test_daemonization.py::test_daemon_options_from_config[beat]
ERROR t/unit/bin/test_daemonization.py::test_daemon_options_from_config[events]
ERROR t/unit/bin/test_worker.py::test_cli
ERROR t/unit/bin/test_worker.py::test_cli_skip_checks
ERROR t/unit/bin/test_worker.py::test_cli_disable_prefetch_flag
8 failed, 2962 passed, 45 skipped, 3 xfailed, 50 warnings, 43 errors, 28817 subtests passed in 122.69s (0:02:02)

FAILED t/unit/concurrency/test_gevent.py::test_TaskPool::test_make_killable_target
FAILED t/unit/contrib/test_pytest.py::test_pytest_celery_marker_registration
FAILED t/unit/utils/test_platforms.py::test_close_open_fds - AssertionError: ...
FAILED t/unit/utils/test_platforms.py::test_maybe_drop_privileges::test_with_guid
FAILED t/unit/utils/test_platforms.py::test_maybe_drop_privileges::test_only_gid
FAILED t/unit/utils/test_platforms.py::test_check_privileges[accept_content0]
FAILED t/unit/utils/test_platforms.py::test_check_privileges[accept_content1]
FAILED t/unit/utils/test_platforms.py::test_check_privileges[accept_content2]

FAILED t/unit/utils/test_platforms.py::test_close_open_fds - AssertionError: ...
2 failed in 1.00s

value 10
first task id 772f496f-8cc2-401e-a05f-b249a5a2a678
children of first task: [<GroupResult: 994694d5-0cb1-46c7-b515-88e83aef1be3 [65c1c92d-2d12-4701-a9b1-a05573b2d8cd, 83145080-8bee-4744-b6bc-85a7c912ab03]>]
 child: GroupResult 994694d5
    (0, 'GroupResult', '994694d5')
    (1, 'AsyncResult', '65c1c92d')
    (1, 'AsyncResult', '83145080')
restored type: GroupResult
  R (0, 'GroupResult', '772f496f')
  R (1, 'GroupResult', '994694d5')
  R (2, 'GroupResult', '65c1c92d')
  R (2, 'GroupResult', '83145080')

diff --git a/celery/canvas.py b/celery/canvas.py
index 7e42d46..3035993 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -25,7 +25,8 @@ from vine import barrier
 
 from celery._state import current_app
 from celery.exceptions import CPendingDeprecationWarning
-from celery.result import GroupResult, allow_join_result
+from celery.result import EagerResult, GroupResult, allow_join_result
+from celery import states
 from celery.utils import abstract

 from celery.utils.collections import ChainMap
 from celery.utils.functional import _regen
@@ -1299,6 +1300,12 @@ class _chain(Signature):
             res = task.clone(fargs, fkwargs).apply(
                 last and (last.get(),), **dict(self.options, **options))
             res.parent, last, (fargs, fkwargs) = last, res, (None, None)
+            if isinstance(res, EagerResult) and res.state in (
+                states.IGNORED, st
…[481 chars omitted]…
   return (self.id, parent and parent.as_tuple()), None
+        children = self.children
+        return (
+            (self.id, parent and parent.as_tuple()),
+            children and [
+                child.as_tuple() if isinstance(child, ResultBase) else child
+                for child in children
+            ],
+        )
 
     def as_list(self):
         """Return as a list of task IDs."""

PromiseProxy IGNORED None
PromiseProxy REJECTED None
nodes: [('GroupResult', 'group-2')]
OK: fan-out preserved
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
988 passed, 1 skipped, 2 xfailed, 9 warnings, 28765 subtests passed in 32.63s

value 10
first task id 5c75d43c-aba1-4855-af1c-cfd16873e5b7
children of first task: [<GroupResult: 031a8177-ea80-4036-85f3-370c1aba5ac0 [48425df5-154f-48fb-8872-506d636a5f1a, 16e7de23-5217-4d5c-aa30-e4fc73c9bfa0]>]
 child: GroupResult 031a8177
    (0, 'GroupResult', '031a8177')
    (1, 'AsyncResult', '48425df5')
    (1, 'AsyncResult', '16e7de23')
restored type: GroupResult
  R (0, 'GroupResult', '5c75d43c')
  R (1, 'GroupResult', '031a8177')
  R (2, 'GroupResult', '48425df5')
  R (2, 'GroupResult', '16e7de23')

!!!!!!!!!!!!!!!!!!! Interrupted: 3 errors during collection !!!!!!!!!!!!!!!!!!!!
3 errors in 1.89s

ERROR t/unit/backends/test_base.py - Failed: 'celery' not found in `markers` ...
!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
1 error in 0.73s

!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!
1 error in 0.60s
!!!!!!!!!!!!!!!!!!! Interrupted: 2 errors during collection !!!!!!!!!!!!!!!!!!!!
2 errors in 1.66s

submitted.