feat: 直接完结通道与报表口径收口（后端）

【直接完结：保留动作，剥离状态】
- transfer_task 权限硬拦截：仅 SUPER_ADMIN / SUPERVISOR。判据取签名保护的
  operator_role，刻意不用客户端可通过 ?operator_id= 伪造的 operator_id
- 直接完结【不再改写】overall_status —— 它只是任务闭环动作，不改变物理状态。
  已出库设备完结后依然是「已出库」，回归 WIP 矩阵的已出库列
- 物理终态保护：create_task / receive_task / transfer_task 三处写入点统一加
  _is_physical_terminal 守卫，禁止用任务名覆写 已入库/在库/已出库。
  历史缺陷：「发货测试」会把设备的「已出库」标识静默抹掉，跨报表口径随之打架

【位置与工序口径】
- _recalc_product_location：已出库且闲置 → 位置清空（货发走就离场）。
  只清「已出库」；已入库/在库的设备确实还在仓库里，位置必须保留
- MOM 出库回调同步清空 current_location_id
- 新增 ProductResponse.current_step：只认活跃主干任务，无活跃任务时返回
  宏观终态或空。不再让已 COMPLETED 的历史工序（扫码出库/测试…）冒充"当前工序"

【售后烙印去死锁】
- 拆分 OUTBOUND_QC_STEPS(发货测试) / AFTER_SALES_REPAIR_STEPS(售后维修)
- resolve_phase_for_step 与 _mark_after_sales_if_reactivated 双双改为仅
  「售后维修」触发 AFTER_SALES，解除「已出库 + 发货测试」被永久烙印的死锁
- PRODUCTION_OVERALL_STEPS 补入「发货测试」，使出厂质检在生产阶段合法
  （否则 _enforce_step_isolation 会把已出库设备的发货测试直接 400）

【游魂收口】
- get_wip_matrix / get_wip_matrix_detail 的 is_terminal 增加"名下再无任何活跃
  任务则视同终结态"，让被直接完结的生产设备接受时间筛选，不再恒挂在看板上
  冒充在制。两处必须一字不差同步，否则会出现"矩阵有数、下钻为空"

backend/app/api/v1/endpoints/webhooks.py

