生产任务 · 工艺步骤级派发 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 将生产任务派发粒度从「整条任务 + 单一负责人」改为「工艺步骤 + 多人负责人 + 计划开始时间」；「我的生产任务」按步骤负责人过滤；执行弹窗内仅可操作派给自己的步骤。

Architecture: 新增 biz_production_task_step_assignee（对齐 biz_qc_task_assignee）；步骤表增加 planned_start_time、completed_by；PUT /production-tasks/{id}/steps/dispatch 批量 sync 负责人；mine 鉴权从 assert_task_assignee 改为「任务级任一步 assignee OR 步骤级 assignee」+ 步骤写操作 assert_step_assignee；前端新增 DispatchStepDialog.vue，移除任务级负责人表单，隐藏任务转发/改派入口。

Tech Stack: MySQL 5.7 · FastAPI · SQLAlchemy · Pydantic v2 · Vue 3 · Element Plus · pnpm

Design Spec: docs/superpowers/specs/2026-06-25-production-step-dispatch-design.md

File Map

文件	职责
`docs/sql/migrations/2026-06-25-production-step-dispatch.sql`	步骤字段 + assignee 表
`docs/sql/migrations/2026-06-25-production-step-dispatch-backfill.sql`	可选：旧任务 task assignee → 全步骤 assignee
`docs/sql/zkxw.sql`	同步 DDL
`backend/app/models/models.py`	`BizProductionTaskStep` 新列 + `BizProductionTaskStepAssignee`
`backend/app/models/__init__.py`	export 新模型
`backend/app/crud/production_task_step_assignee.py`	新建 sync / enrich / validate
`backend/app/crud/production_task.py`	mine 过滤、dispatch、assert、completed_by、create 去 assignee 必填
`backend/app/schemas/production_task.py`	`ProductionTaskStepDispatchBody`；Create 去必填 assignee
`backend/app/api/production.py`	`PUT steps/dispatch`；`_assert_*` 改造
`web-admin/src/api/production-task.ts`	`fetchDispatchProductionTaskSteps`；create 去 assigneeUserId
`web-admin/src/views/production/production-work-order/DispatchStepDialog.vue`	新建步骤派发表格弹窗
`web-admin/src/views/production/production-work-order/DispatchTaskDialog.vue`	移除任务负责人；B 入口打开步骤派发
`web-admin/src/views/production/production-work-order/WorkOrderTaskSubList.vue`	步骤派发按钮；隐藏转发/改派；展示步骤负责人
`web-admin/src/views/production/production-work-order/TaskExecutionDialog.vue`	mine 步骤过滤 + 只读样式
`web-admin/src/views/production/production-work-order/StepDetailPanel.vue`	`canEditStep` 控制按钮
`web-admin/src/views/production/production-work-order/useTaskStepEditor.ts`	`canEditStep` / 完成写 completed_by 透传

Task 1: 数据库迁移

Files:

Create: docs/sql/migrations/2026-06-25-production-step-dispatch.sql
Create: docs/sql/migrations/2026-06-25-production-step-dispatch-backfill.sql
Modify: docs/sql/zkxw.sql（biz_production_task_step 与新建表片段）
[ ] Step 1: 编写主 migration

-- docs/sql/migrations/2026-06-25-production-step-dispatch.sql

ALTER TABLE `biz_production_task_step`
  ADD COLUMN `planned_start_time` datetime DEFAULT NULL COMMENT '计划开始时间' AFTER `step_status`;

ALTER TABLE `biz_production_task_step`
  ADD COLUMN `completed_by` varchar(64) DEFAULT NULL COMMENT '完成步骤的用户ID' AFTER `completed_time`;

