Merge pull request #299 from voidborne-d/fix/task-mutation-race-condition

fix: eliminate TOCTOU race condition in concurrent task mutations

dashboard/server.py

@@ -142,7 +142,12 @@ def load_tasks():
 def save_tasks(tasks):
     task_data_dir = get_task_data_dir()
     atomic_json_write(task_data_dir / 'tasks_source.json', tasks)
-    # Trigger refresh (异步，不阻塞，避免僵尸进程)
+    _trigger_refresh()
+
+
+def _trigger_refresh():
+    """Trigger live data refresh in background."""
+    task_data_dir = get_task_data_dir()
     script = task_data_dir.parent / 'scripts' / 'refresh_live_data.py'
     if not script.exists():
         script = SCRIPTS / 'refresh_live_data.py'
@@ -155,6 +160,43 @@ def save_tasks(tasks):
     threading.Thread(target=_refresh, daemon=True).start()
 
 
+def modify_tasks(modifier):
+    """Atomically read-modify-write the tasks file.
+
+    ``modifier(tasks)`` receives the current task list, mutates it in place
+    (or returns a new list), and the result is persisted while the file lock
+    is held.  This avoids the TOCTOU race inherent in separate
+    ``load_tasks()`` / ``save_tasks()`` calls when background threads
+    (dispatch callbacks, periodic scanner) and the HTTP handler mutate tasks
+    concurrently.
+    """
+    task_data_dir = get_task_data_dir()
+    path = task_data_dir / 'tasks_source.json'
+    atomic_json_update(path, modifier, default=[])
+    _trigger_refresh()
+
+
+def modify_task(task_id, updater):
+    """Atomically update a single task identified by *task_id*.
+
+    ``updater(task)`` receives the task dict and should mutate it in place.
+    Returns ``True`` if the task was found and updated, ``False`` otherwise.
+    """
+    found = [False]
+
+    def _modifier(tasks):
+        task = next((t for t in tasks if t.get('id') == task_id), None)
+        if task is None:
+            return tasks
+        updater(task)
+        task['updatedAt'] = now_iso()
+        found[0] = True
+        return tasks
+
+    modify_tasks(_modifier)
+    return found[0]
+
+
 def handle_task_action(task_id, action, reason):
     """Stop/cancel/resume a task from the dashboard."""
     tasks = load_tasks()
@@ -1070,15 +1112,17 @@ def _resolve_openclaw_bin():
 
 
 def _update_task_scheduler(task_id, updater):
-    tasks = load_tasks()
-    task = next((t for t in tasks if t.get('id') == task_id), None)
-    if not task:
-        return False
-    sched = _ensure_scheduler(task)
-    updater(task, sched)
-    task['updatedAt'] = now_iso()
-    save_tasks(tasks)
-    return True
+    """Atomically update a task's scheduler state.
+
+    Uses ``modify_task`` to hold the file lock for the entire
+    read-modify-write cycle, preventing concurrent dispatch threads and
+    the periodic scanner from clobbering each other's writes.
+    """
+    def _apply(task):
+        sched = _ensure_scheduler(task)
+        updater(task, sched)
+
+    return modify_task(task_id, _apply)
 
 
 def get_scheduler_state(task_id):
@@ -1104,6 +1148,7 @@ def get_scheduler_state(task_id):
 
 
 def handle_scheduler_retry(task_id, reason=''):
+    # Pre-check before acquiring lock (avoids holding lock for error paths)
     tasks = load_tasks()
     task = next((t for t in tasks if t.get('id') == task_id), None)
     if not task:
@@ -1112,16 +1157,24 @@ def handle_scheduler_retry(task_id, reason=''):
     if state in _TERMINAL_STATES or state == 'Blocked':
         return {'ok': False, 'error': f'任务 {task_id} 当前状态 {state} 不支持重试'}
 
