zkxw/docs/superpowers/plans/2026-08-31-production-cascade-complete.md

生产完成任务/完成工单状态级联 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: 在 production_task 中抽出 _force_complete_task_steps；update_task_status(ywc) 去掉步骤拦截并调用强制完成步骤。complete_production_work_order 校验 conclusion 后批量级联，跳过中途 sync_production_work_order_status_from_tasks，最后置工单 ywc。前端去掉完成任务的步骤/I/O/储物格拦截；完成工单改为结论弹窗，API 用请求体传参。

Tech Stack: Python 3.9 / FastAPI / SQLAlchemy / pytest·unittest；Vue 3 / TypeScript / Element Plus（web-admin）

Spec: docs/superpowers/specs/2026-08-31-production-cascade-complete-design.md

---

File map

文件	职责
Modify: backend/app/schemas/production_work_order.py	ProductionWorkOrderComplete 增加必填 conclusion
Modify: backend/app/crud/production_task.py	_force_complete_task_steps；update_task_status 去拦截并级联步骤
Modify: backend/app/crud/production_work_order.py	complete_production_work_order 级联任务/步骤
Create: backend/tests/test_production_cascade_complete.py	MagicMock 单测
Modify: web-admin/src/api/production-work-order.ts	complete 改为 body data
Modify: web-admin/src/views/production/production-work-order/CompleteProductionTaskDialog.vue	支持工单模式标题/说明
Modify: web-admin/src/views/production/production-work-order/useTaskStepEditor.ts	去掉完成任务前置校验
Modify: web-admin/src/views/production/production-work-order/index.vue	完成工单接结论弹窗
Optional: web-admin/src/views/production/my-production-task/index.vue	对齐调用（无 UI 入口）

---

Task 1: Schema — 完成工单请求增加 conclusion

Files:

• Modify: backend/app/schemas/production_work_order.py

• Create: backend/tests/test_production_cascade_complete.py

• [ ] Step 1: 写失败单测

创建 backend/tests/test_production_cascade_complete.py：

"""生产完成任务/工单状态级联。"""
import unittest
from datetime import datetime
from unittest.mock import MagicMock, patch

from pydantic import ValidationError

from app.schemas.production_work_order import ProductionWorkOrderComplete


class TestProductionWorkOrderCompleteSchema(unittest.TestCase):
    def test_requires_conclusion(self):
        with self.assertRaises(ValidationError):
            ProductionWorkOrderComplete()

    def test_accepts_cg_and_remark(self):
        body = ProductionWorkOrderComplete(conclusion="cg", remark="ok")
        self.assertEqual(body.conclusion, "cg")
        self.assertEqual(body.remark, "ok")


if __name__ == "__main__":
    unittest.main()

• Step 2: 跑测确认失败

cd backend && source venv/bin/activate && pytest tests/test_production_cascade_complete.py::TestProductionWorkOrderCompleteSchema -v

Expected: test_requires_conclusion FAIL（当前无 conclusion 字段，可无参构造）。

• Step 3: 改 Schema

在 backend/app/schemas/production_work_order.py：

class ProductionWorkOrderComplete(BaseModel):
    conclusion: str = Field(..., description="生产结论 cg=成功 sb=失败，写入本次级联完成的任务")
    remark: Optional[str] = None

确保已导入 Field、Optional。

• Step 4: 跑测确认通过

同 Step 2。Expected: PASS

• Step 5: Commit

git add backend/app/schemas/production_work_order.py backend/tests/test_production_cascade_complete.py
git commit -m "feat(backend): require conclusion on production work order complete"

---

Task 2: _force_complete_task_steps + 完成任务去拦截

Files:

• Modify: backend/app/crud/production_task.py

• Modify: backend/tests/test_production_cascade_complete.py

• [ ] Step 1: 追加失败单测

在测试文件追加（保留 Task 1 的 schema 类）：

from app.crud import production_task as production_task_svc