@@ -256,6 +256,14 @@ async def mom_outbound_webhook(
         product.status = "OUTBOUND"
         changed = True
 
+    # 🚚 同步出清厂内位置（与 _recalc_product_location 的「货发走就离场」对齐）：
+    #    本回调的匹配条件之一就是 current_location_id == "virtual_warehouse"，
+    #    即设备此刻还挂在仓库池里。既然 MOM 已确认发货，就不该再显示为厂内仓库/工位。
+    #    置空后产品列表的「当前位置」显示为「—」。
+    if product.current_location_id is not None:
+        product.current_location_id = None
+        changed = True
+
     # 记录出库日志（优先"在库"任务，其次该产品最新任务）
     outbound_task = (
         await db.execute(

backend/app/core/lifecycle.py

@@ -29,21 +29,35 @@ PHASE_LABELS = {
 }
 
 # ============================================================
-# 售后专属工序名 — 一旦选中即代表设备进入售后生命周期
+# 售后专属工序名
 # ============================================================
-STEP_SHIP_TEST = "发货测试"          # 售后返修后的出货测试
+STEP_SHIP_TEST = "发货测试"          # 出厂 / 发货前测试
 STEP_AFTER_SALES_REPAIR = "售后维修"  # 售后返修本体
 
+# 售后区独立成列所用的两个工序名（前端分列 / 大屏返厂统计仍以它为准）
 AFTER_SALES_ONLY_STEPS = frozenset({STEP_SHIP_TEST, STEP_AFTER_SALES_REPAIR})
 
+# 🔧 工序语义细分（2026-09-17）——「售后专属」不等于「回流返厂」：
+#    · 出厂质检（发货测试）：设备可能压根没回厂，只是补做发货前测试。
+#      它【不】构成回流信号，选中它不应把设备永久烙进售后生命周期。
+#    · 明确修机指令（售后维修）：设备确实是回厂返修 → 判定回流。
+#    历史缺陷：两者混为一谈，导致「已出库 + 发货测试」被打上不可回退的
+#    AFTER_SALES 烙印（售后死锁），而「已出库」的物理终态也被抹掉。
+OUTBOUND_QC_STEPS = frozenset({STEP_SHIP_TEST})
+AFTER_SALES_REPAIR_STEPS = frozenset({STEP_AFTER_SALES_REPAIR})
+
 # ============================================================
 # 各阶段合法的工序名 / 宏观状态名
 # ============================================================
 
-# ── 生产制造阶段（原有词表，一字未改，保证存量流程不受影响）──
+# ── 生产制造阶段 ──
+#   🔧 2026-09-17 补入「发货测试」：出厂质检现在被允许在生产阶段执行
+#      （设备的 lifecycle_phase 不再因选中它而翻成售后）。若不补，
+#      _enforce_step_isolation 会把「已出库设备的发货测试」直接 400 掉。
 PRODUCTION_TASK_STEPS = ("备货", "生产", "测试", "维修", "在库")
 PRODUCTION_OVERALL_STEPS = (
     "备货", "生产", "测试", "维修", "在库", "待仓库收货", "已入库", "已出库",
+    STEP_SHIP_TEST,
 )
 
 # ── 售后回流阶段：只保留「发货测试 / 售后维修 / 入库出库」──
@@ -99,12 +113,15 @@ def is_placeholder_step(step: str | None) -> bool:
 
 
 def resolve_phase_for_step(current_phase: str | None, step: str | None) -> str:
-    """选定售后专属工序 → 设备进入售后生命周期（单向，不可回退）
+    """选定【明确修机指令】→ 设备进入售后生命周期（单向，不可回退）
 
-    这是"无历史记录的老设备"进入售后阶段的唯一入口：它首次选定
+    ⚠️ 只有「售后维修」触发，出厂质检（发货测试）【不】触发。
-    「发货测试 / 售后维修」时即被判定为售后回流设备。
+       「已出库的设备补做发货测试」不代表它回厂返修；若在此烙印，
+       设备会被永久打成售后（该标志单向不可回退），这就是售后死锁的第二个入口。
+
+    这是"无历史记录的老设备"进入售后阶段的入口：首次选定「售后维修」即判定回流。
     """
-    if step and step.strip() in AFTER_SALES_ONLY_STEPS:
+    if step and step.strip() in AFTER_SALES_REPAIR_STEPS:
         return LIFECYCLE_AFTER_SALES
     return current_phase or LIFECYCLE_PRODUCTION
 

backend/app/schemas/product.py

@@ -53,6 +53,14 @@ class ProductResponse(BaseModel):
     current_location_name: str | None = None
     macro_status: str | None = None  # 🔧 后端预计算的任务树状态（免前端逐条展开）
     overall_status: str | None = None
+    # 【当前工序】—— 只反映"此刻在做什么"，绝不拿历史工序冒充。
+    #   · 有活跃主干任务(WIP/PENDING) → 该任务工序名
+    #   · 否则产品处于宏观终态(待仓库收货/已入库/在库/已出库) → 该终态（表达"货在哪"）
+    #   · 否则（活已干完、只剩工序名残留）→ ""（前端显示「—」）
+    # ⚠️ 必须与 overall_status 分开：ProductResponse.overall_status 会被"最新主干任务名"
+    #    覆盖（见 product_service 的 overall_names），于是已 COMPLETED 的历史工序
+    #    （扫码出库 / 测试 / 发货测试…）会被当成"当前工序"长期展示。
+    current_step: str = ""
     status: str
     # 🔧 生命周期阶段：PRODUCTION(生产制造/发货测试) | AFTER_SALES(出库后返厂售后维修)
     lifecycle_phase: str = "PRODUCTION"

backend/app/schemas/task.py

@@ -97,8 +97,10 @@ class TaskTransferRequest(BaseModel):
     finish_directly: bool = Field(
         False,
         description=(
-            "直接完结：闭环当前任务但【不产生任何下游任务】，"
+            "直接完结：闭环当前任务但【不产生任何下游任务】。"
-            "产品 overall_status 保持原样（不入库 → 不会变成「待仓库收货」）。"
+            "它只是一个任务闭环动作，【不改动】产品的 overall_status —— "
+            "已出库的设备完结后依然是已出库（不入库，也不会变成「待仓库收货」）。"
+            "权限：仅 SUPER_ADMIN / SUPERVISOR 可调用，其余角色 403。"
             "置 True 时忽略 next_tasks / next_assignees。"
         ),
     )

backend/app/services/dashboard_service.py

@@ -918,9 +918,13 @@ async def get_wip_matrix(
       - task_name: 按当前工序聚合（dimension_key 为工序名，附主负责人）
     """
     from datetime import timezone as dt_timezone
+    from sqlalchemy.orm import aliased
     from app.models.task import Task
     from app.models.product import Product
 
+    # 外层已 join 了 tasks，内层判活跃任务必须换别名，否则语义无歧义
+    _ActiveTask = aliased(Task, name="active_task")
+
     # 每台设备按主任务创建时间倒序，取第一条即「最新主任务」
     result = await db.execute(
         select(
@@ -936,6 +940,15 @@ async def get_wip_matrix(
             Product.overall_status,
             Product.status,
             Product.lifecycle_phase,
+            # 🔧 该产品名下是否还有 ANY 活跃任务（含协助分支）——
+            #    注意：这比下面的 is_active 更宽。is_active 只看"最新一条主干任务"，
+            #    而这里扫描该产品的全部任务，用于 is_terminal 兜底判定。
+            #    ⚠️ 必须用别名：外层 FROM 里已有一个 tasks，同名会让 SQLAlchemy
+            #       自动关联掉内层 FROM 而报 "no FROM clauses"。别名后语义才无歧义。
+            select(_ActiveTask.id).where(
+                _ActiveTask.product_id == Product.id,
+                _ActiveTask.status.in_(["WIP", "PENDING"]),
+            ).exists().label("has_any_active"),
         )
         .outerjoin(Task, Task.product_id == Product.id)
         .where(
@@ -952,7 +965,7 @@ async def get_wip_matrix(
     # 每台设备 → (spec, 当前工序/负责人, 当前负责人ID)
     device_cur: dict[str, tuple] = {}
     seen: set[str] = set()
-    for pid, spec, product_name, task_name, assignee, tstatus, created, completed, loc, overall_status, product_status, lifecycle in rows:
+    for pid, spec, product_name, task_name, assignee, tstatus, created, completed, loc, overall_status, product_status, lifecycle, has_any_active in rows:
         if pid in seen:
             continue
         seen.add(pid)
@@ -975,6 +988,13 @@ async def get_wip_matrix(
             or overall_status in ("待仓库收货", "已入库", "在库", "已出库")
             or str(product_status).upper() in ("ARCHIVED", "OUTBOUND")
             or task_name in ("扫码入库", "扫码出库")
+            # 🔧 兜底（2026-09-17）：设备名下【再无任何活跃任务】= 真的没活了 → 视同终结态。
+            #    历史缺陷：生产设备被【直接完结】后 overall_status 停在工序名（如「测试」），
+            #    既不满足上面任何一条，又因 is_terminal=False 对时间筛选完全免疫 ——
+            #    于是它永远挂在看板上冒充"在制"（报表游魂）。
+            #    补上这一条后，它照常接受 since~until 过滤，不再污染在制口径。
+            #    仍保留原有的终结态判定（本条件是「或」，只会让更多设备被正确归类为终结）。
+            or not has_any_active
         )
         if is_terminal:
             if anchor is not None and anchor.tzinfo is None:
@@ -1094,11 +1114,15 @@ async def get_wip_matrix_detail(
     按其在库三态（已完成/已入库/已出库）或工序名归类到 process。
     """
     from datetime import timezone as dt_timezone
+    from sqlalchemy.orm import aliased
     from app.models.task import Task
     from app.models.product import Product
     from app.models.holiday import Holiday
     from app.core.time_utils import get_beijing_time, to_beijing, working_duration_hours
 
+    # 与外层 join 的 tasks 区分开（理由见 get_wip_matrix）
+    _ActiveTask = aliased(Task, name="active_task")
+
     hres = await db.execute(select(Holiday.day))
     holidays = {r[0] for r in hres}
 
@@ -1119,6 +1143,12 @@ async def get_wip_matrix_detail(
             Task.created_at,
             Task.completed_at,
             Task.received_at,
+            # 🔧 与 get_wip_matrix 同口径：该产品名下是否还有 ANY 活跃任务
+            #    （同样必须用别名，理由见 get_wip_matrix 处注释）
+            select(_ActiveTask.id).where(
+                _ActiveTask.product_id == Product.id,
+                _ActiveTask.status.in_(["WIP", "PENDING"]),
+            ).exists().label("has_any_active"),
         )
         .outerjoin(Task, Task.product_id == Product.id)
         .where(
@@ -1137,7 +1167,7 @@ async def get_wip_matrix_detail(
     matched: list[WipMatrixDetailRow] = []
     raw_names: set[str] = set()
 
-    for pid, serial, ext, mat_name, spec, loc, overall, pstatus, lifecycle, task_name, assignee, tstatus, created, completed, received in rows:
+    for pid, serial, ext, mat_name, spec, loc, overall, pstatus, lifecycle, task_name, assignee, tstatus, created, completed, received, has_any_active in rows:
         if pid in seen:
             continue
         seen.add(pid)
@@ -1154,11 +1184,15 @@ async def get_wip_matrix_detail(
         is_active = tstatus in ("WIP", "PENDING")
 
         # ── 动静分离：仅终结(收口/在库/出库)设备按 since~until 过滤；在制设备无条件全量 ──
+        #    ⚠️ 必须与 get_wip_matrix 一字不差同步，否则会出现
+        #       「矩阵格子里有数、点进来下钻却是空的」。
         is_terminal = (not is_active) and (
             loc == "virtual_warehouse"
             or overall in ("待仓库收货", "已入库", "在库", "已出库")
             or str(pstatus).upper() in ("ARCHIVED", "OUTBOUND")
             or task_name in ("扫码入库", "扫码出库")
+            # 🔧 兜底：名下再无任何活跃任务 → 视同终结态，接受时间筛选（防报表游魂）
+            or not has_any_active
         )
         if is_terminal:
             if anchor is not None and anchor.tzinfo is None:

backend/app/services/product_service.py

@@ -673,6 +673,8 @@ async def get_all_products(
 
     # 🔧 动态主干状态名：只从主干任务中获取最高优先级任务的 task_name（宏观状态名）
     overall_names: dict[uuid.UUID, str] = {}
+    # 🔧 【当前工序】专用：仅「活跃主干任务」的工序名（见下方 main_stmt 循环填充）
+    active_step_map: dict[uuid.UUID, str] = {}
     if product_ids:
         from sqlalchemy import and_, func as sa_func, case as sa_case
         main_where = and_(
@@ -694,7 +696,7 @@ async def get_all_products(
             .group_by(Task.product_id)
         ).subquery("mp")
         main_stmt = (
-            select(Task.product_id, Task.task_name)
+            select(Task.product_id, Task.task_name, Task.status)
             .join(max_prio, and_(
                 Task.product_id == max_prio.c.product_id,
                 prio_expr == max_prio.c.prio,
@@ -706,6 +708,11 @@ async def get_all_products(
         main_result = await db.execute(main_stmt)
         for row in main_result:
             overall_names[row[0]] = row[1]
+            # 🔧 供【当前工序】专用：只收【活跃】主干任务的工序名。
+            #    overall_names 保持原语义不动（它连 COMPLETED 的任务也收，
+            #    是"最新主干任务名"，被 overall_status 复用，改动影响面太大）。
+            if row[2] in ("WIP", "PENDING"):
+                active_step_map[row[0]] = row[1]
 
     # 🔧 当前位置：汇总所有活跃任务（WIP/PENDING，不分主线/分支）的负责人，去重保序
     active_assignees_map: dict[uuid.UUID, list[str]] = {}
@@ -825,6 +832,28 @@ async def get_all_products(
             return "COMPLETED"
         return macro_map.get(p.id) or "PENDING"
 
+    # 宏观终态 —— 表达的是"货在哪"，不是"在做什么"。
+    # 这类值即使当前没有活跃任务也必须照常展示（业务要求保留物理标识）。
+    _TERMINAL_OVERALL = ("待仓库收货", "已入库", "在库", "已出库")
+
+    def _resolve_current_step(p: Product) -> str:
+        """【当前工序】—— 只反映"此刻在做什么"，绝不拿已完结的历史工序冒充。
+
+        · 有活跃主干任务(WIP/PENDING) → 该任务工序名
+        · 否则产品处于宏观终态 → 该终态（"货在哪"，保留）
+        · 否则（活已干完、只剩工序名残留）→ ""（前端显示「—」）
+
+        历史缺陷：ProductResponse.overall_status 会被"最新主干任务名"覆盖
+        （见上方 overall_names），于是已 COMPLETED 的「扫码出库 / 测试 / 发货测试」
+        会长期冒充"当前工序"，误导管理者以为活还在干。
+        """
+        raw = (p.overall_status or "").strip()
+        if not raw:
+            return ""
+        if raw in _TERMINAL_OVERALL:
+            return raw
+        return active_step_map.get(p.id, "")
+
     return [
         ProductResponse(
             id=p.id,
@@ -849,6 +878,7 @@ async def get_all_products(
             ),
             macro_status=_resolve_macro_status(p),
             overall_status=overall_names.get(p.id) or p.overall_status,
+            current_step=_resolve_current_step(p),
             status=p.status,
             lifecycle_phase=p.lifecycle_phase,
             created_at=p.created_at,

backend/app/services/task_service.py

@@ -11,6 +11,7 @@ from app.models.task import Task, TaskRecord, TASK_STATUS_PENDING, TASK_STATUS_W
 from app.models.notification import Notification, NOTIFY_TRANSFER, NOTIFY_REJECT
 from app.core.time_utils import get_beijing_time
 from app.core.lifecycle import (
+    AFTER_SALES_REPAIR_STEPS,
     LIFECYCLE_AFTER_SALES,
     PRODUCTION_ONLY_STEPS,
     allowed_steps,
@@ -91,26 +92,61 @@ async def _recalc_product_location(
     result = await db.execute(stmt)
     main_task = result.scalar_one_or_none()
 
+    # 上面的排序把 WIP(3) > PENDING(2) 放最前，所以取到的这条即最高优先级：
+    # 它是 WIP/PENDING 说明产品仍有活跃主线任务，否则视为闲置。
+    has_active_main = main_task is not None and main_task.status in (
+        TASK_STATUS_WIP, TASK_STATUS_PENDING,
+    )
+
     new_location = main_task.assignee_id if main_task else None
 
+    # 🚚 「货发走就离场」——已出库且闲置的设备，厂内不再有它的位置。
+    #    直接完结后若继续停在最后经手人名下（或残留 virtual_warehouse），
+    #    列表里会与「已出库」自相矛盾。置空后前端统一显示「—」。
+    #    ⚠️ 只清「已出库」：已入库 / 在库 的设备确实还在仓库里，位置必须保留。
+    if product.overall_status == "已出库" and not has_active_main:
+        new_location = None
+
     if product.current_location_id != new_location:
         product.current_location_id = new_location
         await db.flush()  # 唯一的落盘点
 
 
+# 绝对物理终态 — 描述「设备此刻物理上在哪」的宏观状态，由 MOM 仓储/发货回调驱动。
+# 接收 / 派发 / 转交任务时【禁止】用任务名覆写它们：否则「已出库」会被
+# 「发货测试」这类工序名静默抹掉，设备在按终态口径统计的报表里就不再是已出库
+# （WIP 矩阵因为优先看活跃任务而看不出问题，但全局概览 / 大屏会把它移出已完结）。
+#
+# ⚠️ 刻意不含「待仓库收货」：那是「车间完工待实收」的过渡态，设备随后仍会被
+#    重新派活，宏观状态理应随工序更新。
+PHYSICAL_TERMINAL_OVERALL = ("已入库", "在库", "已出库")
+
+
+def _is_physical_terminal(overall: str | None) -> bool:
+    """该宏观状态是否为「绝对物理终态」（不应被任务名覆写）"""
+    return (overall or "").strip() in PHYSICAL_TERMINAL_OVERALL
+
+
 async def _mark_after_sales_if_reactivated(
     db: AsyncSession, product_id: uuid.UUID,
+    steps: str | list[str | None] | None = None,
 ) -> bool:
-    """出库后又被派发新任务 = 设备回流返厂 → 生命周期切到 AFTER_SALES。
+    """出库后又接到【明确修机指令】= 设备回流返厂 → 生命周期切到 AFTER_SALES。
 
-    判定依据：产品当前处于「已出库」终态（overall_status == '已出库' 或
+    判定依据（必须同时满足）：
-    status == 'OUTBOUND'），却又产生了新的在制任务。
+      1. 产品当前处于「已出库」终态（overall_status == '已出库' 或 status == 'OUTBOUND'）
+      2. 本次接到的工序是明确修机指令（售后维修）
+
+    ⚠️ 出厂质检（发货测试）【不】触发。设备可能压根没回厂，只是补做发货前测试。
+       历史缺陷：两者混为一谈，导致「已出库 + 发货测试」被打上不可回退的
+       AFTER_SALES 烙印（售后死锁），且设备的物理终态「已出库」被工序名抹掉。
 
     该标志单向：一旦进入 AFTER_SALES 不再回退，这样前端就能把
     生产阶段的「发货测试」与回流后的「售后维修」区分开。
 
     调用时机必须早于调用方改写 overall_status，否则会漏判。
 
+    :param steps: 本次接到的工序名；转交场景可能多分支，故接受列表。
     返回是否发生了翻转。product 已在本 session 加载时 db.get 直接命中
     identity map，不产生额外查询。
     """
@@ -118,6 +154,10 @@ async def _mark_after_sales_if_reactivated(
     if product is None or product.lifecycle_phase == LIFECYCLE_AFTER_SALES:
         return False
 
+    names = [steps] if isinstance(steps, str) else [s for s in (steps or []) if s]
+    if not any(n.strip() in AFTER_SALES_REPAIR_STEPS for n in names):
+        return False
+
     is_outbound = (
         product.overall_status == "已出库"
         or (product.status or "").upper() == "OUTBOUND"
@@ -376,11 +416,18 @@ async def create_task(db: AsyncSession, data: TaskCreate) -> TaskResponse:
     product_result = await db.execute(select(Product).where(Product.id == data.product_id))
     product = product_result.scalar_one_or_none()
     if product:
-        # 🔧 出库后再次派发任务 = 设备回流返厂 → 切到售后生命周期（须早于下方改写 overall_status）
+        # 🔧 出库后又接到【明确修机指令】= 设备回流返厂 → 切到售后生命周期
-        await _mark_after_sales_if_reactivated(db, data.product_id)
+        #   （须早于下方改写 overall_status）。出厂质检（发货测试）不触发。
+        await _mark_after_sales_if_reactivated(db, data.product_id, data.task_name)
         # 🔧 选项隔离：售后回流设备禁止被排回「备货 / 生产」等前期工序（防伪造传参）
         _enforce_step_isolation(product, data.task_name, action="创建任务")
-        if data.task_name and (not data.parent_task_id or data.task_type in ("TRANSFER", "RECOVERY")):
+        # 🔒 只主线任务同步宏观状态；且【绝对物理终态保护】—— 设备已是
+        #    已入库/在库/已出库 时禁止用任务名覆写，否则「已出库」会被工序名抹掉。
+        if (
+            data.task_name
+            and (not data.parent_task_id or data.task_type in ("TRANSFER", "RECOVERY"))
+            and not _is_physical_terminal(product.overall_status)
+        ):
             product.overall_status = "已入库" if "virtual_warehouse" in data.task_name else data.task_name
         # 🔧 双字段同步：overall_status 改动后必须对齐 status，
         #    否则出库回流设备的 status 会永远停在 OUTBOUND，污染统计口径
@@ -675,17 +722,27 @@ async def receive_task(
     product_result = await db.execute(select(Product).where(Product.id == task.product_id))
     product = product_result.scalar_one_or_none()
     if product:
-        # 🔧 出库后任务被重新接收 = 设备回流返厂 → 切到售后生命周期（须早于下方改写 overall_status）
+        # 🔧 出库后任务被重新接收 = 设备回流返厂 → 切到售后生命周期
-        await _mark_after_sales_if_reactivated(db, task.product_id)
+        #   （须早于下方改写 overall_status）。出厂质检（发货测试）不触发。
+        await _mark_after_sales_if_reactivated(
+            db, task.product_id, task_name or task.task_name,
+        )
         # 🔧 选项隔离：接收时选定的工序必须落在该产品当前生命周期阶段的合法集合内
         #（task_name 为 None 时表示本次未指定工序，跳过校验，不阻断 PC 端「直接接收」）
         _enforce_step_isolation(product, task_name, action="接收任务")
         if task.assignee_id:
             product.current_location_id = task.assignee_id
-        if task_name and (
+        # 🔒 只主线任务同步宏观状态；且【绝对物理终态保护】——
-            not task.parent_task_id or task.task_type in ("TRANSFER", "RECOVERY")
+        #    设备已是 已入库/在库/已出库 时禁止用任务名覆写。
+        #    这是「发货测试 抹掉 已出库」的关键拦截点：若不拦，设备会在
+        #    概览/大屏等按终态口径统计的报表里不再是已出库，而矩阵因优先看
+        #    活跃任务看不出问题（跨报表口径就此打架）。
+        if (
+            task_name
+            and (not task.parent_task_id or task.task_type in ("TRANSFER", "RECOVERY"))
+            and not _is_physical_terminal(product.overall_status)
         ):
-            product.overall_status = task_name  # 只主线任务同步宏观状态
+            product.overall_status = task_name
         # 🔧 双字段同步：无条件对齐一次，顺带治愈历史残留
         #    （如整体已是「发货测试」而 status 还停在 OUTBOUND）
         sync_product_status(product)
@@ -861,6 +918,18 @@ async def transfer_task(
     # 权限校验：本人 或 管理员/主管 可转交
     _check_permission(task.assignee_id, operator_id, operator_role)
 
+    # ── 🏁 直接完结：额外收紧为「仅管理员/主管」──
+    #    这是【正向】拦截（非管理员一律拒绝），刻意不复用 _check_permission 的反向写法：
+    #    后者在 operator_id 为空时会静默放行，而直接完结是不可逆的收官动作
+    #    （不产生下游任务、产品落终态），必须默认拒绝。
+    #    ⚠️ 判据只能是 operator_role —— 它来自签名保护的 JWT；
+    #       operator_id 是客户端可通过 ?operator_id= 自行填写的，不可作为权限依据。
+    if request.finish_directly and not (operator_role and operator_role in ADMIN_ROLES):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="仅超管和主管有权限进行直接完结操作",
+        )
+
     # 校验：不能重复完成
     if task.status == TASK_STATUS_COMPLETED:
         raise HTTPException(
@@ -917,13 +986,6 @@ async def transfer_task(
     )
     product = product_result.scalar_one_or_none()
 
-    # 🔧 出库后又被转交出新任务 = 设备回流返厂 → 切到售后生命周期。
-    #   必须在阶段校验之前执行，否则"已出库设备被转交到生产工序"会被误放行。
-    #   ⚠️ 直接完结【不产生新任务】，不构成"回流返厂"信号，必须跳过：
-    #   否则一台已出库设备的正常收官，会把产品误翻成售后机（该标志单向不可回退）。
-    if product and not finish_directly:
-        await _mark_after_sales_if_reactivated(db, task.product_id)
-
     # 兼容新旧格式
     # 🔧 直接完结：不解析任何下家，分支强制为空，直接命中下方"无下家"兜底分支。
     if finish_directly:
@@ -943,6 +1005,17 @@ async def transfer_task(
     has_warehouse = any(a == VIRTUAL_WAREHOUSE for _, a in branches)
     real_branches = [(tn, a) for tn, a in branches if a != VIRTUAL_WAREHOUSE]
 
+    # 🔧 出库后又接到【明确修机指令】= 设备回流返厂 → 切到售后生命周期。
+    #   必须在下方阶段校验之前执行，否则"已出库设备被转交到生产工序"会被误放行。
+    #   （故本调用安排在分支解析之后、_reject_cross_phase_steps 之前。）
+    #   ⚠️ 直接完结【不产生新任务】，不构成"回流返厂"信号，必须跳过：
+    #      否则一台已出库设备的正常收官，会把产品误翻成售后机（该标志单向不可回退）。
+    #   ⚠️ 出厂质检（发货测试）同样不触发 —— 详见 _mark_after_sales_if_reactivated。
+    if product and not finish_directly:
+        await _mark_after_sales_if_reactivated(
+            db, task.product_id, [tn for tn, _ in real_branches],
+        )
+
     # 🔧 选项隔离：售后回流设备不允许被转交到「备货 / 生产 / 测试 / 维修」。
     #   转交允许自由填写工序名（喷漆/老化…），故只精准拦跨阶段词，不用白名单。
     if product:
@@ -1014,23 +1087,44 @@ async def transfer_task(
                 task_id=nt.id,
             ))
 
-    # --- 更新 Product 的 current_location_id ---
+    # --- 更新 Product 的 current_location_id / overall_status ---
+    # 🏁 直接完结【不改动 overall_status】：它只是一个「任务闭环动作」，
+    #    不改变产品的物理状态。已出库的设备完结后依然是已出库，
+    #    从而回归 WIP 矩阵的【已出库】列（无活跃任务 → 落终结态判定）。
+    #    下方 has_warehouse / real_branches 两个分支对直接完结均为假，
+    #    overall_status 自然原样保留。
     if product:
         if has_warehouse and not real_branches:
+            # 转入虚拟仓库池：位置一定改（下一手才能在这里看到并接手），
+            # 但宏观状态要分情况 ——
+            #   · 绝对物理终态（已入库 / 在库 / 已出库）：禁止改写成「待仓库收货」。
+            #     已出库设备售后返修时选择【入库】是为了「放回池子让下一个人继续转」，
+            #     改成「待仓库收货」会抹掉它的物理终态，且与 MOM 侧「已发货」的记录冲突
+            #     （MOM 认为货已出门，Track 却说「等仓库收货」，双方对不上账）。
+            #     已入库 / 在库设备本就在仓库，再入库更不该回退成「待收货」。
+            #   · 普通生产设备（备货/生产/测试/维修…）：维持原行为，正常置为「待仓库收货」。
             product.current_location_id = VIRTUAL_WAREHOUSE
-            product.overall_status = "待仓库收货"
+            if not _is_physical_terminal(product.overall_status):
+                product.overall_status = "待仓库收货"
         elif real_branches:
             product.current_location_id = real_branches[0][1]
-            if not task.parent_task_id or task.task_type in ("TRANSFER", "RECOVERY"):
+            # 🔒 只主线任务同步宏观状态；且【绝对物理终态保护】——
+            #    设备已是 已入库/在库/已出库 时禁止用任务名覆写，
+            #    否则「已出库」会被「发货测试」这类工序名抹掉。
+            if (
+                (not task.parent_task_id or task.task_type in ("TRANSFER", "RECOVERY"))
+                and not _is_physical_terminal(product.overall_status)
+            ):
                 product.overall_status = real_branches[0][0]
-        # 🔧 双字段同步：两个分支统一在这里对齐一次（原先只有入库分支硬编码
+        # 🔧 双字段同步：各分支统一在这里对齐一次（原先只有入库分支硬编码
         #    product.status="COMPLETED"，普通转交分支完全没同步 → 残留 OUTBOUND）
+        #    注意顺序：必须在上面的 overall_status 赋值【之后】，否则 status 同步到旧值。
         sync_product_status(product)
 
         # 🔧 位置回溯：如果有新任务创建，优先新任务负责人；否则回溯到父任务
         #    「直接完结」(finish_directly) 与旧版空分支都落到这里：
-        #    overall_status 在上面两个分支里都没被赋值，故原样保留
+        #    current_location_id 回溯到最后一手经手人（业务确认：位置停在经手人合理）。
-        #    （已出库 / 在库 / 售后维修… 不会被改写成「待仓库收货」）。
+        #    注意它只改 location，不碰 overall_status，故上面的终态不会被覆盖。
         if not real_branches and not has_warehouse:
             await _recalc_product_location(db, task.product_id, task_id)
 
@@ -1131,8 +1225,11 @@ async def complete_task(
     # --- 4. 可选：创建下一步任务（转交） ---
     next_task = None
     if request.next_task_name and request.next_assignee_id:
-        # 🔧 出库后又被转交出新任务 = 设备回流返厂 → 切到售后生命周期
+        # 🔧 出库后又接到【明确修机指令】= 设备回流返厂 → 切到售后生命周期
-        await _mark_after_sales_if_reactivated(db, task.product_id)
+        #   （出厂质检「发货测试」不触发）
+        await _mark_after_sales_if_reactivated(
+            db, task.product_id, request.next_task_name,
+        )
         # 🚀 智能父节点继承算法
         # 主线任务转交 → 保持平级继承（主分支永远在一维主干上）
         # 协助分支转交 → 认当前任务为父（形成向外无限延伸的孙子节点树枝）