Class TestConcurrentModifyTask

tests/test_task_mutation_race.py:175–231

Simulate the race that existed before the fix: two threads concurrently modifying different fields of the same task.

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")}'