From 64efa0606c96633cd2a32b1db8803d18ab6b9de3 Mon Sep 17 00:00:00 2001 From: weihaoxuan <641627652@qq.com> Date: Tue, 8 Sep 2026 16:37:15 +0800 Subject: [PATCH] fix(file-patch): serialize concurrent updates --- .../tests/test_file_patch_concurrency.py | 95 +++++++++++++++++++ ga.py | 48 ++++++++-- 2 files changed, 135 insertions(+), 8 deletions(-) create mode 100644 frontends/tests/test_file_patch_concurrency.py diff --git a/frontends/tests/test_file_patch_concurrency.py b/frontends/tests/test_file_patch_concurrency.py new file mode 100644 index 000000000..87909a9b4 --- /dev/null +++ b/frontends/tests/test_file_patch_concurrency.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import multiprocessing +import builtins +import importlib.util +import os +import sys +import threading +from pathlib import Path + + +ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(ROOT)) + + +def _load_ga(): + spec = importlib.util.spec_from_file_location("ga_file_patch_under_test", ROOT / "ga.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + previous_agent_loop = sys.modules.pop("agent_loop", None) + try: + spec.loader.exec_module(module) + finally: + if previous_agent_loop is not None: sys.modules["agent_loop"] = previous_agent_loop + return module + + +def _patch(path: str, old: str, new: str, ready): + ga = _load_ga() + ready.wait() + print(ga.file_patch(path, old, new), flush=True) + + +def test_concurrent_file_patches_are_serialized(tmp_path): + path = tmp_path / "state.txt" + path.write_text("A=0\nB=0\n", encoding="utf-8") + ready = multiprocessing.get_context("spawn").Event() + workers = [ + multiprocessing.get_context("spawn").Process(target=_patch, args=(str(path), "A=0", "A=1", ready)), + multiprocessing.get_context("spawn").Process(target=_patch, args=(str(path), "B=0", "B=1", ready)), + ] + for worker in workers: worker.start() + ready.set() + for worker in workers: + worker.join(timeout=10) + assert worker.exitcode == 0 + assert path.read_text(encoding="utf-8") == "A=1\nB=1\n" + + +def test_same_process_patches_do_not_read_a_stale_snapshot(tmp_path, monkeypatch): + ga = _load_ga() + + path = tmp_path / "state.txt" + path.write_text("A=0\nB=0\n", encoding="utf-8") + both_read = threading.Event() + read_count = 0 + read_count_lock = threading.Lock() + real_open = builtins.open + + class SynchronizedReader: + def __init__(self, file): self.file = file + def __enter__(self): self.file.__enter__(); return self + def __exit__(self, *args): return self.file.__exit__(*args) + def read(self): + nonlocal read_count + content = self.file.read() + with read_count_lock: + read_count += 1 + if read_count == 2: both_read.set() + both_read.wait(timeout=0.2) + return content + + def synchronized_open(file, mode="r", *args, **kwargs): + handle = real_open(file, mode, *args, **kwargs) + return SynchronizedReader(handle) if os.fspath(file) == str(path) and mode == "r" else handle + + monkeypatch.setattr(builtins, "open", synchronized_open) + start = threading.Barrier(3) + results = [] + + def patch(old, new): + start.wait() + results.append(ga.file_patch(str(path), old, new)) + + threads = [ + threading.Thread(target=patch, args=("A=0", "A=1")), + threading.Thread(target=patch, args=("B=0", "B=1")), + ] + for thread in threads: thread.start() + start.wait() + for thread in threads: + thread.join(timeout=5) + assert not thread.is_alive() + assert all(result["status"] == "success" for result in results) + assert path.read_text(encoding="utf-8") == "A=1\nB=1\n" diff --git a/ga.py b/ga.py index e1ee909f2..368d1ebcc 100644 --- a/ga.py +++ b/ga.py @@ -1,4 +1,5 @@ -import sys, os, re, json, time, threading, importlib, webbrowser +import sys, os, re, json, time, threading, importlib, webbrowser, hashlib, weakref +from contextlib import contextmanager from datetime import datetime from pathlib import Path import tempfile, traceback, subprocess, itertools, collections, difflib, shutil @@ -209,16 +210,47 @@ def file_patch(path: str, old_content: str, new_content: str): path = str(Path(path).resolve()) try: if not os.path.exists(path): return {"status": "error", "msg": "file not found"} - with open(path, 'r', encoding='utf-8') as f: full_text = f.read() if not old_content: return {"status": "error", "msg": "old_content is blank"} - count = full_text.count(old_content) - if count == 0: return {"status": "error", "msg": "old_content is not found. Suggestion: use file_read to check current file content, make more small patches. Don't huge overwrite (even with code)"} - if count > 1: return {"status": "error", "msg": f"find {count} matches, unable to determine unique position. Provide a longer, more specific old_content to ensure uniqueness. Suggestion: include context lines to enhance features, or modify in smaller segments."} - updated_text = full_text.replace(old_content, new_content) - with open(path, 'w', encoding='utf-8', newline=_file_newline(path)) as f: f.write(updated_text) - return {"status": "success", "msg": "file patched successfully"} + with _file_patch_lock(path): + with open(path, 'r', encoding='utf-8') as f: full_text = f.read() + count = full_text.count(old_content) + if count == 0: return {"status": "error", "msg": "old_content is not found. Suggestion: use file_read to check current file content, make more small patches. Don't huge overwrite (even with code)"} + if count > 1: return {"status": "error", "msg": f"find {count} matches, unable to determine unique position. Provide a longer, more specific old_content to ensure uniqueness. Suggestion: include context lines to enhance features, or modify in smaller segments."} + updated_text = full_text.replace(old_content, new_content) + with open(path, 'w', encoding='utf-8', newline=_file_newline(path)) as f: f.write(updated_text) + return {"status": "success", "msg": "file patched successfully"} except Exception as e: return {"status": "error", "msg": str(e)} +_file_patch_thread_locks = weakref.WeakValueDictionary() +_file_patch_thread_locks_guard = threading.Lock() + +def _file_patch_thread_lock(path): + with _file_patch_thread_locks_guard: + return _file_patch_thread_locks.setdefault(path, threading.Lock()) + +@contextmanager +def _file_patch_lock(path): + lock_dir = os.path.join(tempfile.gettempdir(), 'genericagent-file-locks') + os.makedirs(lock_dir, exist_ok=True) + lock_path = os.path.join(lock_dir, hashlib.sha256(path.encode('utf-8')).hexdigest() + '.lock') + with _file_patch_thread_lock(path): + with open(lock_path, 'a+b') as lock: + if not os.path.getsize(lock_path): lock.write(b'0'); lock.flush() + lock.seek(0) + if os.name == 'nt': + import msvcrt + msvcrt.locking(lock.fileno(), msvcrt.LK_LOCK, 1) + else: + import fcntl + fcntl.flock(lock.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + if os.name == 'nt': + lock.seek(0); msvcrt.locking(lock.fileno(), msvcrt.LK_UNLCK, 1) + else: + fcntl.flock(lock.fileno(), fcntl.LOCK_UN) + _read_dirs = set() def _scan_files(base, depth=2): try: