diff --git a/dashboard/server.py b/dashboard/server.py index 3e7651e..c70ecd0 100644 --- a/dashboard/server.py +++ b/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) diff --git a/tests/test_task_mutation_race.py b/tests/test_task_mutation_race.py new file mode 100644 index 0000000..2aa376b --- /dev/null +++ b/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'