diff --git a/tests/test_cache_writes.py b/tests/test_cache_writes.py index 195809038..a9bf7d94e 100644 --- a/tests/test_cache_writes.py +++ b/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 diff --git a/tests/test_memory_log.py b/tests/test_memory_log.py index bcd955eb7..73876d299 100644 --- a/tests/test_memory_log.py +++ b/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) diff --git a/tradingagents/dataflows/files.py b/tradingagents/dataflows/files.py index 98e32afbd..0c7408919 100644 --- a/tradingagents/dataflows/files.py +++ b/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) diff --git a/tradingagents/memory/log.py b/tradingagents/memory/log.py index e06a1afa7..dc1622ccf 100644 --- a/tradingagents/memory/log.py +++ b/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 ---