fix(utils): append_to_jsonl serialisation across threads

adalflow/adalflow/utils/file_io.py

@@ -2,13 +2,18 @@
 import os
 import pickle
 import logging
+import threading
 from typing import Mapping, Any, Optional, List, Dict
 
 
 from adalflow.utils.serialization import to_dict, serialize, _deserialize_object_hook
 
 log = logging.getLogger(__name__)
 
+# Serializes appends so that threads (e.g. trainer workers) cannot interleave
+# partial lines in the same jsonl file.
+_JSONL_APPEND_LOCK = threading.Lock()
+
 
 def save_json(obj: Mapping[str, Any], f: str = "task.json") -> None:
     """Customized Save the object to a json file.
@@ -216,6 +221,8 @@ def append_to_jsonl(f: str, data: Dict[str, Any]) -> None:
     r"""Append data to a jsonl file.
 
     Used by the trace_generator_call decorator to log the generator calls.
+    Appends from multiple threads in one process are serialized; concurrent
+    writers in separate processes are not coordinated.
 
     Args:
         f (str): The file name.
@@ -227,11 +234,11 @@ def append_to_jsonl(f: str, data: Dict[str, Any]) -> None:
         raise ImportError("Please install jsonlines to use this function.")
     os.makedirs(os.path.dirname(f) or ".", exist_ok=True)
     try:
-        with jsonlines.open(f, mode="a") as writer:
-            # call serialize to serialize the object
-            serialized_data = to_dict(data)
-            writer.write(serialized_data)
-            # writer.write(data)
+        # call serialize to serialize the object
+        serialized_data = to_dict(data)
+        with _JSONL_APPEND_LOCK:
+            with jsonlines.open(f, mode="a") as writer:
+                writer.write(serialized_data)
     except Exception as e:
         log.error(f"Error appending data to jsonl file {f}: {e}")
 

adalflow/tests/test_file_io.py

@@ -0,0 +1,26 @@
+import json
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+pytest.importorskip("jsonlines")
+
+from adalflow.utils.file_io import append_to_jsonl
+
+
+def test_append_to_jsonl_is_thread_safe(tmp_path):
+    """Concurrent appends must each produce one complete json line."""
+    path = str(tmp_path / "calls.jsonl")
+    n = 200
+    payload = "x" * 20000
+
+    def write(i):
+        append_to_jsonl(path, {"id": i, "text": payload})
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(write, range(n)))
+
+    with open(path, encoding="utf-8") as f:
+        records = [json.loads(line) for line in f]
+
+    assert sorted(r["id"] for r in records) == list(range(n))