diff --git a/backend/app/api/oa_approval.py b/backend/app/api/oa_approval.py
index 39133d67..5e8669a1 100644
--- a/backend/app/api/oa_approval.py
+++ b/backend/app/api/oa_approval.py
@@ -550,7 +550,8 @@ async def preview_oa_template_flow(
             app_id=app_id,
             initiator_user_id=uid,
             initiator_dept_id=dept_id,
-            initiator_select_user_ids=body.initiator_select_user_ids,
+            # Task 3 将 Body 改为 initiatorSelectByNode；此处先对齐引擎 kwargs
+            initiator_select_by_node=None,
             form_data=body.form_data,
         )
         return success_response(data)
@@ -576,16 +577,14 @@ async def preview_oa_template_flow_get(
             row = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first()
             if row:
                 dept_id = row.dept_id
-        select_ids = None
-        if initiator_select_user_ids:
-            select_ids = [x.strip() for x in initiator_select_user_ids.split(",") if x.strip()]
+        _ = initiator_select_user_ids  # 扁平 query 已废弃；Task 3 删除
         data = preview_start_flow(
             db,
             tenant_no=tenant_no,
             app_id=app_id,
             initiator_user_id=uid,
             initiator_dept_id=dept_id,
-            initiator_select_user_ids=select_ids,
+            initiator_select_by_node=None,
             form_data=None,
         )
         return success_response(data)
@@ -611,7 +610,8 @@ async def start_oa_instance(
             initiator_dept_id=body.initiator_dept_id,
             form_data=body.form_data,
             title=body.title,
-            initiator_select_user_ids=body.initiator_select_user_ids,
+            # Task 3 将 Body 改为 initiatorSelectByNode
+            initiator_select_by_node=None,
             related_instance_ids=body.related_instance_ids,
             source_instance_id=body.source_instance_id,
         )
@@ -788,7 +788,7 @@ async def resubmit_oa_instance(
             tenant_no=tenant_no,
             user_id=user.get("user_id") or "",
             form_data=(body.form_data if body else None),
-            initiator_select_user_ids=(body.initiator_select_user_ids if body else None),
+            initiator_select_by_node=None,
         )
         return success_response(data, message="已再次提交")
     except OaEngineError as e:
diff --git a/backend/app/services/oa_engine.py b/backend/app/services/oa_engine.py
index 988e91bd..e579fcab 100644
--- a/backend/app/services/oa_engine.py
+++ b/backend/app/services/oa_engine.py
@@ -44,6 +44,41 @@ def _dumps(data: Any) -> str:
     return json.dumps(data if data is not None else {}, ensure_ascii=False)
 
 
+def _normalize_initiator_select_by_node(raw: Any) -> Dict[str, List[str]]:
+    if not isinstance(raw, dict):
+        return {}
+    out: Dict[str, List[str]] = {}
+    for k, v in raw.items():
+        nid = str(k or "").strip()
+        if not nid or not isinstance(v, (list, tuple)):
+            continue
+        ids: List[str] = []
+        seen = set()
+        for x in v:
+            uid = str(x or "").strip()
+            if uid and uid not in seen:
+                seen.add(uid)
+                ids.append(uid)
+        out[nid] = ids
+    return out
+
+
+def _ids_for_initiator_select_node(
+    select_by_node: Optional[Dict[str, List[str]]], node_id: str
+) -> List[str]:
+    m = select_by_node or {}
+    nid = str(node_id or "").strip()
+    if not nid:
+        return []
+    return list(m.get(nid) or [])
+
+
+def _load_initiator_select_by_node(inst: Any) -> Dict[str, List[str]]:
+    return _normalize_initiator_select_by_node(
+        _loads(getattr(inst, "initiator_select_json", None), {}) or {}
+    )
+
+
 def _node_type(node: Dict[str, Any]) -> str:
     return (node.get("type") or "").strip()
 
@@ -253,7 +288,7 @@ def _pending_people_for_node(
     initiator_user_id: str,
     initiator_dept_id: Optional[str],
     advanced: Dict[str, Any],
-    initiator_select_user_ids: Optional[List[str]],
+    initiator_select_by_node: Optional[Dict[str, List[str]]],
 ) -> Tuple[List[Dict[str, Any]], str, str]:
     """解析未到达节点的展示人员与签批模式。返回 (people, sign_mode, resolve_source)。"""
     ntype = _node_type(node)
@@ -265,9 +300,11 @@ def _pending_people_for_node(
     elif not sign_mode and assignee_type != "initiator_select":
         sign_mode = "or"
     people: List[Dict[str, Any]] = []
+    node_select_ids = _ids_for_initiator_select_node(
+        initiator_select_by_node, node.get("id") or ""
+    )
     if assignee_type == "initiator_select":
-        ids = [str(x).strip() for x in (initiator_select_user_ids or []) if str(x).strip()]
-        for uid in ids:
+        for uid in node_select_ids:
             people.append(
                 {
                     "id": uid,
@@ -287,7 +324,7 @@ def _pending_people_for_node(
                 initiator_user_id=initiator_user_id,
                 initiator_dept_id=initiator_dept_id,
                 advanced=advanced,
-                initiator_select_user_ids=initiator_select_user_ids,
+                initiator_select_user_ids=node_select_ids,
             )
             for uid in assignees:
                 people.append(
@@ -352,7 +389,7 @@ def _append_pending_nodes_on_path(
     initiator_dept_id: Optional[str],
     advanced: Dict[str, Any],
     form_data: Optional[Dict[str, Any]],
-    initiator_select_user_ids: Optional[List[str]],
+    initiator_select_by_node: Optional[Dict[str, List[str]]],
 ) -> None:
     """按命中路径列出全部后续审批/抄送/结束节点（含人员），对齐发起预览。"""
     path = _iter_resolved_path_nodes(flow_root, form_data)
@@ -402,7 +439,7 @@ def _append_pending_nodes_on_path(
             initiator_user_id=initiator_user_id,
             initiator_dept_id=initiator_dept_id,
             advanced=advanced,
-            initiator_select_user_ids=initiator_select_user_ids,
+            initiator_select_by_node=initiator_select_by_node,
         )
         items.append(
             {
@@ -471,7 +508,7 @@ def build_progress_items(
     events: Optional[List[Any]] = None,
     db: Optional[Session] = None,
     advanced: Optional[Dict[str, Any]] = None,
-    initiator_select_user_ids: Optional[List[str]] = None,
+    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
 ) -> List[Dict[str, Any]]:
     """投影审批进度：发起人 → 树序 tasks（同节点多人合并）→ 未到达；终态收尾。
 
@@ -560,14 +597,19 @@ def build_progress_items(
         form_data = _loads(getattr(inst, "form_data_json", None), {}) or {}
         if not isinstance(form_data, dict):
             form_data = {}
-        # 自选人：优先入参；否则从已有任务里收集
-        select_ids = list(initiator_select_user_ids or [])
-        if not select_ids:
+        # 自选人：优先入参 → 实例 JSON → 旧数据从任务按 node_key 聚合
+        select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
+        if not select_by_node:
+            select_by_node = _load_initiator_select_by_node(inst)
+        if not select_by_node:
             for t in sorted_tasks:
                 src = (getattr(t, "resolve_source", None) or "").strip()
                 uid = (getattr(t, "assignee_user_id", None) or "").strip()
-                if src == "initiator_select" and uid and uid not in select_ids:
-                    select_ids.append(uid)
+                nk = (getattr(t, "node_key", None) or "").strip()
+                if src == "initiator_select" and uid and nk:
+                    bucket = select_by_node.setdefault(nk, [])
+                    if uid not in bucket:
+                        bucket.append(uid)
         if db is not None:
             _append_pending_nodes_on_path(
                 items,
@@ -580,7 +622,7 @@ def build_progress_items(
                 initiator_dept_id=getattr(inst, "initiator_dept_id", None),
                 advanced=advanced or dict(oa_crud.DEFAULT_ADVANCED_SETTINGS),
                 form_data=form_data,
-                initiator_select_user_ids=select_ids,
+                initiator_select_by_node=select_by_node,
             )
         else:
             _append_pending_nodes(
@@ -1023,7 +1065,7 @@ def _walk_preview_nodes(
     initiator_user_id: str,
     initiator_dept_id: Optional[str],
     advanced: Dict[str, Any],
-    initiator_select_user_ids: Optional[List[str]],
+    initiator_select_by_node: Optional[Dict[str, List[str]]],
     form_data: Optional[Dict[str, Any]],
     steps: List[Dict[str, Any]],
 ) -> bool:
@@ -1078,7 +1120,7 @@ def _walk_preview_nodes(
                     initiator_user_id=initiator_user_id,
                     initiator_dept_id=initiator_dept_id,
                     advanced=advanced,
-                    initiator_select_user_ids=initiator_select_user_ids,
+                    initiator_select_by_node=initiator_select_by_node,
                     form_data=data,
                     steps=steps,
                 )
@@ -1098,9 +1140,11 @@ def _walk_preview_nodes(
             meta = _assignee_meta(db, tenant_no, cfg)
             people: List[Dict[str, str]] = []
             resolve_error = ""
+            node_select_ids = _ids_for_initiator_select_node(
+                initiator_select_by_node, cur.get("id") or ""
+            )
             if assignee_type == "initiator_select":
-                ids = initiator_select_user_ids or []
-                people = _preview_people(db, [str(x) for x in ids if str(x).strip()])
+                people = _preview_people(db, [str(x) for x in node_select_ids if str(x).strip()])
                 if not people:
                     meta = meta or "发起人自选"
             else:
@@ -1112,7 +1156,7 @@ def _walk_preview_nodes(
                         initiator_user_id=initiator_user_id,
                         initiator_dept_id=initiator_dept_id,
                         advanced=advanced,
-                        initiator_select_user_ids=initiator_select_user_ids,
+                        initiator_select_user_ids=node_select_ids,
                     )
                     people = _preview_people(db, assignees)
                 except AssigneeResolveError as e:
@@ -1143,7 +1187,7 @@ def preview_start_flow(
     app_id: str,
     initiator_user_id: str,
     initiator_dept_id: Optional[str] = None,
-    initiator_select_user_ids: Optional[List[str]] = None,
+    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
     form_data: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, Any]:
     """发起前预览审批链：条件字段齐后只走命中路径并解析人员。"""
@@ -1182,6 +1226,7 @@ def preview_start_flow(
         }
 
     steps: List[Dict[str, Any]] = []
+    select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
     ok = _walk_preview_nodes(
         db,
         node=root,
@@ -1189,7 +1234,7 @@ def preview_start_flow(
         initiator_user_id=initiator_user_id,
         initiator_dept_id=dept_id or None,
         advanced=_advanced(pub),
-        initiator_select_user_ids=initiator_select_user_ids,
+        initiator_select_by_node=select_by_node,
         form_data=data,
         steps=steps,
     )
@@ -1209,7 +1254,7 @@ def _create_tasks_for_node(
     node: Dict[str, Any],
     ver: OaAppVersion,
     user_id: Optional[str],
-    initiator_select_user_ids: Optional[List[str]] = None,
+    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
     order_num_start: Optional[int] = None,
 ) -> List[OaTask]:
     ntype = _node_type(node)
@@ -1223,6 +1268,16 @@ def _create_tasks_for_node(
         return created
 
     cfg = node.get("approverConfig") or {}
+    assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
+    node_select_ids = _ids_for_initiator_select_node(
+        initiator_select_by_node, node.get("id") or ""
+    )
+    if assignee_type == "initiator_select":
+        if ntype == "cc" and not node_select_ids:
+            return created
+        if ntype == "approval" and not node_select_ids:
+            raise OaEngineError("发起人未选择审批人")
+
     advanced = _advanced(ver)
     try:
         assignees, meta = resolve_assignees(
@@ -1232,7 +1287,7 @@ def _create_tasks_for_node(
             initiator_user_id=inst.initiator_user_id,
             initiator_dept_id=inst.initiator_dept_id,
             advanced=advanced,
-            initiator_select_user_ids=initiator_select_user_ids,
+            initiator_select_user_ids=node_select_ids,
         )
     except AssigneeResolveError as e:
         raise OaEngineError(str(e)) from e
@@ -1319,7 +1374,7 @@ def advance_from(
     ver: OaAppVersion,
     from_node: Optional[Dict[str, Any]],
     user_id: Optional[str],
-    initiator_select_user_ids: Optional[List[str]] = None,
+    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
 ) -> None:
     """从某节点的下一跳开始推进；from_node 为 None 表示从 start 进入。"""
     root = _flow_root(_loads(ver.flow_tree_json, {}))
@@ -1367,7 +1422,7 @@ def advance_from(
                 node=cur,
                 ver=ver,
                 user_id=user_id,
-                initiator_select_user_ids=initiator_select_user_ids,
+                initiator_select_by_node=initiator_select_by_node,
                 order_num_start=next_order,
             )
             next_order += max(len(tasks), 1)
@@ -1395,7 +1450,7 @@ def start_instance(
     initiator_dept_id: Optional[str] = None,
     form_data: Optional[Dict[str, Any]] = None,
     title: Optional[str] = None,
-    initiator_select_user_ids: Optional[List[str]] = None,
+    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
     related_instance_ids: Optional[List[str]] = None,
     source_instance_id: Optional[str] = None,
 ) -> Dict[str, Any]:
@@ -1425,6 +1480,7 @@ def start_instance(
 
     now = datetime.now()
     title_text = (title or app.app_name or "").strip()
+    normalized_select = _normalize_initiator_select_by_node(initiator_select_by_node)
     inst = OaInstance(
         id=generate_uuid(),
         app_id=app_id,
@@ -1433,6 +1489,7 @@ def start_instance(
         initiator_user_id=initiator_user_id,
         initiator_dept_id=dept_id,
         form_data_json=_dumps(data),
+        initiator_select_json=_dumps(normalized_select),
         status="running",
         source_instance_id=None,
         current_round=1,
@@ -1496,7 +1553,7 @@ def start_instance(
         ver=ver,
         from_node=None,
         user_id=initiator_user_id,
-        initiator_select_user_ids=initiator_select_user_ids,
+        initiator_select_by_node=normalized_select,
     )
     db.commit()
     db.refresh(inst)
@@ -1760,7 +1817,14 @@ def complete_task(
     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)
+    advance_from(
+        db,
+        inst=inst,
+        ver=ver,
+        from_node=node,
+        user_id=user_id,
+        initiator_select_by_node=_load_initiator_select_by_node(inst),
+    )
     db.commit()
     return get_instance_detail(db, inst.id, tenant_no=tenant_no)
 
@@ -2080,10 +2144,10 @@ def resubmit_instance(
     tenant_no: str,
     user_id: str,
     form_data: Optional[Dict[str, Any]] = None,
-    initiator_select_user_ids: Optional[List[str]] = None,
+    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
 ) -> Dict[str, Any]:
     """已废弃：同实例再次提交。请复制表单重新发起新实例。"""
-    _ = (db, instance_id, tenant_no, user_id, form_data, initiator_select_user_ids)
+    _ = (db, instance_id, tenant_no, user_id, form_data, initiator_select_by_node)
     raise OaEngineError("同实例再次提交已废弃，请复制表单重新发起")
 
 
diff --git a/backend/tests/test_oa_flow_preview.py b/backend/tests/test_oa_flow_preview.py
index 86d77df9..54e4c5d0 100644
--- a/backend/tests/test_oa_flow_preview.py
+++ b/backend/tests/test_oa_flow_preview.py
@@ -109,7 +109,7 @@ class TestWalkPreviewResolvedPath(unittest.TestCase):
                 initiator_user_id="init",
                 initiator_dept_id=None,
                 advanced={},
-                initiator_select_user_ids=None,
+                initiator_select_by_node=None,
                 form_data={"days": 5},
                 steps=steps,
             )
@@ -133,7 +133,7 @@ class TestWalkPreviewResolvedPath(unittest.TestCase):
                 initiator_user_id="init",
                 initiator_dept_id=None,
                 advanced={},
-                initiator_select_user_ids=None,
+                initiator_select_by_node=None,
                 form_data={"days": 1},
                 steps=steps,
             )
diff --git a/backend/tests/test_oa_initiator_select_by_node.py b/backend/tests/test_oa_initiator_select_by_node.py
new file mode 100644
index 00000000..68c08792
--- /dev/null
+++ b/backend/tests/test_oa_initiator_select_by_node.py
@@ -0,0 +1,108 @@
+"""OA 发起人自选：按节点 map"""
+import unittest
+from unittest.mock import MagicMock, patch
+
+from app.services.oa_engine import (
+    OaEngineError,
+    _create_tasks_for_node,
+    _ids_for_initiator_select_node,
+    _normalize_initiator_select_by_node,
+)
+
+
+class TestNormalizeMap(unittest.TestCase):
+    def test_normalize_dedupe_preserve_order(self):
+        raw = {"n1": ["u1", "u1", " u2 ", ""], "n2": "bad"}
+        out = _normalize_initiator_select_by_node(raw)
+        self.assertEqual(out["n1"], ["u1", "u2"])
+        self.assertNotIn("n2", out)
+
+    def test_ids_for_node(self):
+        m = {"a1": ["u9"]}
+        self.assertEqual(_ids_for_initiator_select_node(m, "a1"), ["u9"])
+        self.assertEqual(_ids_for_initiator_select_node(m, "missing"), [])
+
+
+class TestCreateTasksInitiatorSelect(unittest.TestCase):
+    def _inst(self):
+        inst = MagicMock()
+        inst.id = "inst1"
+        inst.tenant_no = "zkxw"
+        inst.initiator_user_id = "init"
+        inst.initiator_dept_id = "d1"
+        inst.current_round = 1
+        return inst
+
+    def test_cc_empty_skips_without_resolve(self):
+        db = MagicMock()
+        node = {
+            "id": "cc1",
+            "type": "cc",
+            "name": "抄送人",
+            "approverConfig": {"assigneeType": "initiator_select"},
+        }
+        ver = MagicMock()
+        ver.advanced_settings_json = "{}"
+        with patch("app.services.oa_engine.resolve_assignees") as resolve:
+            created = _create_tasks_for_node(
+                db,
+                inst=self._inst(),
+                node=node,
+                ver=ver,
+                user_id="init",
+                initiator_select_by_node={},
+            )
+            resolve.assert_not_called()
+        self.assertEqual(created, [])
+
+    def test_approval_empty_raises(self):
+        db = MagicMock()
+        node = {
+            "id": "a1",
+            "type": "approval",
+            "name": "审批人",
+            "approverConfig": {"assigneeType": "initiator_select", "signMode": "or"},
+        }
+        ver = MagicMock()
+        ver.advanced_settings_json = "{}"
+        with self.assertRaises(OaEngineError):
+            _create_tasks_for_node(
+                db,
+                inst=self._inst(),
+                node=node,
+                ver=ver,
+                user_id="init",
+                initiator_select_by_node={"a1": []},
+            )
+
+    def test_approval_uses_node_ids(self):
+        db = MagicMock()
+        node = {
+            "id": "a1",
+            "type": "approval",
+            "name": "审批人",
+            "approverConfig": {"assigneeType": "initiator_select", "signMode": "or"},
+        }
+        ver = MagicMock()
+        ver.advanced_settings_json = "{}"
+        with patch(
+            "app.services.oa_engine.resolve_assignees",
+            return_value=(["u1"], {"resolve_source": "initiator_select"}),
+        ) as resolve:
+            with patch("app.services.oa_engine.generate_uuid", return_value="t1"):
+                with patch("app.services.oa_engine._next_task_order_num", return_value=1):
+                    created = _create_tasks_for_node(
+                        db,
+                        inst=self._inst(),
+                        node=node,
+                        ver=ver,
+                        user_id="init",
+                        initiator_select_by_node={"a1": ["u1"], "other": ["u9"]},
+                    )
+        self.assertEqual(len(created), 1)
+        kwargs = resolve.call_args.kwargs
+        self.assertEqual(kwargs.get("initiator_select_user_ids"), ["u1"])
+
+
+if __name__ == "__main__":
+    unittest.main()