fix(memory): let one writer at a time change the memory log

- storing a decision and settling outcomes hold a lock across the read, change and write, so runs sharing a log no longer drop each other's entries
- files.locked takes the lock on a .lock file beside the log, across threads and processes

tests/test_cache_writes.py

@@ -7,6 +7,7 @@ from pathlib import Path
 import pandas as pd
 import pytest
 
+from tradingagents.dataflows import files
 from tradingagents.dataflows.vendors import sec_edgar
 from tradingagents.dataflows.vendors.yahoo import ohlcv
 
@@ -82,3 +83,46 @@ def test_a_file_held_open_elsewhere_keeps_its_old_content(monkeypatch, tmp_path)
 
     assert target.read_text() == "old"
     assert list(tmp_path.iterdir()) == [target]
+
+
+def _fake_msvcrt(results):
+    """msvcrt whose locking() raises the next error in ``results``, or succeeds on None."""
+    import types
+
+    calls = []
+
+    def locking(fd, mode, nbytes):
+        calls.append(mode)
+        if mode == 2 and results:   # LK_LOCK; unlocking always succeeds
+            error = results.pop(0)
+            if error is not None:
+                raise error
+
+    return types.SimpleNamespace(LK_LOCK=2, LK_UNLCK=0, locking=locking), calls
+
+
+@pytest.mark.unit
+def test_on_windows_the_lock_is_waited_for_while_another_writer_holds_it(tmp_path, monkeypatch):
+    import errno
+    import sys
+
+    fake, calls = _fake_msvcrt([OSError(errno.EDEADLOCK, "held"), None])
+    monkeypatch.setitem(sys.modules, "msvcrt", fake)
+
+    with open(tmp_path / "log.md.lock", "a+b") as handle, files._hold_windows(handle):
+        pass
+
+    assert calls == [2, 2, 0]
+
+
+@pytest.mark.unit
+def test_on_windows_a_lock_that_cannot_be_taken_is_reported_not_retried_forever(tmp_path, monkeypatch):
+    import errno
+    import sys
+
+    fake, _ = _fake_msvcrt([OSError(errno.EINVAL, "bad handle")] * 3)
+    monkeypatch.setitem(sys.modules, "msvcrt", fake)
+
+    with pytest.raises(OSError, match="bad handle"), open(tmp_path / "log.md.lock", "a+b") as handle, \
+            files._hold_windows(handle):
+        pass

tests/test_memory_log.py

@@ -1029,3 +1029,34 @@ def test_a_longer_window_asks_for_enough_price_history(monkeypatch):
     raw, alpha, days, resolved = settlement.fetch_returns("NVDA", "2026-06-01", 21, benchmark="SPY")
 
     assert days == 21 and resolved is not None, (raw, alpha, days, resolved)
+
+
+@pytest.mark.unit
+def test_concurrent_writers_keep_every_decision_and_outcome(tmp_path):
+    """Graphs sharing one log record and settle at the same time; none may lose another's write."""
+    import threading
+
+    path = tmp_path / "memory.md"
+    tickers = [f"T{i}" for i in range(8)]
+    dates = [f"2026-01-{day:02d}" for day in range(1, 13)]
+    errors = []
+
+    def work(ticker):
+        log = TradingMemoryLog({"memory_log_path": str(path)})
+        try:
+            for date in dates:
+                log.store_decision(ticker, date, "**Rating**: Buy")
+                log.update_with_outcome(ticker, date, 0.01, 0.005, 5, "held up")
+        except Exception as exc:  # noqa: BLE001 — collected for the assertion
+            errors.append(exc)
+
+    threads = [threading.Thread(target=work, args=(t,)) for t in tickers]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    entries = TradingMemoryLog({"memory_log_path": str(path)}).load_entries()
+    assert errors == []
+    assert len(entries) == len(tickers) * len(dates)
+    assert all(not e["pending"] for e in entries)

tradingagents/dataflows/files.py

@@ -1,9 +1,11 @@
-"""Cache files written whole, safe under concurrent writers."""
+"""Files written whole, and held by one writer at a time, under concurrent writers."""
 
+import errno
 import logging
 import os
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from pathlib import Path
 
 logger = logging.getLogger(__name__)
@@ -29,3 +31,49 @@ def replace_file(path, write: Callable[[str], None]) -> None:
     except BaseException:
         temp.unlink(missing_ok=True)
         raise
