docs(oa): add auto-pass-when-approver-is-initiator implementation plan

Break the approved design into TDD tasks for engine core, tests, and designer switch.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-07-27 17:37:07 +08:00
1 parent 9b76aa654e
commit b9193b3175
1 file changed
+699
@@ -0,0 +1,699 @@
# OA 审批人为发起人时自动通过 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:** 在 OA 应用高级设置增加默认关闭的开关;开启后,审批节点建待办若含发起人本人,则系统代为同意并推进(或签整节点过、会签仅本人、可连续自动通过),进度意见为「系统自动通过」。
**Architecture:** 在 `DEFAULT_ADVANCED_SETTINGS` 增加 `autoPassWhenApproverIsInitiator`。从 `complete_task` 抽出无 commit 的同意收口 `_approve_task_core`;人工同意仍走校验 + commit。在 `advance_from` 对审批节点 `_create_tasks_for_node` 之后调用 `_maybe_auto_pass_initiator`:开关开且存在本人 pending 时调用核心(跳过 `commentRequired`,意见固定「系统自动通过」),或签/会签收口与现网一致;推进时嵌套 `advance_from` 自然连续自动通过。设计器高级设置增加对应开关。
**Tech Stack:** Python 3.9 / FastAPI OA 引擎、SQLAlchemy、unittest + SQLite in-memory;Vue 3 + Element Plus(`web-admin` 设计器)。
**Spec:** `docs/superpowers/specs/2026-07-27-oa-auto-pass-initiator-approver-design.md`
---
## File map
| 文件 | 职责 |
|------|------|
| `backend/app/crud/oa_approval.py` | 默认高级设置增加键 |
| `backend/app/services/oa_engine.py` | 抽 `_approve_task_core`;`_maybe_auto_pass_initiator`;`advance_from` 挂钩;`complete_task` 改调核心 |
| `backend/tests/test_oa_auto_pass_initiator.py` | 新测:开关关/开、或签、会签、连续、意见必填 |
| `web-admin/src/views/oa-approval/designer/index.vue` | 高级设置 UI + reactive / loadDraft |
---
### Task 1: 默认设置键 + 失败用例骨架
**Files:**
- Modify: `backend/app/crud/oa_approval.py`(`DEFAULT_ADVANCED_SETTINGS`)
- Create: `backend/tests/test_oa_auto_pass_initiator.py`
- [ ] **Step 1: 写入默认键**
在 `DEFAULT_ADVANCED_SETTINGS` 增加一行:
```python
DEFAULT_ADVANCED_SETTINGS: Dict[str, Any] = {
"supervisorFallbackToDeptLeader": True,
"allowWithdrawRunning": True,
"approverDedup": False,
"commentRequired": False,
"allowAddSign": True,
"allowReturn": True,
"autoPassWhenApproverIsInitiator": False,
}
```
- [ ] **Step 2: 写失败测试(开关开 + 或签单人=发起人,期望自动通过;当前应 FAIL)**
创建 `backend/tests/test_oa_auto_pass_initiator.py`:
```python
"""OA:审批人为发起人时自动通过"""
import json
import unittest
from datetime import datetime
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class TestAutoPassInitiator(unittest.TestCase):
def setUp(self):
from app.models.models import OaAppVersion, OaInstance, OaTask
self.engine = create_engine("sqlite:///:memory:")
OaAppVersion.__table__.create(self.engine)
OaInstance.__table__.create(self.engine)
OaTask.__table__.create(self.engine)
Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
self.db = Session()
self.OaAppVersion = OaAppVersion
self.OaInstance = OaInstance
self.OaTask = OaTask
self.now = datetime.now()
def tearDown(self):
self.db.close()
self.engine.dispose()
def _seed(self, *, flow, advanced, initiator="u0", tasks=None):
self.db.add(
self.OaAppVersion(
id="v1",
app_id="app1",
version_no=1,
status="published",
is_latest_published="1",
form_schema_json="{}",
flow_tree_json=json.dumps(flow),
advanced_settings_json=json.dumps(advanced),
tenant_no="zkxw",
create_dept_id="",
create_time=self.now,
update_time=self.now,
del_flag=0,
)
)
self.db.add(
self.OaInstance(
id="i1",
app_id="app1",
app_version_id="v1",
title="t",
initiator_user_id=initiator,
initiator_dept_id="d1",
form_data_json="{}",
status="running",
tenant_no="zkxw",
create_dept_id="",
create_time=self.now,
update_time=self.now,
del_flag=0,
)
)
for t in tasks or []:
self.db.add(t)
self.db.commit()
def test_or_sign_sole_initiator_auto_passes_to_end(self):
"""开关开:或签唯一审批人=发起人 → 提交推进后自动通过到结束。"""
from app.services.oa_engine import advance_from
flow = {
"id": "s",
"type": "start",
"child": {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0"],
"signMode": "or",
},
"child": {"id": "e1", "type": "end"},
},
}
self._seed(
flow=flow,
advanced={"autoPassWhenApproverIsInitiator": True},
)
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u0"], {"resolve_source": "users"}),
), patch(
"app.services.oa_engine._next_task_order_num",
return_value=1,
):
advance_from(
self.db, inst=inst, ver=ver, from_node=None, user_id="u0"
)
tasks = self.db.query(self.OaTask).filter_by(instance_id="i1").all()
self.assertEqual(len(tasks), 1)
self.assertEqual(tasks[0].status, "approved")
self.assertEqual(tasks[0].comment_text, "系统自动通过")
self.assertEqual(inst.status, "completed")
if __name__ == "__main__":
unittest.main()
```
- [ ] **Step 3: 跑测确认失败**
Run:
```bash
cd backend && source venv/bin/activate && pytest tests/test_oa_auto_pass_initiator.py::TestAutoPassInitiator::test_or_sign_sole_initiator_auto_passes_to_end -v
```
Expected: FAIL(任务仍 `pending` 或实例仍 `running`,因尚未实现自动通过)
- [ ] **Step 4: Commit**
```bash
git add backend/app/crud/oa_approval.py backend/tests/test_oa_auto_pass_initiator.py
git commit -m "test(oa): add failing auto-pass-when-approver-is-initiator case"
```
---
### Task 2: 抽取 `_approve_task_core` 并实现自动通过挂钩
**Files:**
- Modify: `backend/app/services/oa_engine.py`
- Test: `backend/tests/test_oa_auto_pass_initiator.py`
- [ ] **Step 1: 在 `oa_engine.py` 顶部常量附近增加固定意见文案**
```python
_AUTO_PASS_COMMENT = "系统自动通过"
```
- [ ] **Step 2: 从 `complete_task` 抽出同意收口(拒绝逻辑可留在 `complete_task`)**
在 `complete_task` 上方新增(签名与行为对齐现网同意分支;**不** `commit`、**不**校验 `commentRequired`):
```python
def _approve_task_core(
db: Session,
*,
task: OaTask,
inst: OaInstance,
ver: OaAppVersion,
user_id: str,
comment: Optional[str],
) -> bool:
"""将 pending 审批任务置为 approved 并按或签/会签决定是否推进。
返回 True 表示已调用 advance_from 推进;False 表示会签未齐、未推进。
调用方负责 commit(若需要)。
"""
now = datetime.now()
task.comment_text = (comment or "").strip() or None
task.status = "approved"
task.update_by = user_id
task.update_time = now
db.add(task)
db.flush()
sign_mode = (task.sign_mode or "or").strip() or "or"
pending_same = (
db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.node_key == task.node_key,
OaTask.task_type == "approval",
OaTask.id != task.id,
OaTask.status == "pending",
OaTask.del_flag == 0,
)
.count()
)
if sign_mode == "and" and pending_same > 0:
return False
if sign_mode != "and":
for s in (
db.query(OaTask)
.filter(
OaTask.instance_id == inst.id,
OaTask.node_key == task.node_key,
OaTask.id != task.id,
OaTask.status == "pending",
OaTask.del_flag == 0,
)
.all()
):
s.status = "done"
s.update_by = user_id
s.update_time = now
db.add(s)
root = _flow_root(_loads(ver.flow_tree_json, {}))
node = _find_node(root, task.node_key or "")
if not node:
raise OaEngineError("找不到对应流程节点")
advance_from(
db,
inst=inst,
ver=ver,
from_node=node,
user_id=user_id,
initiator_select_by_node=_load_initiator_select_by_node(inst),
)
return True
```
将 `complete_task` 的 approve 分支改为:校验 `commentRequired` 后调用 `_approve_task_core(...)`,再 `db.commit()` 并 `return get_instance_detail(...)`。拒绝分支保持原样。
- [ ] **Step 3: 实现 `_maybe_auto_pass_initiator`**
```python
def _maybe_auto_pass_initiator(
db: Session,
*,
inst: OaInstance,
ver: OaAppVersion,
tasks: List[OaTask],
user_id: Optional[str],
) -> None:
advanced = _advanced(ver)
if not advanced.get("autoPassWhenApproverIsInitiator"):
return
if inst.status != "running":
return
initiator = (inst.initiator_user_id or "").strip()
if not initiator:
return
self_task = None
for t in tasks:
if (
t.task_type == "approval"
and t.status == "pending"
and (t.assignee_user_id or "") == initiator
):
self_task = t
break
if not self_task:
return
_approve_task_core(
db,
task=self_task,
inst=inst,
ver=ver,
user_id=initiator,
comment=_AUTO_PASS_COMMENT,
)
```
- [ ] **Step 4: 在 `advance_from` 审批节点建待办后挂钩**
将审批分支从:
```python
if not tasks:
raise OaEngineError("审批节点未生成任务")
return
```
改为:
```python
if not tasks:
raise OaEngineError("审批节点未生成任务")
_maybe_auto_pass_initiator(
db,
inst=inst,
ver=ver,
tasks=tasks,
user_id=user_id,
)
return
```
说明:若自动通过触发嵌套 `advance_from`,下一节点同样会挂钩,从而实现连续自动通过。会签未齐时核心返回 False,外层 `return` 后实例停在当前节点。
- [ ] **Step 5: 跑 Task 1 单测应通过**
```bash
cd backend && source venv/bin/activate && pytest tests/test_oa_auto_pass_initiator.py::TestAutoPassInitiator::test_or_sign_sole_initiator_auto_passes_to_end -v
```
Expected: PASS
- [ ] **Step 6: 跑现有同意/或签回归**
```bash
cd backend && source venv/bin/activate && pytest tests/test_oa_engine.py::TestCompleteTaskOrSign -v
```
Expected: PASS
- [ ] **Step 7: Commit**
```bash
git add backend/app/services/oa_engine.py backend/tests/test_oa_auto_pass_initiator.py
git commit -m "feat(oa): auto-pass approval when assignee is initiator"
```
---
### Task 3: 补齐规格场景单测
**Files:**
- Modify: `backend/tests/test_oa_auto_pass_initiator.py`
- [ ] **Step 1: 追加以下测试方法(沿用 `setUp` / `_seed`)**
```python
def test_switch_off_keeps_pending(self):
from app.services.oa_engine import advance_from
flow = {
"id": "s",
"type": "start",
"child": {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0"],
"signMode": "or",
},
"child": {"id": "e1", "type": "end"},
},
}
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": False})
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u0"], {"resolve_source": "users"}),
), patch(
"app.services.oa_engine._next_task_order_num", return_value=1
):
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
t = self.db.query(self.OaTask).filter_by(instance_id="i1").one()
self.assertEqual(t.status, "pending")
self.assertEqual(inst.status, "running")
def test_or_sign_multi_cancels_siblings(self):
from app.services.oa_engine import advance_from
flow = {
"id": "s",
"type": "start",
"child": {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0", "u2"],
"signMode": "or",
},
"child": {"id": "e1", "type": "end"},
},
}
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u0", "u2"], {"resolve_source": "users"}),
), patch(
"app.services.oa_engine._next_task_order_num", return_value=1
):
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
by_user = {
t.assignee_user_id: t
for t in self.db.query(self.OaTask).filter_by(instance_id="i1").all()
}
self.assertEqual(by_user["u0"].status, "approved")
self.assertEqual(by_user["u0"].comment_text, "系统自动通过")
self.assertEqual(by_user["u2"].status, "done")
self.assertEqual(inst.status, "completed")
def test_and_sign_only_completes_self(self):
from app.services.oa_engine import advance_from
flow = {
"id": "s",
"type": "start",
"child": {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0", "u2"],
"signMode": "and",
},
"child": {"id": "e1", "type": "end"},
},
}
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u0", "u2"], {"resolve_source": "users"}),
), patch(
"app.services.oa_engine._next_task_order_num", return_value=1
):
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
by_user = {
t.assignee_user_id: t
for t in self.db.query(self.OaTask).filter_by(instance_id="i1").all()
}
self.assertEqual(by_user["u0"].status, "approved")
self.assertEqual(by_user["u2"].status, "pending")
self.assertEqual(inst.status, "running")
def test_consecutive_nodes_both_auto_pass(self):
from app.services.oa_engine import advance_from
flow = {
"id": "s",
"type": "start",
"child": {
"id": "a1",
"type": "approval",
"name": "一审",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0"],
"signMode": "or",
},
"child": {
"id": "a2",
"type": "approval",
"name": "二审",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0"],
"signMode": "or",
},
"child": {"id": "e1", "type": "end"},
},
},
}
self._seed(flow=flow, advanced={"autoPassWhenApproverIsInitiator": True})
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u0"], {"resolve_source": "users"}),
), patch(
"app.services.oa_engine._next_task_order_num", return_value=1
):
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
tasks = self.db.query(self.OaTask).filter_by(instance_id="i1").all()
self.assertEqual(len(tasks), 2)
self.assertTrue(all(t.status == "approved" for t in tasks))
self.assertTrue(all(t.comment_text == "系统自动通过" for t in tasks))
self.assertEqual(inst.status, "completed")
def test_comment_required_does_not_block_auto_pass(self):
from app.services.oa_engine import advance_from
flow = {
"id": "s",
"type": "start",
"child": {
"id": "a1",
"type": "approval",
"name": "审批人",
"approverConfig": {
"assigneeType": "users",
"userIds": ["u0"],
"signMode": "or",
},
"child": {"id": "e1", "type": "end"},
},
}
self._seed(
flow=flow,
advanced={
"autoPassWhenApproverIsInitiator": True,
"commentRequired": True,
},
)
inst = self.db.query(self.OaInstance).filter_by(id="i1").first()
ver = self.db.query(self.OaAppVersion).filter_by(id="v1").first()
with patch(
"app.services.oa_engine.resolve_assignees",
return_value=(["u0"], {"resolve_source": "users"}),
), patch(
"app.services.oa_engine._next_task_order_num", return_value=1
):
advance_from(self.db, inst=inst, ver=ver, from_node=None, user_id="u0")
t = self.db.query(self.OaTask).filter_by(instance_id="i1").one()
self.assertEqual(t.status, "approved")
self.assertEqual(inst.status, "completed")
```
- [ ] **Step 2: 跑全文件**
```bash
cd backend && source venv/bin/activate && pytest tests/test_oa_auto_pass_initiator.py -v
```
Expected: 全部 PASS
- [ ] **Step 3: 跑 advance_from 相关旧测(mock 不建真实 assignee 时不应误触发)**
```bash
cd backend && source venv/bin/activate && pytest tests/test_oa_engine.py::TestAdvanceFrom -v
```
Expected: PASS(mock 的 `fake_task` 无 `assignee_user_id==initiator` 匹配或开关默认关;若 FAIL,给 mock task 设 `assignee_user_id="u1"` 且 `task_type="approval"`、`status="pending"`,或 advanced 显式 `autoPassWhenApproverIsInitiator: False`)
- [ ] **Step 4: Commit**
```bash
git add backend/tests/test_oa_auto_pass_initiator.py backend/tests/test_oa_engine.py
git commit -m "test(oa): cover auto-pass or/and/consecutive/commentRequired cases"
```
---
### Task 4: 设计器高级设置 UI
**Files:**
- Modify: `web-admin/src/views/oa-approval/designer/index.vue`
- [ ] **Step 1: reactive 默认值增加字段**
```typescript
const advanced = reactive({
supervisorFallbackToDeptLeader: true,
allowWithdrawRunning: true,
commentRequired: false,
allowAddSign: true,
allowReturn: true,
approverDedup: false,
autoPassWhenApproverIsInitiator: false
})
```
- [ ] **Step 2: `loadDraft` 同步**
在 `advanced.approverDedup = !!a.approverDedup` 后增加:
```typescript
advanced.autoPassWhenApproverIsInitiator = !!a.autoPassWhenApproverIsInitiator
```
- [ ] **Step 3: 模板增加开关(放在「审批人去重」之前或之后均可;推荐去重之前,语义更清晰)**
```vue
<el-form-item label="审批人为发起人时自动通过">
<el-switch v-model="advanced.autoPassWhenApproverIsInitiator" />
<div class="oa-designer__hint">
开启后,解析到的审批人含发起人本人时,系统代为同意并推进;或签整节点通过,会签仅完成本人;可连续自动通过多个节点。默认关闭。
</div>
</el-form-item>
```
若文件尚无 `.oa-designer__hint` 样式,在同文件 `<style scoped>` 增加:
```scss
.oa-designer__hint {
margin-top: 6px;
font-size: 12px;
line-height: 1.5;
color: var(--mint-text-muted, #64748b);
}
```
`saveDraft` 已用 `advancedSettingsJson: { ...advanced }`,无需另改。
- [ ] **Step 4: 手动核对(无前端单测时)**
1. 打开任一 OA 应用设计器 → 高级设置,见新开关默认关。
2. 打开开关 → 保存草稿 → 刷新,仍为开。
3. 发布后新发起:第一步审批人=自己时进度见「系统自动通过」。
- [ ] **Step 5: Commit**
```bash
git add web-admin/src/views/oa-approval/designer/index.vue
git commit -m "feat(oa): add auto-pass-when-approver-is-initiator advanced switch"
```
---
### Task 5: 全量回归与收尾
- [ ] **Step 1: 跑 OA 相关后端测**
```bash
cd backend && source venv/bin/activate && pytest tests/test_oa_auto_pass_initiator.py tests/test_oa_engine.py tests/test_oa_assignee.py tests/test_oa_initiator_select_by_node.py -v
```
Expected: PASS
- [ ] **Step 2: 若改了 backend/web-admin 代码结构,按仓库约定增量更新图谱**
```bash
./scripts/update-graph.sh backend
```
(仅前端 Task 4 时也可 `./scripts/update-graph.sh frontend`;两边都改则 `all`。)
- [ ] **Step 3: 最终确认无未提交实现**
```bash
git status -sb
```
---
## Spec coverage checklist
| Spec 要求 | Task |
|-----------|------|
| `autoPassWhenApproverIsInitiator` 默认 false | Task 1 |
| 抽同意内核 + 建待办后挂钩 | Task 2 |
| 或签整过 / 会签只过本人 / 连续 / 跳过意见必填 | Task 2–3 |
| 意见「系统自动通过」 | Task 2–3 |
| 设计器开关 + tip | Task 4 |
| 不加独立 event / 预览旁注 / 不去重引擎 | 刻意不做 |
| 加签不触发整节点再自动过 | 挂钩仅在 `advance_from` 新建审批任务后;加签不走该路径 |
## Placeholder / consistency review
- 键名统一:`autoPassWhenApproverIsInitiator`
- 意见文案统一:`系统自动通过` / `_AUTO_PASS_COMMENT`
- `_approve_task_core` 不 commit;`complete_task` 负责 commit
- 无 TBD/TODO 占位