class TestForceCompleteTaskSteps(unittest.TestCase):
    def test_marks_incomplete_steps_ywc(self):
        db = MagicMock()
        step_dzx = MagicMock(step_status="dzx", completed_by=None, completed_time=None)
        step_jxz = MagicMock(step_status="jxz", completed_by=None, completed_time=None)
        q = MagicMock()
        db.query.return_value = q
        q.filter.return_value = q
        q.all.return_value = [step_dzx, step_jxz]
        now = datetime(2026, 8, 31, 12, 0, 0)

        production_task_svc._force_complete_task_steps(
            db, "task-1", "t1", "user-1", now
        )

        self.assertEqual(step_dzx.step_status, "ywc")
        self.assertEqual(step_dzx.completed_by, "user-1")
        self.assertEqual(step_dzx.completed_time, now)
        self.assertEqual(step_jxz.step_status, "ywc")
        db.add.assert_any_call(step_dzx)
        db.commit.assert_not_called()


class TestUpdateTaskStatusCascadeSteps(unittest.TestCase):
    @patch("app.crud.production_task._consume_task_input_cell_products")
    @patch("app.crud.production_work_order.sync_production_work_order_status_from_tasks")
    @patch("app.crud.production_task._force_complete_task_steps")
    def test_ywc_no_longer_blocks_on_incomplete_steps(
        self, mock_force, mock_sync, mock_consume
    ):
        db = MagicMock()
        task = MagicMock(
            id="task-1",
            tenant_no="t1",
            del_flag=0,
            task_status="jxz",
            work_order_id="wo-1",
            started_time=None,
            completed_time=None,
            conclusion=None,
            remark=None,
        )
        q = MagicMock()
        db.query.return_value = q
        q.filter.return_value = q
        q.first.return_value = task

        result = production_task_svc.update_task_status(
            db, "task-1", "t1", "user-1", "ywc", conclusion="cg"
        )

        mock_force.assert_called_once()
        self.assertEqual(task.task_status, "ywc")
        self.assertEqual(task.conclusion, "cg")
        mock_consume.assert_called_once()
        mock_sync.assert_called_once()
        db.commit.assert_called_once()
        self.assertIs(result, task)

• Step 2: 跑测确认失败

cd backend && source venv/bin/activate && pytest tests/test_production_cascade_complete.py::TestForceCompleteTaskSteps tests/test_production_cascade_complete.py::TestUpdateTaskStatusCascadeSteps -v

Expected: FAIL（函数未定义或仍抛步骤未完成）。

• Step 3: 实现 _force_complete_task_steps

在 production_task.py 的 update_task_status 上方新增：

def _force_complete_task_steps(
    db: Session,
    task_id: str,
    tenant_no: str,
    user_id: Optional[str],
    now: datetime,
) -> int:
    """将任务下未完成步骤置为 ywc；不 commit。返回更新条数。"""
    steps = (
        db.query(BizProductionTaskStep)
        .filter(
            and_(
                BizProductionTaskStep.task_id == task_id,
                BizProductionTaskStep.tenant_no == tenant_no,
                BizProductionTaskStep.del_flag == 0,
                BizProductionTaskStep.step_status != "ywc",
            )
        )
        .all()
    )
    for step in steps:
        step.step_status = "ywc"
        if user_id:
            step.completed_by = user_id
        step.completed_time = now
        step.update_by = user_id
        step.update_time = now
        db.add(step)
    return len(steps)

• Step 4: 改 update_task_status

删除 incomplete_step_count 校验整段，改为：

    if status == "ywc":
        concl = (conclusion or "").strip()
        if concl not in ("cg", "sb"):
            raise ValueError("须选择生产结论（成功或失败）")
        if previous_status != "ywc":
            _force_complete_task_steps(db, task_id, tenant_no, user_id, now)
        task.conclusion = concl
        if remark is not None:
            task.remark = (remark or "").strip() or None
        if previous_status != "ywc" and not task.completed_time:
            task.completed_time = now
        if previous_status != "ywc":
            _consume_task_input_cell_products(db, task.id, user_id, now)

保留末尾 sync + commit。

• Step 5: 跑测确认通过

同 Step 2。Expected: PASS

• Step 6: Commit