+
+
+@contextmanager
+def locked(path) -> Iterator[None]:
+    """Hold ``path`` for one writer at a time, across threads and processes.
+
+    For a file read, changed and written back: without the lock, two writers
+    each read the same text and the second write drops the first's change. The
+    lock is taken on a ``.lock`` file beside ``path``.
+    """
+    with open(f"{path}.lock", "a+b") as handle:
+        hold = _hold_windows if os.name == "nt" else _hold_posix
+        with hold(handle):
+            yield
+
+
+@contextmanager
+def _hold_posix(handle) -> Iterator[None]:
+    import fcntl
+
+    fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+    try:
+        yield
+    finally:
+        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+
+
+@contextmanager
+def _hold_windows(handle) -> Iterator[None]:
+    import msvcrt
+
+    handle.seek(0)
+    while True:
+        try:
+            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
+            break
+        except OSError as exc:
+            # LK_LOCK gives up after about ten seconds of another writer's hold;
+            # any other failure is not a wait.
+            if exc.errno != errno.EDEADLOCK:
+                raise
+    try:
+        yield
+    finally:
+        handle.seek(0)
+        msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)

tradingagents/memory/log.py

@@ -4,6 +4,7 @@ import re
 from pathlib import Path
 
 from tradingagents.agents.rating import parse_rating
+from tradingagents.dataflows.files import locked
 
 
 class TradingMemoryLog:
@@ -41,20 +42,21 @@ class TradingMemoryLog:
         """
         if not self._log_path:
             return
-        # Idempotency guard: fast raw-text scan instead of full parse. Any entry
-        # for this ticker and date blocks another, pending or settled: a re-run
-        # after the outcome landed would otherwise count the same decision twice
-        # in past context and in every aggregate over the log.
-        if self._log_path.exists():
-            raw = self._log_path.read_text(encoding="utf-8")
-            for line in raw.splitlines():
-                if line.startswith(f"[{trade_date} | {ticker} |") and line.endswith("]"):
-                    return
-        rating = rating or parse_rating(final_trade_decision)
-        tag = f"[{trade_date} | {ticker} | {rating} | pending]"
-        entry = f"{tag}\n\nDECISION:\n{final_trade_decision}{self._SEPARATOR}"
-        with open(self._log_path, "a", encoding="utf-8") as f:
-            f.write(entry)
+        with locked(self._log_path):
+            # Idempotency guard: fast raw-text scan instead of full parse. Any entry
+            # for this ticker and date blocks another, pending or settled: a re-run
+            # after the outcome landed would otherwise count the same decision twice
+            # in past context and in every aggregate over the log.
+            if self._log_path.exists():
+                raw = self._log_path.read_text(encoding="utf-8")
+                for line in raw.splitlines():
+                    if line.startswith(f"[{trade_date} | {ticker} |") and line.endswith("]"):
+                        return
+            rating = rating or parse_rating(final_trade_decision)
+            tag = f"[{trade_date} | {ticker} | {rating} | pending]"
+            entry = f"{tag}\n\nDECISION:\n{final_trade_decision}{self._SEPARATOR}"
+            with open(self._log_path, "a", encoding="utf-8") as f:
+                f.write(entry)
 
     # --- Read ---
 
@@ -135,51 +137,51 @@ class TradingMemoryLog:
         """
         if not self._log_path or not self._log_path.exists():
             return
+        with locked(self._log_path):
+            text = self._log_path.read_text(encoding="utf-8")
+            blocks = text.split(self._SEPARATOR)
 
-        text = self._log_path.read_text(encoding="utf-8")
-        blocks = text.split(self._SEPARATOR)
+            pending_prefix = f"[{trade_date} | {ticker} |"
+            raw_pct = f"{raw_return:+.1%}"
+            alpha_pct = f"{alpha_return:+.1%}"
 
-        pending_prefix = f"[{trade_date} | {ticker} |"
-        raw_pct = f"{raw_return:+.1%}"
-        alpha_pct = f"{alpha_return:+.1%}"
+            updated = False
+            new_blocks = []
+            for block in blocks:
+                stripped = block.strip()
+                if not stripped:
+                    new_blocks.append(block)
+                    continue
 
-        updated = False
-        new_blocks = []
-        for block in blocks:
-            stripped = block.strip()
-            if not stripped:
-                new_blocks.append(block)
-                continue
+                lines = stripped.splitlines()
+                tag_line = lines[0].strip()
 
-            lines = stripped.splitlines()
-            tag_line = lines[0].strip()
+                if (
+                    not updated
+                    and tag_line.startswith(pending_prefix)
+                    and tag_line.endswith("| pending]")
+                ):
+                    fields = [f.strip() for f in tag_line[1:-1].split("|")]
+                    rating = fields[2]
+                    new_tag = self._resolved_tag(
+                        trade_date, ticker, rating, raw_pct, alpha_pct, holding_days, resolution_date
+                    )
+                    rest = "\n".join(lines[1:])
+                    new_blocks.append(
+                        f"{new_tag}\n\n{rest.lstrip()}\n\nREFLECTION:\n{reflection}"
+                    )
+                    updated = True
+                else:
+                    new_blocks.append(block)
 