-    sched = _ensure_scheduler(task)
-    sched['retryCount'] = int(sched.get('retryCount') or 0) + 1
-    sched['lastRetryAt'] = now_iso()
-    sched['lastDispatchTrigger'] = 'taizi-retry'
-    _scheduler_add_flow(task, f'触发重试第{sched["retryCount"]}次：{reason or "超时未推进"}')
-    task['updatedAt'] = now_iso()
-    save_tasks(tasks)
+    result = {'retryCount': 0, 'state': state}
 
-    dispatch_for_state(task_id, task, state, trigger='taizi-retry')
-    return {'ok': True, 'message': f'{task_id} 已触发重试派发', 'retryCount': sched['retryCount']}
+    def _apply(task):
+        cur = task.get('state', '')
+        if cur in _TERMINAL_STATES or cur == 'Blocked':
+            return  # state changed between pre-check and lock; skip
+        sched = _ensure_scheduler(task)
+        sched['retryCount'] = int(sched.get('retryCount') or 0) + 1
+        sched['lastRetryAt'] = now_iso()
+        sched['lastDispatchTrigger'] = 'taizi-retry'
+        _scheduler_add_flow(task, f'触发重试第{sched["retryCount"]}次：{reason or "超时未推进"}')
+        result['retryCount'] = sched['retryCount']
+        result['state'] = cur
+
+    modify_task(task_id, _apply)
+
+    dispatch_for_state(task_id, task, result['state'], trigger='taizi-retry')
+    return {'ok': True, 'message': f'{task_id} 已触发重试派发', 'retryCount': result['retryCount']}
 
 
 def handle_scheduler_escalate(task_id, reason=''):
@@ -1159,6 +1212,7 @@ def handle_scheduler_escalate(task_id, reason=''):
 
 
 def handle_scheduler_rollback(task_id, reason=''):
+    # Pre-check before acquiring lock
     tasks = load_tasks()
     task = next((t for t in tasks if t.get('id') == task_id), None)
     if not task:
@@ -1169,115 +1223,142 @@ def handle_scheduler_rollback(task_id, reason=''):
     if not snap_state:
         return {'ok': False, 'error': f'任务 {task_id} 无可用回滚快照'}
 
-    old_state = task.get('state', '')
-    task['state'] = snap_state
-    task['org'] = snapshot.get('org', task.get('org', ''))
-    task['now'] = f'↩️ 太子调度自动回滚：{reason or "恢复到上个稳定节点"}'
-    task['block'] = '无'
-    sched['retryCount'] = 0
-    sched['escalationLevel'] = 0
-    sched['stallSince'] = None
-    sched['lastProgressAt'] = now_iso()
-    _scheduler_add_flow(task, f'执行回滚：{old_state} → {snap_state}，原因：{reason or "停滞恢复"}')
-    task['updatedAt'] = now_iso()
-    save_tasks(tasks)
+    result = {'snap_state': snap_state}
 
-    if snap_state not in _TERMINAL_STATES:
-        dispatch_for_state(task_id, task, snap_state, trigger='taizi-rollback')
+    def _apply(task):
+        sched = _ensure_scheduler(task)
+        snapshot = sched.get('snapshot') or {}
+        s_state = snapshot.get('state')
+        if not s_state:
+            return  # snapshot cleared between pre-check and lock
+        old_state = task.get('state', '')
+        task['state'] = s_state
+        task['org'] = snapshot.get('org', task.get('org', ''))
+        task['now'] = f'↩️ 太子调度自动回滚：{reason or "恢复到上个稳定节点"}'
+        task['block'] = '无'
+        sched['retryCount'] = 0
+        sched['escalationLevel'] = 0
+        sched['stallSince'] = None
+        sched['lastProgressAt'] = now_iso()
+        _scheduler_add_flow(task, f'执行回滚：{old_state} → {s_state}，原因：{reason or "停滞恢复"}')
+        result['snap_state'] = s_state
 
-    return {'ok': True, 'message': f'{task_id} 已回滚到 {snap_state}'}
+    modify_task(task_id, _apply)
+
+    if result['snap_state'] not in _TERMINAL_STATES:
+        dispatch_for_state(task_id, task, result['snap_state'], trigger='taizi-rollback')
+
+    return {'ok': True, 'message': f'{task_id} 已回滚到 {result["snap_state"]}'}
 
 
 def handle_scheduler_scan(threshold_sec=600):