git add backend/app/crud/production_task.py backend/tests/test_production_cascade_complete.py
git commit -m "feat(backend): cascade-complete steps when finishing production task"

---

Task 3: 完成工单级联任务与步骤

Files:

• Modify: backend/app/crud/production_work_order.py

• Modify: backend/tests/test_production_cascade_complete.py

• [ ] Step 1: 追加失败单测

from app.crud import production_work_order as production_work_order_svc
from unittest.mock import ANY


class TestCompleteProductionWorkOrderCascade(unittest.TestCase):
    @patch("app.crud.pilot_outbound_status_sync.sync_schedule_from_production_work_order")
    @patch("app.crud.production_task._consume_task_input_cell_products")
    @patch("app.crud.production_task._force_complete_task_steps")
    @patch("app.crud.production_work_order.get_production_work_order")
    def test_cascades_incomplete_tasks_and_steps(
        self, mock_get_wo, mock_force, mock_consume, mock_sync_sched
    ):
        db = MagicMock()
        wo = MagicMock(id="wo-1", work_order_status="jxz", remark=None)
        mock_get_wo.return_value = wo

        task_a = MagicMock(
            id="ta",
            task_status="dzx",
            completed_time=None,
            conclusion=None,
            remark=None,
        )
        q = MagicMock()
        db.query.return_value = q
        q.filter.return_value = q
        q.all.return_value = [task_a]

        body = ProductionWorkOrderComplete(conclusion="sb", remark="batch fail")
        result = production_work_order_svc.complete_production_work_order(
            db, "wo-1", "t1", "user-1", body
        )

        mock_force.assert_called_once_with(db, "ta", "t1", "user-1", ANY)
        self.assertEqual(task_a.task_status, "ywc")
        self.assertEqual(task_a.conclusion, "sb")
        self.assertEqual(task_a.remark, "batch fail")
        mock_consume.assert_called_once()
        self.assertEqual(wo.work_order_status, "ywc")
        self.assertEqual(wo.remark, "batch fail")
        mock_sync_sched.assert_called_once()
        db.commit.assert_called_once()
        self.assertIs(result, wo)

    @patch("app.crud.production_work_order.get_production_work_order")
    def test_rejects_invalid_conclusion(self, mock_get_wo):
        db = MagicMock()
        mock_get_wo.return_value = MagicMock(work_order_status="jxz")
        # 绕过 pydantic：直接传 MagicMock 模拟 data
        bad = MagicMock(conclusion="xx", remark=None)
        with self.assertRaises(ValueError):
            production_work_order_svc.complete_production_work_order(
                db, "wo-1", "t1", "user-1", bad
            )

    @patch("app.crud.production_work_order.get_production_work_order")
    def test_rejects_already_completed(self, mock_get_wo):
        db = MagicMock()
        mock_get_wo.return_value = MagicMock(work_order_status="ywc")
        with self.assertRaises(ValueError) as ctx:
            production_work_order_svc.complete_production_work_order(
                db,
                "wo-1",
                "t1",
                "user-1",
                ProductionWorkOrderComplete(conclusion="cg"),
            )
        self.assertIn("已完成", str(ctx.exception))

• Step 2: 跑测确认失败

cd backend && source venv/bin/activate && pytest tests/test_production_cascade_complete.py::TestCompleteProductionWorkOrderCascade -v

Expected: FAIL

• Step 3: 重写 complete_production_work_order

替换函数体（按需确认 BizProductionTask、datetime 已导入）：