-            if (
-                not updated
-                and tag_line.startswith(pending_prefix)
-                and tag_line.endswith("| pending]")
-            ):
-                fields = [f.strip() for f in tag_line[1:-1].split("|")]
-                rating = fields[2]
-                new_tag = self._resolved_tag(
-                    trade_date, ticker, rating, raw_pct, alpha_pct, holding_days, resolution_date
-                )
-                rest = "\n".join(lines[1:])
-                new_blocks.append(
-                    f"{new_tag}\n\n{rest.lstrip()}\n\nREFLECTION:\n{reflection}"
-                )
-                updated = True
-            else:
-                new_blocks.append(block)
+            if not updated:
+                return
 
-        if not updated:
-            return
-
-        new_blocks = self._apply_rotation(new_blocks)
-        new_text = self._SEPARATOR.join(new_blocks)
-        tmp_path = self._log_path.with_suffix(".tmp")
-        tmp_path.write_text(new_text, encoding="utf-8")
-        tmp_path.replace(self._log_path)
+            new_blocks = self._apply_rotation(new_blocks)
+            new_text = self._SEPARATOR.join(new_blocks)
+            tmp_path = self._log_path.with_suffix(".tmp")
+            tmp_path.write_text(new_text, encoding="utf-8")
+            tmp_path.replace(self._log_path)
 
     def batch_update_with_outcomes(self, updates: list[dict]) -> None:
         """Apply multiple outcome updates in a single read + atomic write.
@@ -189,50 +191,50 @@ class TradingMemoryLog:
         """
         if not self._log_path or not self._log_path.exists() or not updates:
             return
+        with locked(self._log_path):
+            text = self._log_path.read_text(encoding="utf-8")
+            blocks = text.split(self._SEPARATOR)
 
-        text = self._log_path.read_text(encoding="utf-8")
-        blocks = text.split(self._SEPARATOR)
+            update_map = {(u["trade_date"], u["ticker"]): u for u in updates}
 
-        update_map = {(u["trade_date"], u["ticker"]): u for u in updates}
+            new_blocks = []
+            for block in blocks:
+                stripped = block.strip()
+                if not stripped:
+                    new_blocks.append(block)
+                    continue
 
-        new_blocks = []
-        for block in blocks:
-            stripped = block.strip()
-            if not stripped:
-                new_blocks.append(block)
-                continue
+                lines = stripped.splitlines()
+                tag_line = lines[0].strip()
 
-            lines = stripped.splitlines()
-            tag_line = lines[0].strip()
+                matched = False
+                for (trade_date, ticker), upd in list(update_map.items()):
+                    pending_prefix = f"[{trade_date} | {ticker} |"
+                    if tag_line.startswith(pending_prefix) and tag_line.endswith("| pending]"):
+                        fields = [f.strip() for f in tag_line[1:-1].split("|")]
+                        rating = fields[2]
+                        raw_pct = f"{upd['raw_return']:+.1%}"
+                        alpha_pct = f"{upd['alpha_return']:+.1%}"
+                        new_tag = self._resolved_tag(
+                            trade_date, ticker, rating, raw_pct, alpha_pct,
+                            upd["holding_days"], upd.get("resolution_date"),
+                        )
+                        rest = "\n".join(lines[1:])
+                        new_blocks.append(
+                            f"{new_tag}\n\n{rest.lstrip()}\n\nREFLECTION:\n{upd['reflection']}"
+                        )
+                        del update_map[(trade_date, ticker)]
+                        matched = True
+                        break
 
-            matched = False
-            for (trade_date, ticker), upd in list(update_map.items()):
-                pending_prefix = f"[{trade_date} | {ticker} |"
-                if tag_line.startswith(pending_prefix) and tag_line.endswith("| pending]"):
-                    fields = [f.strip() for f in tag_line[1:-1].split("|")]
-                    rating = fields[2]
-                    raw_pct = f"{upd['raw_return']:+.1%}"
-                    alpha_pct = f"{upd['alpha_return']:+.1%}"
-                    new_tag = self._resolved_tag(
-                        trade_date, ticker, rating, raw_pct, alpha_pct,
-                        upd["holding_days"], upd.get("resolution_date"),
-                    )
-                    rest = "\n".join(lines[1:])
-                    new_blocks.append(
-                        f"{new_tag}\n\n{rest.lstrip()}\n\nREFLECTION:\n{upd['reflection']}"
-                    )
-                    del update_map[(trade_date, ticker)]
-                    matched = True
-                    break
+                if not matched:
+                    new_blocks.append(block)
 
-            if not matched:
-                new_blocks.append(block)
-
-        new_blocks = self._apply_rotation(new_blocks)
-        new_text = self._SEPARATOR.join(new_blocks)
-        tmp_path = self._log_path.with_suffix(".tmp")
-        tmp_path.write_text(new_text, encoding="utf-8")
-        tmp_path.replace(self._log_path)
+            new_blocks = self._apply_rotation(new_blocks)
+            new_text = self._SEPARATOR.join(new_blocks)
+            tmp_path = self._log_path.with_suffix(".tmp")
+            tmp_path.write_text(new_text, encoding="utf-8")
+            tmp_path.replace(self._log_path)
 
     # --- Helpers ---