+    """Periodic stall scanner — runs in a background thread.
+
+    Uses ``modify_tasks`` to hold the file lock during the mutation phase,
+    preventing concurrent dispatch callbacks and HTTP handlers from
+    clobbering each other's writes (fixes TOCTOU race between the old
+    ``load_tasks()`` / ``save_tasks()`` pair).
+
+    Side-effects (dispatch, escalation wake) are executed *after* the lock
+    is released so they don't block other writers.
+    """
     threshold_sec = max(60, int(threshold_sec or 600))
-    tasks = load_tasks()
     now_dt = datetime.datetime.now(datetime.timezone.utc)
+    # Collect dispatch/escalation work to execute after the lock is released
     pending_retries = []
     pending_escalates = []
     pending_rollbacks = []
     actions = []
-    changed = False
 
-    for task in tasks:
-        task_id = task.get('id', '')
-        state = task.get('state', '')
-        if not task_id or state in _TERMINAL_STATES or task.get('archived'):
-            continue
-        if state == 'Blocked':
-            continue
+    def _scan(tasks):
+        changed = False
+        for task in tasks:
+            task_id = task.get('id', '')
+            state = task.get('state', '')
+            if not task_id or state in _TERMINAL_STATES or task.get('archived'):
+                continue
+            if state == 'Blocked':
+                continue
 
-        sched = _ensure_scheduler(task)
-        task_threshold = int(sched.get('stallThresholdSec') or threshold_sec)
-        last_progress = _parse_iso(sched.get('lastProgressAt') or task.get('updatedAt'))
-        if not last_progress:
-            continue
-        stalled_sec = max(0, int((now_dt - last_progress).total_seconds()))
-        if stalled_sec < task_threshold:
-            continue
+            sched = _ensure_scheduler(task)
+            task_threshold = int(sched.get('stallThresholdSec') or threshold_sec)
+            last_progress = _parse_iso(sched.get('lastProgressAt') or task.get('updatedAt'))
+            if not last_progress:
+                continue
+            stalled_sec = max(0, int((now_dt - last_progress).total_seconds()))
+            if stalled_sec < task_threshold:
+                continue
 
