fix(duty): allow re-saving historical assigns without duty staff role

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/duty_schedule.py

@@ -290,6 +290,8 @@ def save_day(
     for slot_in in body.slots:
         order += 1
         slot = existing_by_type.get(slot_in.shiftType)
+        # 已有 slot 上历史 assign 用户可再保存（D4）；新建 slot 必须具备值班人员角色（D7）
+        grandfathered_user_ids: set = set()
         if slot is None:
             slot = DutySlot(
                 id=generate_id(),
@@ -317,6 +319,11 @@ def save_day(
                 .filter(DutyAssignment.duty_slot_id == slot.id, DutyAssignment.del_flag == 0)
                 .all()
             )
+            grandfathered_user_ids = {
+                a.user_id
+                for a in existing_assigns
+                if a.source == SOURCE_ASSIGN and a.user_id
+            }
             # body.assignments 视为该班次最终「指定」名单；grab 全部保留，除非超容时拒绝缩容
             grab_rows = [a for a in existing_assigns if a.source == SOURCE_GRAB]
             new_assign_count = len(slot_in.assignments or [])
@@ -352,7 +359,8 @@ def save_day(
             if not person_name:
                 raise DutyBizError("指定员工姓名不能为空")
             if item.source == SOURCE_ASSIGN and uid:
-                assert_user_is_duty_staff(db, tenant_no, uid)
+                if uid not in grandfathered_user_ids:
+                    assert_user_is_duty_staff(db, tenant_no, uid)
             db.add(
                 DutyAssignment(
                     id=generate_id(),
@@ -444,13 +452,28 @@ def list_duty_staff_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:
 
 
 def assert_user_is_duty_staff(db: Session, tenant_no: str, user_id: str) -> None:
+    uid = (user_id or "").strip()
+    if not uid:
+        raise DutyBizError("用户无效或不具备值班人员角色")
+    user = (
+        db.query(SysUser)
+        .filter(
+            SysUser.id == uid,
+            SysUser.tenant_no == tenant_no,
+            SysUser.del_flag == 0,
+            SysUser.status == "1",
+        )
+        .first()
+    )
+    if not user:
+        raise DutyBizError("用户无效或不具备值班人员角色")
     role_id = resolve_duty_staff_role_id(db, tenant_no)
     if not role_id:
         raise DutyBizError("未配置「值班人员」角色，请联系管理员")
     link = (
         db.query(SysUserRole)
         .filter(
-            SysUserRole.user_id == user_id,
+            SysUserRole.user_id == uid,
             SysUserRole.role_id == role_id,
             SysUserRole.tenant_no == tenant_no,
             SysUserRole.del_flag == 0,
@@ -458,7 +481,7 @@ def assert_user_is_duty_staff(db: Session, tenant_no: str, user_id: str) -> None
         .first()
     )
     if not link:
-        raise DutyBizError("仅可选择「值班人员」角色的用户")
+        raise DutyBizError("用户无效或不具备值班人员角色")
 
 
 def _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]:

backend/tests/test_duty_schedule.py

@@ -13,6 +13,7 @@ from app.models.duty_models import (
     SHIFT_EVENING,
     SHIFT_LABELS,
     SHIFT_MORNING,
+    SOURCE_ASSIGN,
     SOURCE_GRAB,
 )
 from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn
@@ -375,3 +376,177 @@ def test_grab_rejects_non_duty_staff():
     db.query.side_effect = query_side_effect
     with pytest.raises(DutyBizError, match="值班人员"):
         grab(db, "tok", SHIFT_EVENING, "u1")
+
+
+def test_save_day_grandfathers_existing_assign_without_role():
+    """历史指定用户无值班人员角色时，原样再保存不得因角色校验失败。"""
+    db = MagicMock()
+    day = SimpleNamespace(
+        id="day1",
+        duty_date=date(2026, 8, 20),
+        access_token="tok",
+        link_status=LINK_OPEN,
+        tenant_no="t1",
+        del_flag=0,
+        update_by=None,
+        update_time=None,
+    )
+    slot = SimpleNamespace(
+        id="slot1",
+        duty_day_id="day1",
+        shift_type=SHIFT_EVENING,
+        capacity=2,
+        del_flag=0,
+        order_num=1,
+        update_by=None,
+        update_time=None,
+    )
+    old_assign = SimpleNamespace(
+        id="a_old",
+        duty_slot_id="slot1",
+        source=SOURCE_ASSIGN,
+        user_id="u_old",
+        person_name="老员工",
+        del_flag=0,
+        update_by=None,
+        update_time=None,
+    )
+
+    day_q = MagicMock()
+    day_q.filter.return_value.first.return_value = day
+    slot_list_q = MagicMock()
+    slot_list_q.filter.return_value.all.return_value = [slot]
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
+    role_q = MagicMock()
+    role_q.filter.return_value.first.return_value = role
+    ur_q = MagicMock()
+    ur_q.filter.return_value.first.return_value = None  # 无角色关联
+    user_q = MagicMock()
+    user_q.filter.return_value.first.return_value = SimpleNamespace(
+        id="u_old", tenant_no="t1", del_flag=0, status="1"
+    )
+
+    def query_side_effect(model):
+        name = getattr(model, "__name__", "")
+        if name == "DutyDay":
+            return day_q
+        if name == "DutySlot":
+            return slot_list_q
+        if name == "DutyAssignment":
+            m = MagicMock()
+            m.filter.return_value.count.return_value = 1
+            m.filter.return_value.all.return_value = [old_assign]
+            return m
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
+        if name == "SysUser":
+            return user_q
+        return MagicMock()
+
+    db.query.side_effect = query_side_effect
+
+    body = DutyDaySaveBody(
+        dutyDate=date(2026, 8, 20),
+        slots=[
+            DutySlotSaveIn(
+                shiftType=SHIFT_EVENING,
+                capacity=2,
+                assignments=[
+                    DutyAssignmentIn(source="assign", userId="u_old", personName="老员工")
+                ],
+            )
+        ],
+    )
+
+    # 不得因「无值班人员角色」抛错；序列化侧 mock 足够跑完即可
+    result = save_day(db, "t1", {"user_id": "op1"}, body)
+    assert result is not None
+    db.commit.assert_called()
+
+
+def test_save_day_rejects_new_assign_without_role():
+    """新指定用户无值班人员角色时必须拒绝。"""
+    db = MagicMock()
+    day = SimpleNamespace(
+        id="day1",
+        duty_date=date(2026, 8, 20),
+        access_token="tok",
+        link_status=LINK_OPEN,
+        tenant_no="t1",
+        del_flag=0,
+        update_by=None,
+        update_time=None,
+    )
+    slot = SimpleNamespace(
+        id="slot1",
+        duty_day_id="day1",
+        shift_type=SHIFT_EVENING,
+        capacity=2,
+        del_flag=0,
+        order_num=1,
+        update_by=None,
+        update_time=None,
+    )
+    old_assign = SimpleNamespace(
+        id="a_old",
+        duty_slot_id="slot1",
+        source=SOURCE_ASSIGN,
+        user_id="u_old",
+        person_name="老员工",
+        del_flag=0,
+        update_by=None,
+        update_time=None,
+    )
+
+    day_q = MagicMock()
+    day_q.filter.return_value.first.return_value = day
+    slot_list_q = MagicMock()
+    slot_list_q.filter.return_value.all.return_value = [slot]
+    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
+    role_q = MagicMock()
+    role_q.filter.return_value.first.return_value = role
+    ur_q = MagicMock()
+    ur_q.filter.return_value.first.return_value = None
+    user_q = MagicMock()
+    user_q.filter.return_value.first.return_value = SimpleNamespace(
+        id="u_new", tenant_no="t1", del_flag=0, status="1"
+    )
+
+    def query_side_effect(model):
+        name = getattr(model, "__name__", "")
+        if name == "DutyDay":
+            return day_q
+        if name == "DutySlot":
+            return slot_list_q
+        if name == "DutyAssignment":
+            m = MagicMock()
+            m.filter.return_value.count.return_value = 1
+            m.filter.return_value.all.return_value = [old_assign]
+            return m
+        if name == "SysRole":
+            return role_q
+        if name == "SysUserRole":
+            return ur_q
+        if name == "SysUser":
+            return user_q
+        return MagicMock()
+
+    db.query.side_effect = query_side_effect
+
+    body = DutyDaySaveBody(
+        dutyDate=date(2026, 8, 20),
+        slots=[
+            DutySlotSaveIn(
+                shiftType=SHIFT_EVENING,
+                capacity=2,
+                assignments=[
+                    DutyAssignmentIn(source="assign", userId="u_new", personName="新员工")
+                ],
+            )
+        ],
+    )
+
+    with pytest.raises(DutyBizError, match="值班人员"):
+        save_day(db, "t1", {"user_id": "op1"}, body)