fix(file-patch): serialize concurrent updates

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"

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: