iflytek/astron-agent · error · CustomException

CodeEnum.ASYNC_TASK_CANCEL_ERROR

CodeEnum.ASYNC_TASK_CANCEL_ERROR

Error message

Failed to cancel async task

What it means

CeleryTaskProcessor.cancel_task wraps any exception raised by the caller-supplied cancel_func (e.g. AsyncResult.revoke on a Celery app) in a CustomException with code ASYNC_TASK_CANCEL_ERROR. It signals that a previously launched Celery async task could not be cancelled, with the underlying cause stored in cause_error.

Solutions

1. Inspect cause_error on the raised CustomException to identify the underlying exception (broker connection vs. invalid task id).
2. Verify the Celery broker (Redis/RabbitMQ) is reachable from the workflow service.
3. Confirm the task_id passed to cancel_func refers to a still-running task; finished tasks cannot be revoked.
4. Ensure cancel_func signature matches (app, **kwargs) and all forwarded kwargs are supported.
5. Wrap cancel_task in try/except in the caller and treat cancellation of already-finished tasks as non-fatal.

Example fix

// before
cancel_func = lambda app, task_id: app.control.revoke(task_id, terminate=True)
processor.cancel_task(cancel_func)  # TypeError: missing task_id -> ASYNC_TASK_CANCEL_ERROR
// after
processor.cancel_task(cancel_func, task_id=result.id, terminate=True)

def can_cancel(processor, task_id):
    res = celery_app.AsyncResult(task_id)
    return res.state in ("PENDING", "STARTED", "RETRY")

def is_running(res) -> bool:
    return getattr(res, "state", None) in ("PENDING", "STARTED", "RETRY")

try:
    processor.cancel_task(cancel_func, task_id=task_id)
except CustomException as e:
    logger.warning(f"Cancel failed for {task_id}: {e}")  # inspect e.cause_error / cause

When it happens

Trigger: Calling cancel_task with a cancel_func that raises: revoking a task id that no longer exists or already finished, Celery broker (Redis/RabbitMQ) being unreachable, or passing a cancel_func whose signature does not match the **kwargs forwarded to it.

Common situations: Revoking a workflow async task after its Celery TTL expired; broker connection lost during a deploy; passing kwargs (e.g. task_id, terminate) that the cancel_func does not accept.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• 21601
• 21602
• 23900
• 23900
• 23900

AI-assisted analysis of iflytek/astron-agent@5e758547a8 (2026-09-12). Data as JSON: /api/errors/b786f20e81c1ea18. Report an issue: GitHub.

Appendix: source

Thrown at core/workflow/extensions/middleware/asynchronous/manager.py:31

class CeleryTaskProcessor(AsyncTaskService, Service):
    """Celery-based implementation of the AsyncTaskService."""

    def __init__(self) -> None:
        self.app: Celery = app

    def launch_task(self, task_func: Callable, *args: Any, **kwargs: Any) -> str:
        """Launch a celery task and return the task id."""
        if not hasattr(task_func, "delay"):
            msg = f"Task function {task_func} does not have a delay method"
            raise ValueError(msg)
        result: AsyncResult = task_func.delay(*args, **kwargs)
        return result.id

    def cancel_task(self, cancel_func: Callable[[Any], None], **kwargs: Any) -> None:
        try:
            cancel_func(self.app, **kwargs)
        except Exception as e:
            raise CustomException(
                CodeEnum.ASYNC_TASK_CANCEL_ERROR,
                err_msg="Failed to cancel async task",
                cause_error=str(e),
            ) from e

View on GitHub (pinned to 5e758547a8)