From 4d98e98b5775a05eea8f3dca1906759d68c0d755 Mon Sep 17 00:00:00 2001
From: xiaoxia <dev@xiaoxiajianji.com>
Date: Sat, 19 Sep 2026 08:42:07 +0800
Subject: [PATCH] =?UTF-8?q?fix(worker):=20#1970=20=E6=B3=A8=E5=86=8C=20AI?=
 =?UTF-8?q?=20=E6=A0=87=E7=AD=BE=20Celery=20=E4=BB=BB=E5=8A=A1=EF=BC=88wor?=
 =?UTF-8?q?ker.tag=5Fatom=5Fclip=20unregistered=EF=BC=89=20(#1987)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
---
 apps/worker/worker_app/celery_app.py          |  4 +
 .../test_1970_celery_ai_tag_registration.py   | 82 +++++++++++++++++++
 2 files changed, 86 insertions(+)
 create mode 100644 tests/unit/test_1970_celery_ai_tag_registration.py

diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py
index 915d43a36..8d9683bf4 100755
--- a/apps/worker/worker_app/celery_app.py
+++ b/apps/worker/worker_app/celery_app.py
@@ -28,6 +28,10 @@ celery_app.conf.imports = (
     "worker_app.tasks.health",
     "worker_app.tasks.ingest",
     "worker_app.tasks.atom_clips",
+    # #1970 片段级 AI 标签：必须显式 import 注册，否则 worker 报
+    # "Received unregistered task of type 'worker.tag_atom_clip'"
+    "worker_app.tasks.atom_clip_tagging",
+    "worker_app.tasks.backfill_atom_clip_tags",
     "worker_app.tasks.classification",
     "worker_app.tasks.generation",
     "worker_app.tasks.voice_extraction",
diff --git a/tests/unit/test_1970_celery_ai_tag_registration.py b/tests/unit/test_1970_celery_ai_tag_registration.py
new file mode 100644
index 000000000..c1a3e0a89
--- /dev/null
+++ b/tests/unit/test_1970_celery_ai_tag_registration.py
@@ -0,0 +1,82 @@
+"""#1970 AI 标签 Celery 任务注册回归测试。
+
+背景：staging 上 worker.generate_atom_clips 正常派发 tag_atom_clip，
+但消费端报 "Received unregistered task of type 'worker.tag_atom_clip'"，
+根因是 celery_app.conf.imports 漏列任务模块，worker 进程从未 import 之。
+
+注意：tests/unit 下大量旧测试在 import 期向 sys.modules 注入
+worker_app.celery_app 的 MagicMock 且不还原，全量收集时会污染本测试，
+因此这里用 AST 静态解析 + 隔离子进程验证，不依赖 sys.modules 状态。
+"""
+
+from __future__ import annotations
+
+import ast
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+CELERY_APP_PY = REPO_ROOT / "apps" / "worker" / "worker_app" / "celery_app.py"
+
+REQUIRED_MODULES = (
+    "worker_app.tasks.atom_clip_tagging",
+    "worker_app.tasks.backfill_atom_clip_tags",
+)
+
+
+def _conf_imports_values() -> set[str]:
+    """从 celery_app.py AST 中提取 celery_app.conf.imports 元组的字符串项。"""
+    tree = ast.parse(CELERY_APP_PY.read_text(encoding="utf-8"))
+    values: set[str] = set()
+    for node in ast.walk(tree):
+        if not (isinstance(node, ast.Assign) and len(node.targets) == 1):
+            continue
+        target = node.targets[0]
+        # celery_app.conf.imports = (...) 或 conf.imports = (...)
+        if not (isinstance(target, ast.Attribute) and target.attr == "imports"):
+            continue
+        if isinstance(node.value, (ast.Tuple, ast.List)):
+            for elt in node.value.elts:
+                if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
+                    values.add(elt.value)
+    return values
+
+
+def test_ai_tag_modules_in_celery_imports():
+    imports = _conf_imports_values()
+    for module in REQUIRED_MODULES:
+        assert module in imports, f"{module} 未加入 celery_app.conf.imports"
+
+
+def test_ai_tag_tasks_registered_in_isolated_process():
+    """隔离子进程（无 conftest / 无 sys.modules mock）真实加载 Celery app。"""
+    # 模拟 worker 启动时按 conf.imports import 任务模块的行为；
+    # 只导入 AI 标签两个模块（其他模块依赖 cv2 等本地未安装的重依赖）。
+    code = (
+        "import importlib, sys; "
+        "from worker_app.celery_app import celery_app; "
+        "mods = [m for m in celery_app.conf.imports or () "
+        "if 'atom_clip_tagging' in m or 'backfill_atom_clip_tags' in m]; "
+        "[importlib.import_module(m) for m in mods]; "
+        "missing = [n for n in "
+        "['worker.tag_atom_clip', 'worker.backfill_atom_clip_tags'] "
+        "if n not in celery_app.tasks]; "
+        "sys.exit(1 if missing or len(mods) < 2 else 0)"
+    )
+    env = os.environ.copy()
+    paths = [
+        str(REPO_ROOT),
+        str(REPO_ROOT / "apps" / "worker"),
+        str(REPO_ROOT / "packages"),
+    ]
+    env["PYTHONPATH"] = os.pathsep.join(paths) + os.pathsep + env.get("PYTHONPATH", "")
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=60,
+    )
+    assert result.returncode == 0, "隔离子进程中任务未注册成功:\n" f"stdout={result.stdout}\nstderr={result.stderr}"