LQConstPlanWriterAgent/foundation/infrastructure/messaging/tasks.py

from .celery_app import app
from core.construction_write.workflow_manager import workflow_manager
from foundation.infrastructure.tracing import TraceContext
from foundation.observability.logger.loggering import write_logger as logger


def _is_non_retryable_error(error: Exception) -> bool:
    error_str = str(error).lower()
    non_retryable_tokens = [
        "401",
        "403",
        "unauthorized",
        "forbidden",
        "invalid api key",
        "incorrect api key",
        "authentication",
        "permission denied",
    ]
    return any(token in error_str for token in non_retryable_tokens)


@app.task(bind=True, queue="construction_write")
def submit_outline_generation_task(self, task_info: dict, _system_trace_id: str = None):
    """Celery task for construction-write outline generation."""
    # 恢复 trace_id 上下文
    if _system_trace_id:
        TraceContext.set_trace_id(_system_trace_id)
        logger.info(f"Celery任务恢复 trace_id: {_system_trace_id}")

    callback_task_id = task_info.get("callback_task_id")
    project_name = task_info.get("project_info", {}).get("project_name", "")
    user_id = task_info.get("user_id", "")

    logger.info(f"=== Celery任务接收调试 ===")
    logger.info(f"队列ID: {self.request.id}")
    logger.info(f"回调任务ID: {callback_task_id}")
    logger.info(f"用户ID: {user_id}")
    logger.info(f"项目: {project_name}")
    logger.info(f"开始执行大纲生成业务逻辑")

    try:
        self.update_state(
            state="current",
            meta={
                "current": 0,
                "total": 100,
                "status": "outline_generation_started",
                "callback_task_id": callback_task_id,
                "project_name": project_name,
            },
        )

        result = workflow_manager.submit_outline_generation_sync(task_info)

        self.update_state(
            state="current",
            meta={
                "current": 100,
                "total": 100,
                "status": "outline_generation_completed",
                "callback_task_id": result.get("callback_task_id"),
                "overall_task_status": result.get("overall_task_status"),
            },
        )

        return {
            "status": "success",
            "callback_task_id": result.get("callback_task_id"),
            "overall_task_status": result.get("overall_task_status"),
            "result": result,
        }
    except Exception as exc:
        logger.error(f"大纲生成任务失败: {exc}")
        logger.exception("详细错误信息:")
        if _is_non_retryable_error(exc):
            logger.error("检测到不可重试错误，Celery任务不再重试")
            raise
        self.retry(countdown=60, max_retries=2, exc=exc)
        raise