CREATE TABLE IF NOT EXISTS `biz_production_task_step_assignee` (
  `id` varchar(64) NOT NULL COMMENT '主键',
  `step_id` varchar(64) NOT NULL COMMENT '生产任务步骤ID',
  `user_id` varchar(64) NOT NULL COMMENT '负责人用户ID',
  `tenant_no` varchar(64) NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `order_num` int DEFAULT 0 COMMENT '展示顺序',
  `remark` varchar(512) DEFAULT NULL COMMENT '备注',
  `create_by` varchar(64) DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL COMMENT '更新时间',
  `del_flag` tinyint NOT NULL DEFAULT 0 COMMENT '删除标记',
  PRIMARY KEY (`id`),
  KEY `idx_step_assignee_step` (`step_id`),
  KEY `idx_step_assignee_user` (`user_id`, `tenant_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='生产任务步骤负责人';

Step 2: 编写可选 backfill（旧数据兼容）

-- docs/sql/migrations/2026-06-25-production-step-dispatch-backfill.sql
-- 将仍有 task.current_assignee_user_id 且无步骤负责人的任务，回填到全部步骤

INSERT INTO biz_production_task_step_assignee (
  id, step_id, user_id, tenant_no, order_num,
  create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
  REPLACE(UUID(), '-', ''),
  s.id,
  t.current_assignee_user_id,
  s.tenant_no,
  0,
  t.create_by,
  IFNULL(t.create_dept_id, ''),
  NOW(),
  t.update_by,
  NOW(),
  0
FROM biz_production_task t
JOIN biz_production_task_step s ON s.task_id = t.id AND s.del_flag = 0
WHERE t.del_flag = 0
  AND t.current_assignee_user_id IS NOT NULL
  AND TRIM(t.current_assignee_user_id) <> ''
  AND NOT EXISTS (
    SELECT 1 FROM biz_production_task_step_assignee a
    WHERE a.step_id = s.id AND a.del_flag = 0
  );

Step 3: 在开发/测试库执行主 migration

Run:

mysql -u ... -p ... zkxw < docs/sql/migrations/2026-06-25-production-step-dispatch.sql

Expected: 两列新增成功；biz_production_task_step_assignee 表存在。

Step 4: 同步 docs/sql/zkxw.sql 对应 DDL

Task 2: Backend 模型

Files:

Modify: backend/app/models/models.py
Modify: backend/app/models/__init__.py
[ ] Step 1: 扩展 BizProductionTaskStep

在 step_status 后、started_time 前增加：

planned_start_time = Column(DateTime, nullable=True, comment="计划开始时间")

在 completed_time 后增加：

completed_by = Column(String(64), nullable=True, comment="完成步骤的用户ID")

Step 2: 新增 BizProductionTaskStepAssignee

放在 BizProductionTaskStep 附近，字段对齐 migration SQL（参照 BizQcTaskAssignee）。

Step 3: 注册 export

backend/app/models/__init__.py：

from app.models.models import (
    # ...
    BizProductionTaskStepAssignee,
)
__all__ = [
    # ...
    "BizProductionTaskStepAssignee",
]

Task 3: Backend Schema

Files:

Modify: backend/app/schemas/production_task.py
[ ] Step 1: 创建任务 body 去掉 assignee 必填

class ProductionTaskCreate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    work_order_id: str
    assignee_user_id: Optional[str] = Field(None, alias="assigneeUserId")  # deprecated，可选
    task_type: Optional[str] = 'cell_storage'
    task_name: Optional[str] = None
    route_id: Optional[str] = None
    description: Optional[str] = None
    remark: Optional[str] = None

Step 2: 新增步骤派发 body

class ProductionTaskStepDispatchItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    assignee_user_ids: List[str] = Field(..., alias="assigneeUserIds", min_length=1)
    planned_start_time: Optional[datetime] = Field(None, alias="plannedStartTime")


class ProductionTaskStepDispatchBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    steps: List[ProductionTaskStepDispatchItem] = Field(..., min_length=1)

Task 4: 步骤负责人 CRUD 模块

Files:

Create: backend/app/crud/production_task_step_assignee.py
[ ] Step 1: 创建模块（复制 QC 模式并改表名）

"""生产任务步骤负责人 CRUD"""
from __future__ import annotations

from datetime import datetime
from typing import Dict, List, Optional, Set

from sqlalchemy.orm import Session

from app.core.db_collation import eq_unicode
from app.core.utils import generate_uuid
from app.models.models import BizProductionTaskStepAssignee, SysUser

# validate_assignee_user_ids — 与 qc_task_assignee.validate_assignee_user_ids 相同逻辑


def sync_assignees_for_step(
    db: Session,
    step_id: str,
    user_ids: List[str],
    *,
    tenant_no: str,
    operator_id: Optional[str] = None,
    create_dept_id: str = "",
) -> None:
    """全量替换步骤负责人（软删旧 + 插入新）。"""
    # 实现同 qc_task_assignee.sync_assignees_for_task，表换 BizProductionTaskStepAssignee，FK 换 step_id


def enrich_step_assignees(
    db: Session, steps: List[Dict], *, tenant_no: str
) -> None:
    """批量填充 assigneeUserIds / assigneeNames / assignees。"""
    step_ids = [s["id"] for s in steps if s.get("id")]
    if not step_ids:
        return
    rows = (
        db.query(BizProductionTaskStepAssignee, SysUser)
        .outerjoin(SysUser, eq_unicode(SysUser.id, BizProductionTaskStepAssignee.user_id))
        .filter(
            BizProductionTaskStepAssignee.step_id.in_(step_ids),
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
        )
        .order_by(BizProductionTaskStepAssignee.order_num.asc())
        .all()
    )
    by_step: Dict[str, List] = {}
    for assignee, user in rows:
        by_step.setdefault(assignee.step_id, []).append((assignee, user))
    for step in steps:
        pairs = by_step.get(step["id"], [])
        step["assigneeUserIds"] = [a.user_id for a, _ in pairs]
        step["assigneeNames"] = [
            f"{(u.nick_name or u.user_name or '')}（{u.user_no or ''}）".strip()
            for _, u in pairs if u
        ]
        step["assignees"] = [
            {
                "userId": a.user_id,
                "nickName": (u.nick_name or u.user_name or "") if u else "",
                "userNo": (u.user_no or "") if u else "",
            }
            for a, u in pairs
        ]


def user_is_step_assignee(
    db: Session, step_id: str, user_id: str, tenant_no: str
) -> bool:
    uid = (user_id or "").strip()
    if not uid:
        return False
    return (
        db.query(BizProductionTaskStepAssignee.id)
        .filter(
            BizProductionTaskStepAssignee.step_id == step_id,
            BizProductionTaskStepAssignee.user_id == uid,
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
        )
        .first()
        is not None
    )


def user_is_any_step_assignee_of_task(
    db: Session, task_id: str, user_id: str, tenant_no: str
) -> bool:
    from app.models.models import BizProductionTaskStep

    uid = (user_id or "").strip()
    if not uid:
        return False
    return (
        db.query(BizProductionTaskStepAssignee.id)
        .join(
            BizProductionTaskStep,
            BizProductionTaskStep.id == BizProductionTaskStepAssignee.step_id,
        )
        .filter(
            BizProductionTaskStep.task_id == task_id,
            BizProductionTaskStep.del_flag == 0,
            BizProductionTaskStepAssignee.user_id == uid,
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
        )
        .first()
        is not None
    )

Step 2: 本地 import smoke test

Run:

cd backend && source venv/bin/activate
python -c "from app.crud import production_task_step_assignee; print('ok')"

Expected: ok

Task 5: production_task.py — 步骤 dict、派发、mine 过滤

Files:

Modify: backend/app/crud/production_task.py
[ ] Step 1: 扩展 _step_to_dict

def _step_to_dict(s: BizProductionTaskStep) -> Dict[str, Any]:
    return {
        # ... existing fields ...
        "stepStatus": s.step_status or "dzx",
        "plannedStartTime": _fmt_dt(getattr(s, "planned_start_time", None)),
        "startedTime": _fmt_dt(s.started_time),
        "completedTime": _fmt_dt(s.completed_time),
        "completedBy": getattr(s, "completed_by", None) or "",
        "completedByName": "",
        "assigneeUserIds": [],
        "assigneeNames": [],
        "assignees": [],
        "remark": s.remark or "",
    }

Step 2: create_production_task 去掉 assignee 必填

删除或注释：

assignee_raw = (data.get("assignee_user_id") or data.get("assigneeUserId") or "").strip()
assignee_user_id = _validate_assignee_user_id(db, assignee_raw, tenant_no)

改为：

assignee_raw = (data.get("assignee_user_id") or data.get("assigneeUserId") or "").strip()
assignee_user_id = assignee_raw.strip() or None
if assignee_raw.strip():
    assignee_user_id = _validate_assignee_user_id(db, assignee_raw, tenant_no)

创建任务时：

current_assignee_user_id=assignee_user_id,  # 可为 NULL

Step 3: 新增 dispatch_task_steps

def dispatch_task_steps(
    db: Session,
    task_id: str,
    *,
    tenant_no: str,
    steps_payload: List[Dict[str, Any]],
    operator_id: Optional[str] = None,
    create_dept_id: str = "",
) -> Dict[str, Any]:
    from app.crud import production_task_step_assignee as step_assignee_svc

    task = get_production_task_row(db, task_id, tenant_no)
    if not task:
        raise ValueError("任务不存在")
    if not steps_payload:
        raise ValueError("请至少指定一个步骤的负责人")

    step_rows = {
        s.id: s
        for s in db.query(BizProductionTaskStep).filter(
            BizProductionTaskStep.task_id == task_id,
            BizProductionTaskStep.tenant_no == tenant_no,
            BizProductionTaskStep.del_flag == 0,
        ).all()
    }

    for item in steps_payload:
        sid = (item.get("step_id") or item.get("stepId") or "").strip()
        if sid not in step_rows:
            raise ValueError(f"步骤不存在或不属于该任务: {sid}")
        user_ids = item.get("assignee_user_ids") or item.get("assigneeUserIds") or []
        step_assignee_svc.sync_assignees_for_step(
            db, sid, user_ids,
            tenant_no=tenant_no,
            operator_id=operator_id,
            create_dept_id=create_dept_id,
        )
        step = step_rows[sid]
        pst = item.get("planned_start_time") or item.get("plannedStartTime")
        if pst is None and "plannedStartTime" in item and item.get("plannedStartTime") is None:
            step.planned_start_time = None
        elif pst:
            from dateutil.parser import parse as _parse_dt
            step.planned_start_time = _parse_dt(str(pst)) if not isinstance(pst, datetime) else pst
        step.update_by = operator_id
        step.update_time = datetime.now()
        db.add(step)

    db.flush()
    detail = get_production_task_detail(db, task_id, tenant_no)
    if not detail:
        raise ValueError("任务不存在")
    return detail

Step 4: 改造 list_production_tasks mine 过滤

新增参数 step_assignee_user_id: Optional[str] = None。

在 current_assignee_user_id 过滤块之后增加：

sa_uid = (step_assignee_user_id or "").strip()
if sa_uid:
    step_task_ids = (
        db.query(BizProductionTaskStep.task_id)
        .join(
            BizProductionTaskStepAssignee,
            BizProductionTaskStepAssignee.step_id == BizProductionTaskStep.id,
        )
        .filter(
            BizProductionTaskStepAssignee.user_id == sa_uid,
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
            BizProductionTaskStep.del_flag == 0,
            BizProductionTask.del_flag == 0,
        )
        .distinct()
        .subquery()
    )
    q = q.filter(
        or_(
            BizProductionTask.id.in_(step_task_ids),
            BizProductionTask.current_assignee_user_id == sa_uid,  # 旧数据 fallback
        )
    )

Step 5: 改造 list_mine_production_tasks

scope=current 时调用：

return list_production_tasks(
    # ...
    step_assignee_user_id=uid,
    current_assignee_user_id=None,  # 不再仅用 task 级
    # ...
)

保留 current_assignee_user_id=uid 作为 OR 条件已在 Step 4 合并；或直接只传 step_assignee_user_id=uid 并在 SQL 内 OR task assignee fallback。

Step 6: 新增 assert 函数

def assert_task_step_assignee_or_legacy(
    db: Session, task_id: str, user_id: str, tenant_no: str
) -> BizProductionTask:
    from app.crud import production_task_step_assignee as step_assignee_svc

    task = get_production_task_row(db, task_id, tenant_no)
    if not task:
        raise ValueError("任务不存在")
    if step_assignee_svc.user_is_any_step_assignee_of_task(db, task_id, user_id, tenant_no):
        return task
    if (task.current_assignee_user_id or "") == (user_id or ""):
        return task
    raise PermissionError("非任务步骤负责人，无权操作")


def assert_step_assignee_for_write(
    db: Session,
    step_id: str,
    user_id: str,
    tenant_no: str,
    *,
    manage: bool = False,
) -> BizProductionTaskStep:
    if manage:
        step = db.query(BizProductionTaskStep).filter(
            BizProductionTaskStep.id == step_id,
            BizProductionTaskStep.tenant_no == tenant_no,
            BizProductionTaskStep.del_flag == 0,
        ).first()
        if not step:
            raise ValueError("步骤不存在")
        return step
    from app.crud import production_task_step_assignee as step_assignee_svc
    step = db.query(BizProductionTaskStep).filter(
        BizProductionTaskStep.id == step_id,
        BizProductionTaskStep.tenant_no == tenant_no,
        BizProductionTaskStep.del_flag == 0,
    ).first()
    if not step:
        raise ValueError("步骤不存在")
    if not step_assignee_svc.user_is_step_assignee(db, step_id, user_id, tenant_no):
        raise PermissionError("非当前步骤负责人，无权操作")
    return step

Step 7: update_task_step 写 completed_by

当 data.get("step_status") == "ywc" 且 user_id 存在：

step.completed_by = user_id

Step 8: get_production_task_detail enrich 步骤 assignee

加载 steps 后调用：

from app.crud.production_task_step_assignee import enrich_step_assignees
enrich_step_assignees(db, steps, tenant_no=tenant_no)
# 可选：_enrich_completed_by_names

Task 6: API 路由与鉴权

Files:

Modify: backend/app/api/production.py
[ ] Step 1: 改造 _assert_production_task_write_access

def _assert_production_task_write_access(
    db: Session,
    user: dict,
    task_id: str,
    tenant_no: str,
    operate_scope: Optional[str],
) -> None:
    scope = (operate_scope or "manage").strip().lower()
    user_id = _get_current_user_id(user)
    role = user.get("role")
    codes = get_user_menu_codes(db, user_id) if role != SUPER_ADMIN_ROLE else set()
    if scope == "mine":
        if role != SUPER_ADMIN_ROLE and "production:my-task:list" not in codes:
            raise PermissionError("无菜单访问权限")
        production_task_svc.assert_task_step_assignee_or_legacy(
            db, task_id, user_id, tenant_no
        )
    elif role != SUPER_ADMIN_ROLE and "production:work-order:list" not in codes:
        raise PermissionError("无菜单访问权限")

Step 2: 新增步骤写鉴权 helper

def _assert_production_task_step_write_access(
    db: Session,
    user: dict,
    step_id: str,
    tenant_no: str,
    operate_scope: Optional[str],
) -> None:
    scope = (operate_scope or "manage").strip().lower()
    user_id = _get_current_user_id(user)
    role = user.get("role")
    codes = get_user_menu_codes(db, user_id) if role != SUPER_ADMIN_ROLE else set()
    manage = scope != "mine"
    if not manage:
        if role != SUPER_ADMIN_ROLE and "production:my-task:list" not in codes:
            raise PermissionError("无菜单访问权限")
    elif role != SUPER_ADMIN_ROLE and "production:work-order:list" not in codes:
        raise PermissionError("无菜单访问权限")
    production_task_svc.assert_step_assignee_for_write(
        db, step_id, user_id, tenant_no, manage=manage
    )

Step 3: 注册 PUT /production-tasks/{task_id}/steps/dispatch

@router.put("/production-tasks/{task_id}/steps/dispatch")
async def dispatch_production_task_steps_api(
    task_id: str,
    body: ProductionTaskStepDispatchBody,
    user: dict = Depends(require_menu_code("production:work-order:list")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        payload = [
            {
                "step_id": it.step_id,
                "assignee_user_ids": it.assignee_user_ids,
                "planned_start_time": it.planned_start_time,
            }
            for it in body.steps
        ]
        detail = production_task_svc.dispatch_task_steps(
            db,
            task_id,
            tenant_no=tenant_no,
            steps_payload=payload,
            operator_id=user.get("user_id"),
            create_dept_id=user.get("dept_id", ""),
        )
        db.commit()
        return success_response(detail, message="步骤派发成功")
    except ValueError as e:
        db.rollback()
        return error_response(400, str(e))

Step 4: 步骤写接口增加 step 级鉴权

在以下 handler 内 _assert_production_task_write_access 之后追加 _assert_production_task_step_write_access：

update_production_task_step
replace_task_step_reagents
replace_task_step_consumables
replace_task_step_configured_reagents
replace_task_step_data_collections
replace_task_step_outputs
delete_production_task_step
copy_production_task_step

示例（update_production_task_step）：

_assert_production_task_write_access(db, user, task_id, tenant_no, operateScope)
_assert_production_task_step_write_access(db, user, step_id, tenant_no, operateScope)

任务级 inputs/outputs/attachments/status 仍只用 task 级 assert（D10：任一步 assignee 即可）。

Step 5: Swagger 手动验证

Run backend，在 /docs 测试：

PUT .../steps/dispatch 返回 steps 含 assigneeUserIds
GET .../production-tasks/mine?scope=current 仅 assignee 可见

Task 7: Frontend API

Files:

Modify: web-admin/src/api/production-task.ts
[ ] Step 1: 新增 dispatch API

export function fetchDispatchProductionTaskSteps(
  taskId: string,
  data: {
    steps: Array<{
      stepId: string
      assigneeUserIds: string[]
      plannedStartTime?: string | null
    }>
  }
) {
  return request.put<any>({
    url: `/web/admin/production-tasks/${taskId}/steps/dispatch`,
    data
  })
}

Step 2: 确认 create 不再传 assigneeUserId

DispatchTaskDialog.vue 将在 Task 9 移除该字段。

Task 8: DispatchStepDialog.vue（新建）

Files:

Create: web-admin/src/views/production/production-work-order/DispatchStepDialog.vue
[ ] Step 1: 实现弹窗骨架

要点：

Props: visible, taskId, taskName?
打开时 fetchProductionTaskDetail(taskId) 加载 steps
表格列：序号、processName、负责人多选（remote fetchUserList）、计划开始时间
本地 state：rows: { stepId, processName, assigneeUserIds, plannedStartTime, userOptions }[]
提交：过滤 assigneeUserIds.length >= 1 的行，调用 fetchDispatchProductionTaskSteps
校验：至少 1 行有负责人
UI：dialog-form-layout、label-position="top"、footer Close/Check 图标
[ ] Step 2: 回显已有 assignee

detail.steps 中若已有 assigneeUserIds / plannedStartTime，填入表单。

Task 9: DispatchTaskDialog.vue 改造

Files:

Modify: web-admin/src/views/production/production-work-order/DispatchTaskDialog.vue
[ ] Step 1: 移除任务级负责人

删除表单项 assigneeUserId、rules.assigneeUserId、form.assigneeUserId 及 create payload 中的 assigneeUserId。

Step 2: B 入口 — 创建后打开步骤派发

在 handleCreateTask 成功、enterEditPhase 之后：

emit('open-step-dispatch', { taskId, taskName: form.taskName })

或由父组件监听 @success 并根据是否选了 routeId 打开 DispatchStepDialog。

推荐：Dialog 内 emit('needStepDispatch', taskId)，父组件 WorkOrderTaskSubList / index.vue 统一挂载 DispatchStepDialog。

Step 3: 若选了 route 且步骤已克隆

创建成功 + expectSteps 为 true 时，关闭编辑阶段或保留编辑的同时弹出步骤派发（产品：B 入口）。

Task 10: WorkOrderTaskSubList.vue 改造

Files:

Modify: web-admin/src/views/production/production-work-order/WorkOrderTaskSubList.vue
[ ] Step 1: 挂载 DispatchStepDialog

<DispatchStepDialog
  v-model:visible="stepDispatchVisible"
  :task-id="stepDispatchTaskId"
  :task-name="stepDispatchTaskName"
  @success="handleTaskMutationSuccess"
/>

Step 2: 管理视角增加「步骤派发」按钮

在 !isMineMode 且任务非 yzf 时显示；文案：无 assignee 时「步骤派发」，否则「调整派发」。

判断：任一步 assigneeUserIds?.length 或后端返回 hasStepDispatch。

Step 3: 隐藏 deprecated 转发/改派

删除或 v-if="false"：

下拉「转发任务」「改派负责人」
ForwardTaskDialog 引用可保留文件但不挂载
[ ] Step 4: 任务卡片 meta 展示步骤负责人摘要

在任务名下方增加 muted 文本，例如：步骤负责人：张三、李四（细胞培养 09:00） — 取前 2 步或折叠展示。

Task 11: 执行弹窗 — 步骤权限与只读

Files:

Modify: web-admin/src/views/production/production-work-order/useTaskStepEditor.ts
Modify: web-admin/src/views/production/production-work-order/TaskExecutionDialog.vue
Modify: web-admin/src/views/production/production-work-order/StepDetailPanel.vue
[ ] Step 1: useTaskStepEditor 增加 canEditStep

import { useUserStore } from '@/store/modules/user'

const userStore = useUserStore()

function canEditStep(step: Record<string, any> | null | undefined): boolean {
  if (!step) return false
  if (operateScope.value !== 'mine') return true
  const uid = userStore.info?.id || userStore.userId || ''
  const ids: string[] = step.assigneeUserIds || []
  return ids.includes(uid)
}

function visibleStepsForMine(steps: Record<string, any>[]) {
  if (operateScope.value !== 'mine') return steps
  return steps.filter((s) => (s.assigneeUserIds || []).length > 0)
}

导出 canEditStep、visibleStepsForMine（或 computed displayStepList）。

[ ] Step 2: TaskExecutionDialog 步骤列表
mine：v-for="element in editor.displayStepList"（仅已派步骤）
对 !canEditStep(element) 的步骤加 class is-readonly（降低 opacity，禁用 drag）
任务级「转发任务」按钮移除
[ ] Step 3: StepDetailPanel 按钮

<el-button
  v-if="editor.mode.value === 'execution' && editor.canEditStep(editor.selectedStep.value)"
  ...
/>

保存 / 开始 / 完成均加 canEditStep 条件；只读时仅展示内容。

Step 4: 步骤写操作前端 guard（可选双保险）

handleStartStep / handleCompleteStep 开头：

if (!canEditStep(step)) {
  ElMessage.warning('非当前步骤负责人，无法操作')
  return
}

Task 12: 字段一致性与图谱

Files:

Run: backend/scripts/check_field_consistency.py（若涉及 CRUD 字段变更）
Run: ./scripts/update-graph.sh backend frontend
[ ] Step 1: 字段一致性检查

Run:

cd backend && source venv/bin/activate
python scripts/check_field_consistency.py

Expected: 无 production_task step 相关新增不一致；若有 _row_to_dict 变更则修复 schema/前端。

Step 2: 更新知识图谱

Run:

./scripts/update-graph.sh backend
./scripts/update-graph.sh frontend

Task 13: 手工测试清单

[ ] B 入口：生产工单 → 派发任务 → 选工艺路线 → 创建 → 步骤派发 → 用户 A/B 在「我的生产任务」见工单，执行时仅各自步骤可操作，其他只读。
[ ] A 入口：无路线创建 → 手工加 3 步 → 步骤派发 → 同上。
[ ] 一步多人：A、B 同为步骤 1 负责人；A 完成步骤 1 后 B 见只读已完成。
[ ] 同人多步：步骤 1、2 均派给 A；A 两步均可操作。
[ ] 计划开始时间：卡片/派发弹窗展示；未到时间仍可执行。
[ ] 未派步骤：mine 执行弹窗不显示该步；manage 可见可操作。
[ ] 改派：manage 调整派发后，旧 assignee mine 列表消失（若无其他步骤），新 assignee 出现。
[ ] 旧数据 backfill（若执行 backfill SQL）：原 task assignee 仍能在 mine 看到任务。
[ ] 回归：任务输入/产出/完成整单、派发质检工单链路不受影响。

Spec Coverage Self-Review

Spec 要求	对应 Task
D1 工单+任务列表	Task 5 mine 过滤不变外层结构
D2 其他步骤只读	Task 11
D3 一步多人任一完成	Task 5 Step 7 completed_by
D4 计划开始时间不阻断	Task 8/11 无前端时间校验
D5 A+B 双入口	Task 8/9/10
D6 未派步骤仅 manage	Task 11 visibleStepsForMine
D7 方案 1 数据模型	Task 1–4
D8 任务 assignee 非必填	Task 3/5
D9 转发 deprecated	Task 10
D10 任务 I/O 任一步 assignee	Task 6 task 级 assert
D11 mine 不可改派	Task 6 dispatch 仅 work-order 权限

无 TBD 占位；类型命名 assigneeUserIds / plannedStartTime 前后端一致。

建议实施顺序

Task 1 → 2 → 3 → 4 → 5 → 6 → 7 → 8 → 9 → 10 → 11 → 12 → 13
         └─ backend 可联调 ─┘       └─ frontend ─────────────┘

Backend Task 1–6 完成后即可用 Swagger 验证 mine 过滤与 dispatch；Frontend Task 7–11 依赖后端 API。

2026-06-25-production-step-dispatch.md 30 KB تاريخچه خام

生产任务 · 工艺步骤级派发 Implementation Plan

File Map

Task 1: 数据库迁移

Task 2: Backend 模型

Task 3: Backend Schema

Task 4: 步骤负责人 CRUD 模块

Task 5: production_task.py — 步骤 dict、派发、mine 过滤

Task 6: API 路由与鉴权

Task 7: Frontend API

Task 8: DispatchStepDialog.vue（新建）

Task 9: DispatchTaskDialog.vue 改造

Task 10: WorkOrderTaskSubList.vue 改造

Task 11: 执行弹窗 — 步骤权限与只读

Task 12: 字段一致性与图谱

Task 13: 手工测试清单

Spec Coverage Self-Review

建议实施顺序

2026-06-25-production-step-dispatch.md 30 KB

تاريخچه خام