-        if not sched.get('stallSince'):
-            sched['stallSince'] = now_iso()
-            changed = True
-
-        retry_count = int(sched.get('retryCount') or 0)
-        max_retry = max(0, int(sched.get('maxRetry') or 1))
-        level = int(sched.get('escalationLevel') or 0)
-
-        if retry_count < max_retry:
-            sched['retryCount'] = retry_count + 1
-            sched['lastRetryAt'] = now_iso()
-            sched['lastDispatchTrigger'] = 'taizi-scan-retry'
-            _scheduler_add_flow(task, f'停滞{stalled_sec}秒，触发自动重试第{sched["retryCount"]}次')
-            pending_retries.append((task_id, state))
-            actions.append({'taskId': task_id, 'action': 'retry', 'stalledSec': stalled_sec})
-            changed = True
-            continue
-
-        if level < 2:
-            next_level = level + 1
-            target = 'menxia' if next_level == 1 else 'shangshu'
-            target_label = '门下省' if next_level == 1 else '尚书省'
-            sched['escalationLevel'] = next_level
-            sched['lastEscalatedAt'] = now_iso()
-            _scheduler_add_flow(task, f'停滞{stalled_sec}秒，升级至{target_label}协调', to=target_label)
-            pending_escalates.append((task_id, state, target, target_label, stalled_sec))
-            actions.append({'taskId': task_id, 'action': 'escalate', 'to': target_label, 'stalledSec': stalled_sec})
-            changed = True
-            continue
-
-        if sched.get('autoRollback', True):
-            rollback_count = int(sched.get('rollbackCount') or 0)
-            max_rollback = int(sched.get('maxRollback') or 3)
-            snapshot = sched.get('snapshot') or {}
-            snap_state = snapshot.get('state')
-            if rollback_count >= max_rollback:
-                # 已达最大回滚次数，标记 Blocked 避免无限循环
-                if state != 'Blocked':
-                    task['state'] = 'Blocked'
-                    task['now'] = f'🚫 连续回滚{rollback_count}次仍无法推进，已自动挂起'
-                    task['block'] = f'连续停滞且回滚{rollback_count}次均失败，需人工介入'
-                    sched['stallSince'] = None
-                    _scheduler_add_flow(task, f'连续回滚{rollback_count}次，自动挂起等待人工介入')
-                    actions.append({'taskId': task_id, 'action': 'blocked', 'reason': f'max rollback {rollback_count}'})
-                    changed = True
-            elif snap_state and snap_state != state:
-                old_state = state
-                task['state'] = snap_state
-                task['org'] = snapshot.get('org', task.get('org', ''))
-                task['now'] = '↩️ 太子调度自动回滚到稳定节点'
-                task['block'] = '无'
-                sched['retryCount'] = 0
-                sched['escalationLevel'] = 0
-                sched['rollbackCount'] = rollback_count + 1
-                sched['stallSince'] = None
-                sched['lastProgressAt'] = now_iso()
-                _scheduler_add_flow(task, f'连续停滞，自动回滚：{old_state} → {snap_state}（第{rollback_count + 1}次）')
-                pending_rollbacks.append((task_id, snap_state))
-                actions.append({'taskId': task_id, 'action': 'rollback', 'toState': snap_state})
+            if not sched.get('stallSince'):
+                sched['stallSince'] = now_iso()
                 changed = True
 
-    if changed:
-        save_tasks(tasks)
+            retry_count = int(sched.get('retryCount') or 0)
+            max_retry = max(0, int(sched.get('maxRetry') or 1))
+            level = int(sched.get('escalationLevel') or 0)
+
+            if retry_count < max_retry:
+                sched['retryCount'] = retry_count + 1
+                sched['lastRetryAt'] = now_iso()
+                sched['lastDispatchTrigger'] = 'taizi-scan-retry'
+                _scheduler_add_flow(task, f'停滞{stalled_sec}秒，触发自动重试第{sched["retryCount"]}次')
+                pending_retries.append((task_id, state))
+                actions.append({'taskId': task_id, 'action': 'retry', 'stalledSec': stalled_sec})
+                changed = True
+                continue
+
+            if level < 2:
+                next_level = level + 1
+                target = 'menxia' if next_level == 1 else 'shangshu'
+                target_label = '门下省' if next_level == 1 else '尚书省'
+                sched['escalationLevel'] = next_level
+                sched['lastEscalatedAt'] = now_iso()
+                _scheduler_add_flow(task, f'停滞{stalled_sec}秒，升级至{target_label}协调', to=target_label)
+                pending_escalates.append((task_id, state, target, target_label, stalled_sec))
+                actions.append({'taskId': task_id, 'action': 'escalate', 'to': target_label, 'stalledSec': stalled_sec})
+                changed = True
+                continue
+
+            if sched.get('autoRollback', True):
+                rollback_count = int(sched.get('rollbackCount') or 0)
+                max_rollback = int(sched.get('maxRollback') or 3)
+                snapshot = sched.get('snapshot') or {}
+                snap_state = snapshot.get('state')
+                if rollback_count >= max_rollback:
+                    if state != 'Blocked':
+                        task['state'] = 'Blocked'
+                        task['now'] = f'🚫 连续回滚{rollback_count}次仍无法推进，已自动挂起'
+                        task['block'] = f'连续停滞且回滚{rollback_count}次均失败，需人工介入'
+                        sched['stallSince'] = None
+                        _scheduler_add_flow(task, f'连续回滚{rollback_count}次，自动挂起等待人工介入')
+                        actions.append({'taskId': task_id, 'action': 'blocked', 'reason': f'max rollback {rollback_count}'})
+                        changed = True
+                elif snap_state and snap_state != state:
+                    old_state = state
+                    task['state'] = snap_state
+                    task['org'] = snapshot.get('org', task.get('org', ''))
+                    task['now'] = '↩️ 太子调度自动回滚到稳定节点'
+                    task['block'] = '无'
+                    sched['retryCount'] = 0
+                    sched['escalationLevel'] = 0
+                    sched['rollbackCount'] = rollback_count + 1
+                    sched['stallSince'] = None
+                    sched['lastProgressAt'] = now_iso()
+                    _scheduler_add_flow(task, f'连续停滞，自动回滚：{old_state} → {snap_state}（第{rollback_count + 1}次）')
+                    pending_rollbacks.append((task_id, snap_state))
+                    actions.append({'taskId': task_id, 'action': 'rollback', 'toState': snap_state})
+                    changed = True
+
+        return tasks  # always return — atomic_json_update requires it
+
+    modify_tasks(_scan)
+
+    # --- Side-effects: dispatch & escalation (outside the file lock) ---
+
+    # Re-read tasks for dispatch context (the task objects from _scan are
+    # no longer held under the lock, but dispatch only needs id + state +
+    # title which are immutable at this point).
+    tasks = load_tasks()
 
     for task_id, state in pending_retries:
         retry_task = next((t for t in tasks if t.get('id') == task_id), None)

