refactor: 位置回溯极简—只跟主干任务(WIP>PENDING>COMPLETED),无视协助分支

backend/app/services/task_service.py

@@ -34,68 +34,58 @@ VIRTUAL_WAREHOUSE = "virtual_warehouse"
 ADMIN_ROLES = {"SUPER_ADMIN", "SUPERVISOR"}
 
 
-async def _recalc_product_location(db: AsyncSession, product_id: uuid.UUID, completed_task_id: uuid.UUID | None = None) -> None:
+async def _recalc_product_location(
+    db: AsyncSession, product_id: uuid.UUID, completed_task_id: uuid.UUID | None = None,
+) -> None:
     """
-    任务完工/结束时触发：位置回溯 父任务优先 策略。
-    ① 如果完工任务有父任务且父任务未完成 → 强制回溯到父任务负责人
-    ② 否则查找所有 WIP 任务 → 最新 WIP 负责人
-    ③ 无 WIP → 兜底最后完工者
-    ④ 全结束 → 置空
+    任务完工/结束时触发：只跟随主干任务（主分支），无视协助分支。
+
+    主干任务定义：
+      parent_task_id IS NULL OR task_type IN ('TRANSFER', 'RECOVERY')
+
+    优先级：
+      WIP > PENDING > COMPLETED/ARCHIVED > None
     """
-    from sqlalchemy import select as sa_select
+    from sqlalchemy import select as sa_select, case as sa_case, func as sa_func
 
-    # ① 父任务优先：如果有父任务且未完成 → 位置给父任务负责人
-    if completed_task_id:
-        task_result = await db.execute(
-            sa_select(Task).where(Task.id == completed_task_id)
-        )
-        current_task = task_result.scalar_one_or_none()
-        if current_task and current_task.parent_task_id:
-            parent_result = await db.execute(
-                sa_select(Task).where(Task.id == current_task.parent_task_id)
-            )
-            parent = parent_result.scalar_one_or_none()
-            if parent and parent.status not in (TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED, TASK_STATUS_CANCELED, TASK_STATUS_ARCHIVED):
-                product_result = await db.execute(
-                    sa_select(Product).where(Product.id == product_id)
-                )
-                product = product_result.scalar_one_or_none()
-                if product and product.current_location_id != parent.assignee_id:
-                    product.current_location_id = parent.assignee_id
-                return
-
-    # ② 无父任务/父任务已完工 → 查找其他 WIP 任务
-    product_result = await db.execute(sa_select(Product).where(Product.id == product_id))
+    product_result = await db.execute(
+        sa_select(Product).where(Product.id == product_id)
+    )
     product = product_result.scalar_one_or_none()
     if not product:
         return
 
-    task_result = await db.execute(
-        sa_select(Task).where(
+    # ── 只查主干任务: parent_task_id IS NULL 或 task_type IN (TRANSFER, RECOVERY) ──
+    stmt = (
+        sa_select(Task)
+        .where(
             Task.product_id == product_id,
-            Task.status == TASK_STATUS_WIP,
-            Task.id != completed_task_id,  # 🔧 排除刚刚完工的任务（避免脏读）
-        ).order_by(Task.created_at.desc())
-    )
-    wip_tasks = task_result.scalars().all()
-
-    if wip_tasks:
-        latest_wip = wip_tasks[0]
-        new_location = latest_wip.assignee_id or product.current_location_id
-    else:
-        # 无进行中任务 → 查找最新的COMPLETED任务负责人（保留最后完工者）
-        done_result = await db.execute(
-            sa_select(Task).where(
-                Task.product_id == product_id,
-                Task.status == TASK_STATUS_COMPLETED,
-            ).order_by(Task.completed_at.desc()).limit(1)
+            sa_func.or_(
+                Task.parent_task_id.is_(None),
+                Task.task_type.in_(["TRANSFER", "RECOVERY"]),
+            ),
         )
-        last_done = done_result.scalar_one_or_none()
-        new_location = last_done.assignee_id if last_done else None
+        .order_by(
+            # 优先级排序: WIP=3, PENDING=2, COMPLETED=1, ARCHIVED=1, else=0
+            sa_case(
+                (Task.status == TASK_STATUS_WIP, 3),
+                (Task.status == TASK_STATUS_PENDING, 2),
+                (Task.status == TASK_STATUS_COMPLETED, 1),
+                (Task.status == TASK_STATUS_ARCHIVED, 1),
+                else_=0,
+            ).desc(),
+            Task.created_at.desc(),
+        )
+        .limit(1)
+    )
+    result = await db.execute(stmt)
+    main_task = result.scalar_one_or_none()
+
+    new_location = main_task.assignee_id if main_task else None
 
     if product.current_location_id != new_location:
         product.current_location_id = new_location
-        await db.flush()  # 🔧 确保位置更新落盘到当前事务
+        await db.flush()  # 唯一的落盘点
 
 
 def _check_permission(task_assignee_id: str | None, operator_id: str | None, operator_role: str | None = None) -> None: