docs: add implementation plan for production cascade complete

Break cascade-complete for production tasks and work orders into backend and web-admin tasks with TDD steps.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-08-31 14:34:32 +08:00
1 parent 5367abf7f2
commit 8054b2abe5
1 file changed
+657
@@ -0,0 +1,657 @@
# 生产完成任务/完成工单状态级联 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`:
```python
"""生产完成任务/工单状态级联。"""
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: 跑测确认失败**
```bash
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`:
```python
class ProductionWorkOrderComplete(BaseModel):
conclusion: str = Field(..., description="生产结论 cg=成功 sb=失败,写入本次级联完成的任务")
remark: Optional[str] = None
```
确保已导入 `Field`、`Optional`。
- [ ] **Step 4: 跑测确认通过**
同 Step 2。Expected: PASS
- [ ] **Step 5: Commit**
```bash
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 类):
```python
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: 跑测确认失败**
```bash
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` 上方新增:
```python
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 校验整段,改为:
```python
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**
```bash
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: 追加失败单测**
```python
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: 跑测确认失败**
```bash
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` 已导入):
```python
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: 跑全文件测试**
```bash
cd backend && source venv/bin/activate && pytest tests/test_production_cascade_complete.py -v
```
Expected: 全部 PASS
- [ ] **Step 5: Commit**
```bash
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: 修改函数**
```ts
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**
```bash
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 增加:
```ts
mode?: 'task' | 'workOrder' // default 'task'
hint?: string
```
```ts
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"`;表单上方:
```vue
<p v-if="hintText" class="hint-muted">{{ hintText }}</p>
```
- [ ] **Step 2: Commit**
```bash
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 与结论弹窗:
```ts
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**
```bash
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`,增加:
```ts
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
}
}
```
模板:
```vue
<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**
```bash
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: 跑图谱脚本**
```bash
./scripts/update-graph.sh all
```
- [ ] **Step 2: 有变更则提交**
```bash
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
- 自动补全业务明细
- 改变派发与步骤负责人权限