tests/test_task_mutation_race.py

@@ -0,0 +1,327 @@
+"""Tests for task mutation atomicity — verifying that concurrent writers
+(dispatch threads, periodic scanner, HTTP handlers) cannot clobber each
+other's changes.
+
+The core issue: the old ``load_tasks()`` + modify + ``save_tasks()`` pattern
+allows two concurrent threads to both read the same snapshot, each modify
+a different field, and the second ``save_tasks()`` overwrites the first's
+changes — a classic TOCTOU (Time-of-Check-Time-of-Use) race.
+
+The fix introduces ``modify_tasks()`` / ``modify_task()`` wrappers around
+``atomic_json_update()`` which hold the file lock for the entire
+read-modify-write cycle.
+"""
+import json
+import pathlib
+import sys
+import threading
+import time
+
+ROOT = pathlib.Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / 'dashboard'))
+sys.path.insert(0, str(ROOT / 'scripts'))
+
+
+def _setup_server(monkeypatch, tmp_path, tasks=None):
+    """Bootstrap server module with isolated data directory."""
+    import server as srv
+
+    data_dir = tmp_path / 'data'
+    data_dir.mkdir()
+    tasks_path = data_dir / 'tasks_source.json'
+    initial = tasks or []
+    tasks_path.write_text(json.dumps(initial, ensure_ascii=False), encoding='utf-8')
+    (data_dir / 'agent_config.json').write_text('{}', encoding='utf-8')
+
+    monkeypatch.setattr(srv, 'DATA', data_dir)
+    monkeypatch.setattr(srv, '_ACTIVE_TASK_DATA_DIR', data_dir)
+    monkeypatch.setattr(srv, 'SCRIPTS', tmp_path / 'scripts')  # avoid real scripts
+    monkeypatch.setattr(srv, '_check_gateway_alive', lambda: False)  # no real dispatch
+    # Suppress refresh subprocess
+    monkeypatch.setattr(srv, '_trigger_refresh', lambda: None)
+
+    return srv, data_dir, tasks_path
+
+
+# ── Test: modify_tasks holds file lock ──
+
+
+class TestModifyTasksAtomicity:
+    """Verify that ``modify_tasks`` uses atomic_json_update under the hood."""
+
+    def test_modify_tasks_exists_and_callable(self, monkeypatch, tmp_path):
+        srv, _, _ = _setup_server(monkeypatch, tmp_path)
+        assert callable(getattr(srv, 'modify_tasks', None)), \
+            'modify_tasks must be a callable function on server module'
+
+    def test_modify_task_exists_and_callable(self, monkeypatch, tmp_path):
+        srv, _, _ = _setup_server(monkeypatch, tmp_path)
+        assert callable(getattr(srv, 'modify_task', None)), \
+            'modify_task must be a callable function on server module'
+
+    def test_modify_task_updates_single_task(self, monkeypatch, tmp_path):
+        task = {
+            'id': 'T-001', 'title': '测试', 'state': 'Doing',
+            'org': '兵部', 'updatedAt': '2026-04-22T00:00:00Z',
+        }
+        srv, _, tasks_path = _setup_server(monkeypatch, tmp_path, [task])
+
+        found = srv.modify_task('T-001', lambda t: t.update({'state': 'Review'}))
+        assert found is True
+
+        data = json.loads(tasks_path.read_text(encoding='utf-8'))
+        assert data[0]['state'] == 'Review'
+        assert 'updatedAt' in data[0]  # auto-stamped
+
+    def test_modify_task_returns_false_for_missing(self, monkeypatch, tmp_path):
+        srv, _, _ = _setup_server(monkeypatch, tmp_path, [])
+        found = srv.modify_task('NONEXISTENT', lambda t: t.update({'state': 'Done'}))
+        assert found is False
+
+    def test_modify_tasks_bulk_update(self, monkeypatch, tmp_path):
+        tasks = [
+            {'id': 'T-A', 'title': 'A', 'state': 'Doing', 'org': '', 'updatedAt': ''},
+            {'id': 'T-B', 'title': 'B', 'state': 'Doing', 'org': '', 'updatedAt': ''},
+        ]
+        srv, _, tasks_path = _setup_server(monkeypatch, tmp_path, tasks)
+
+        def _mark_all_done(tasks):
+            for t in tasks:
+                t['state'] = 'Done'
+            return tasks
+
+        srv.modify_tasks(_mark_all_done)
+
+        data = json.loads(tasks_path.read_text(encoding='utf-8'))
+        assert all(t['state'] == 'Done' for t in data)
+
+
+# ── Test: _update_task_scheduler uses modify_task ──
+
+
+class TestUpdateTaskSchedulerAtomicity:
+    """Verify that ``_update_task_scheduler`` no longer uses the racy
+    ``load_tasks()`` + ``save_tasks()`` pattern."""
+
+    def test_scheduler_update_persists_atomically(self, monkeypatch, tmp_path):
+        task = {
+            'id': 'T-002', 'title': '派发测试', 'state': 'Taizi',
+            'org': '太子', 'updatedAt': '2026-04-22T01:00:00Z',
+        }
+        srv, _, tasks_path = _setup_server(monkeypatch, tmp_path, [task])
+
+        srv._update_task_scheduler('T-002', lambda t, s: s.update({
+            'lastDispatchStatus': 'success',
+            'lastDispatchAgent': 'taizi',
+        }))
+
+        data = json.loads(tasks_path.read_text(encoding='utf-8'))
+        sched = data[0].get('_scheduler', {})
+        assert sched['lastDispatchStatus'] == 'success'
+        assert sched['lastDispatchAgent'] == 'taizi'
+
+    def test_scheduler_update_missing_task(self, monkeypatch, tmp_path):
+        srv, _, _ = _setup_server(monkeypatch, tmp_path, [])
+        result = srv._update_task_scheduler('MISSING', lambda t, s: None)
+        assert result is False
+
+
+# ── Test: handle_scheduler_scan uses modify_tasks ──
+
+
+class TestSchedulerScanAtomicity:
+    """Verify that the periodic scanner mutates tasks under the file lock."""
+
+    def test_scan_stalled_task_triggers_retry(self, monkeypatch, tmp_path):
+        """A task stalled past threshold should get retryCount incremented
+        and the change should be persisted atomically."""
+        import datetime
+
+        old_ts = (
+            datetime.datetime.now(datetime.timezone.utc)
+            - datetime.timedelta(seconds=700)
+        ).isoformat()
+
+        task = {
+            'id': 'T-003', 'title': '停滞任务', 'state': 'Zhongshu',
+            'org': '中书省', 'updatedAt': old_ts,
+            '_scheduler': {
+                'enabled': True, 'stallThresholdSec': 600, 'maxRetry': 2,
+                'retryCount': 0, 'escalationLevel': 0, 'autoRollback': True,
+                'lastProgressAt': old_ts, 'stallSince': None,
+                'lastDispatchStatus': 'idle', 'rollbackCount': 0,
+                'snapshot': {'state': 'Taizi', 'org': '太子', 'now': '', 'savedAt': old_ts, 'note': 'init'},
+            },
+        }
+        srv, _, tasks_path = _setup_server(monkeypatch, tmp_path, [task])
+
+        # Suppress dispatch side-effects
+        monkeypatch.setattr(srv, 'dispatch_for_state', lambda *a, **kw: None)
+        monkeypatch.setattr(srv, 'wake_agent', lambda *a, **kw: None)
+
+        result = srv.handle_scheduler_scan(threshold_sec=600)
+        assert result['ok'] is True
+        assert result['count'] >= 1
+
+        data = json.loads(tasks_path.read_text(encoding='utf-8'))
+        sched = data[0].get('_scheduler', {})
+        assert sched['retryCount'] == 1
+        assert sched['lastDispatchTrigger'] == 'taizi-scan-retry'
+
+
+# ── Test: concurrent modify_task calls don't clobber ──
+
+
+class TestConcurrentModifyTask:
+    """Simulate the race that existed before the fix: two threads
+    concurrently modifying different fields of the same task."""
+
+    def test_concurrent_writes_both_persist(self, monkeypatch, tmp_path):
+        """Two threads updating different scheduler fields should both
+        be visible in the final state (no lost updates)."""
+        task = {
+            'id': 'T-RACE', 'title': '竞争测试', 'state': 'Doing',
+            'org': '兵部', 'updatedAt': '2026-04-22T02:00:00Z',
+            '_scheduler': {
+                'enabled': True, 'stallThresholdSec': 600, 'maxRetry': 2,
+                'retryCount': 0, 'escalationLevel': 0, 'autoRollback': True,
+                'lastProgressAt': '2026-04-22T02:00:00Z', 'stallSince': None,
+                'lastDispatchStatus': 'idle', 'rollbackCount': 0,
+                'field_a': 'initial_a', 'field_b': 'initial_b',
+                'snapshot': {'state': 'Assigned', 'org': '尚书省', 'now': '', 'savedAt': '', 'note': 'init'},
+            },
+        }
+        srv, _, tasks_path = _setup_server(monkeypatch, tmp_path, [task])
+        monkeypatch.setattr(srv, '_trigger_refresh', lambda: None)
+
+        barrier = threading.Barrier(2, timeout=5)
+        errors = []
+
+        def update_field_a():
+            try:
+                barrier.wait()
+                srv.modify_task('T-RACE', lambda t: t.setdefault('_scheduler', {}).update({'field_a': 'updated_a'}))
+            except Exception as e:
+                errors.append(e)
+
+        def update_field_b():
+            try:
+                barrier.wait()
+                srv.modify_task('T-RACE', lambda t: t.setdefault('_scheduler', {}).update({'field_b': 'updated_b'}))
+            except Exception as e:
+                errors.append(e)
+
+        t1 = threading.Thread(target=update_field_a)
+        t2 = threading.Thread(target=update_field_b)
+        t1.start()
+        t2.start()
+        t1.join(timeout=10)
+        t2.join(timeout=10)
+
+        assert not errors, f'Thread errors: {errors}'
+
+        data = json.loads(tasks_path.read_text(encoding='utf-8'))
+        sched = data[0].get('_scheduler', {})
+
+        # With atomic modify_task, BOTH updates must be visible.
+        # The old load_tasks/save_tasks pattern would lose one.
+        assert sched['field_a'] == 'updated_a', \
+            f'field_a lost: {sched.get("field_a")}'
+        assert sched['field_b'] == 'updated_b', \
+            f'field_b lost: {sched.get("field_b")}'
+
+
+# ── Test: source audit — no racy load/save in scheduler paths ──
+
+
+class TestSourceAudit:
+    """Verify that the critical concurrent paths no longer use the racy
+    ``load_tasks()`` + ``save_tasks()`` pattern."""
+
+    def test_update_task_scheduler_no_load_save(self):
+        """_update_task_scheduler should not call load_tasks or save_tasks directly."""
+        import inspect
+        import server as srv
+
+        source = inspect.getsource(srv._update_task_scheduler)
+        assert 'load_tasks' not in source, \
+            '_update_task_scheduler still calls load_tasks() — should use modify_task()'
+        assert 'save_tasks' not in source, \
+            '_update_task_scheduler still calls save_tasks() — should use modify_task()'
+
+    def test_handle_scheduler_scan_no_save_tasks(self):
+        """handle_scheduler_scan should use modify_tasks, not save_tasks()."""
+        import ast
+        import inspect
+        import server as srv
+
+        source = inspect.getsource(srv.handle_scheduler_scan)
+        tree = ast.parse(source)
+        # Check that save_tasks() is not called in the function body
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                func = node.func
+                name = ''
+                if isinstance(func, ast.Name):
+                    name = func.id
+                elif isinstance(func, ast.Attribute):
+                    name = func.attr
+                assert name != 'save_tasks', \
+                    'handle_scheduler_scan still calls save_tasks() — should use modify_tasks()'
+
+    def test_modify_tasks_uses_atomic_json_update(self):
+        """modify_tasks must delegate to atomic_json_update for lock safety."""
+        import inspect
+        import server as srv
+
+        source = inspect.getsource(srv.modify_tasks)
+        assert 'atomic_json_update' in source, \
+            'modify_tasks must use atomic_json_update for file-level locking'
+
+    def test_modify_task_delegates_to_modify_tasks(self):
+        """modify_task should use modify_tasks (or atomic_json_update) internally."""
+        import inspect
+        import server as srv
+
+        source = inspect.getsource(srv.modify_task)
+        assert 'modify_tasks' in source or 'atomic_json_update' in source, \
+            'modify_task should delegate to modify_tasks or atomic_json_update'
+
+    def test_handle_scheduler_retry_uses_modify_task(self):
+        """handle_scheduler_retry should use modify_task instead of load/save."""
+        import inspect
+        import server as srv
+
+        source = inspect.getsource(srv.handle_scheduler_retry)
+        assert 'modify_task' in source, \
+            'handle_scheduler_retry should use modify_task for atomic updates'
+
+    def test_handle_scheduler_rollback_uses_modify_task(self):
+        """handle_scheduler_rollback should use modify_task instead of load/save."""
+        import inspect
+        import server as srv
+
+        source = inspect.getsource(srv.handle_scheduler_rollback)
+        assert 'modify_task' in source, \
+            'handle_scheduler_rollback should use modify_task for atomic updates'
+
+
+# ── Test: backward compatibility — load_tasks/save_tasks still exist ──
+
+
+class TestBackwardCompatibility:
+    """HTTP handler paths still use load_tasks/save_tasks for now;
+    ensure they remain functional."""
+
+    def test_load_tasks_still_works(self, monkeypatch, tmp_path):
+        tasks = [{'id': 'T-BC', 'title': '兼容性', 'state': 'Doing'}]
+        srv, _, _ = _setup_server(monkeypatch, tmp_path, tasks)
+        loaded = srv.load_tasks()
+        assert len(loaded) == 1
+        assert loaded[0]['id'] == 'T-BC'
+
+    def test_save_tasks_still_works(self, monkeypatch, tmp_path):
+        srv, _, tasks_path = _setup_server(monkeypatch, tmp_path, [])
+        srv.save_tasks([{'id': 'T-NEW', 'title': '新', 'state': 'Pending'}])
+        data = json.loads(tasks_path.read_text(encoding='utf-8'))
+        assert data[0]['id'] == 'T-NEW'