def complete_production_work_order(
    db: Session,
    row_id: str,
    tenant_no: str,
    user_id: str,
    data: Optional[ProductionWorkOrderComplete] = None,
) -> BizProductionWorkOrder:
    wo = get_production_work_order(db, row_id, tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    if wo.work_order_status == "ywc":
        raise ValueError("工单已完成")
    if not data:
        raise ValueError("须选择生产结论（成功或失败）")
    concl = (getattr(data, "conclusion", None) or "").strip()
    if concl not in ("cg", "sb"):
        raise ValueError("须选择生产结论（成功或失败）")
    remark_val = None
    if getattr(data, "remark", None) is not None:
        remark_val = (data.remark or "").strip() or None

    now = datetime.now()
    from app.crud import production_task as production_task_svc

    incomplete_tasks = (
        db.query(BizProductionTask)
        .filter(
            BizProductionTask.work_order_id == wo.id,
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
            BizProductionTask.task_status != "ywc",
        )
        .all()
    )
    for task in incomplete_tasks:
        production_task_svc._force_complete_task_steps(
            db, task.id, tenant_no, user_id, now
        )
        task.task_status = "ywc"
        task.conclusion = concl
        if remark_val is not None:
            task.remark = remark_val
        if not task.completed_time:
            task.completed_time = now
        task.update_by = user_id
        task.update_time = now
        db.add(task)
        production_task_svc._consume_task_input_cell_products(
            db, task.id, user_id, now
        )

    wo.work_order_status = "ywc"
    wo.update_by = user_id
    wo.update_time = now
    if remark_val is not None:
        wo.remark = remark_val
    db.add(wo)

    from app.crud.pilot_outbound_status_sync import sync_schedule_from_production_work_order

    sync_schedule_from_production_work_order(db, wo, user_id=user_id)
    db.commit()
    db.refresh(wo)
    return wo

关键约束：

• 不要在循环内调用 update_task_status（会 commit + 把工单 sync 成 jxz）。
• 不要调用 sync_production_work_order_status_from_tasks。

• 查询 task_status != "ywc"，已完成任务结论不动。

• [ ] Step 4: 跑全文件测试

cd backend && source venv/bin/activate && pytest tests/test_production_cascade_complete.py -v

Expected: 全部 PASS

• Step 5: Commit

git add backend/app/crud/production_work_order.py backend/tests/test_production_cascade_complete.py
git commit -m "feat(backend): cascade complete tasks and steps when finishing work order"

---

Task 4: 前端 API — complete 改请求体

Files:

• Modify: web-admin/src/api/production-work-order.ts

• [ ] Step 1: 修改函数

export function fetchCompleteProductionWorkOrder(
  id: string,
  data: { conclusion: string; remark?: string }
) {
  return request.put({
    url: `/web/admin/production-work-orders/${id}/complete`,
    data
  })
}

• Step 2: Commit

git add web-admin/src/api/production-work-order.ts
git commit -m "fix(web-admin): send production work order complete payload in body"

---

Task 5: 结论弹窗支持「完成工单」模式

Files:

• Modify: web-admin/src/views/production/production-work-order/CompleteProductionTaskDialog.vue

• [ ] Step 1: 扩展 props 与模板

Props 增加：

mode?: 'task' | 'workOrder'  // default 'task'
hint?: string

const title = computed(() => {
  if (props.mode === 'workOrder') return '完成工单'
  return props.isCompletedEdit ? '修改生产结论' : '完成任务'
})

const hintText = computed(() => {
  if (props.hint) return props.hint
  if (props.mode === 'workOrder') {
    return '将同步完成该工单下所有未完成的生产任务及其工艺步骤。'
  }
  return ''
})

模板：:title="title"；表单上方：

<p v-if="hintText" class="hint-muted">{{ hintText }}</p>

• Step 2: Commit

git add web-admin/src/views/production/production-work-order/CompleteProductionTaskDialog.vue
git commit -m "feat(web-admin): support work-order mode on production conclusion dialog"

---

Task 6: 完成任务 — 去掉前端拦截

Files:

• Modify: web-admin/src/views/production/production-work-order/useTaskStepEditor.ts

• [ ] Step 1: 精简 handleCompleteTask

去掉 validateTaskIoRowsComplete、validateAllStepsComplete、储物格 confirm。保留 flush 与结论弹窗：

  async function handleCompleteTask() {
    if (!taskDetail.value?.id || taskActionLoading.value) return
    const flushResult = await flushAllDirtyScopes({ silent: true })
    if (!flushResult.ok) return

    const isCompleted = taskDetail.value.taskStatus === 'ywc'
    completeConclusionIsEdit.value = isCompleted
    completeConclusionInitial.conclusion =
      (taskDetail.value as { conclusion?: string }).conclusion || ''
    completeConclusionInitial.remark = taskDetail.value.remark || ''
    completeConclusionVisible.value = true
  }

用 grep 确认 validateAllStepsComplete / validateTaskIoRowsComplete 无其它引用后可删除函数定义。

• Step 2: Commit

git add web-admin/src/views/production/production-work-order/useTaskStepEditor.ts
git commit -m "fix(web-admin): allow finish production task without step or IO gates"

---

Task 7: 生产工单页 — 完成工单接结论弹窗

Files:

• Modify: web-admin/src/views/production/production-work-order/index.vue

• Optional: web-admin/src/views/production/my-production-task/index.vue

• [ ] Step 1: 接线 index.vue

引入 CompleteProductionTaskDialog，增加：

const completeWoVisible = ref(false)
const completeWoSubmitting = ref(false)
const completeWoTarget = ref<ProductionWorkOrderItem | null>(null)

async function handleCompleteSingle(row: ProductionWorkOrderItem) {
  if (row.workOrderStatus !== 'dpz' && row.workOrderStatus !== 'jxz') {
    ElMessage.warning('仅待配置或进行中的工单可完成')
    return
  }
  completeWoTarget.value = row
  completeWoVisible.value = true
}

async function submitCompleteWorkOrder(payload: { conclusion: string; remark?: string }) {
  const row = completeWoTarget.value
  if (!row || completeWoSubmitting.value) return
  completeWoSubmitting.value = true
  try {
    await fetchCompleteProductionWorkOrder(row.id, {
      conclusion: payload.conclusion,
      remark: payload.remark
    })
    ElMessage.success('工单已完成')
    completeWoVisible.value = false
    completeWoTarget.value = null
    await refreshActiveTab()
  } catch {
    /* global error handling */
  } finally {
    completeWoSubmitting.value = false
  }
}

模板：

<CompleteProductionTaskDialog
  v-model:visible="completeWoVisible"
  mode="workOrder"
  :submitting="completeWoSubmitting"
  @submit="submitCompleteWorkOrder"
/>

移除该流程的 ElMessageBox.confirm。

• Step 2: 可选对齐 my-production-task

复制同一弹窗接线，或将旧 handleCompleteSingle 改为提示「请在生产工单管理页完成工单」。推荐复制接线以保持路径一致（卡片仍不展示按钮）。

• Step 3: 手工回归

1. 未完成步骤 → 完成任务选成功 → 步骤与任务均 ywc。
2. 完成工单选失败 → 未完成任务 ywc+sb、步骤 ywc、工单 ywc；已完成任务结论不变。
3. 单步执行/完成步骤仍可用。
4. 我的生产任务无「完成工单」；完成任务行为同 1。

• Step 4: Commit

git add web-admin/src/views/production/production-work-order/index.vue \
  web-admin/src/views/production/my-production-task/index.vue
git commit -m "feat(web-admin): cascade-complete work order via conclusion dialog"

---

Task 8: 更新知识图谱

• Step 1: 跑图谱脚本

./scripts/update-graph.sh all

• Step 2: 有变更则提交

git add backend/graphify-out web-admin/graphify-out
git commit -m "chore: update graphify after production cascade complete"

无变更则跳过。

---

Spec coverage（自审）

Spec 要求	Task
完成任务去步骤拦截 + 强制完成步骤	Task 2
完成任务去 I/O/储物格前端拦截	Task 6
完成工单 conclusion + 级联	Task 1, 3, 5, 7
步骤完成人/时间 = 当前用户/现在	Task 2
含 dpz/dzx/jxz；已完成不动	Task 3
同一事务、不中途 sync jxz	Task 3
API body	Task 4
我的生产任务不新增完成工单入口	Task 7（showCardComplete 仍限制）
单步按钮保留	未改 StepDetailActionBar

非目标

• 新建独立 force-complete API
• 自动补全业务明细
• 改变派发与步骤负责人权限