fix(memory): settle each decision once, and keep each outcome as soon as it is known

- a settlement pass holds the log's settlement lock, so runs sharing a log never pay for the same reflection twice; only outcomes actually written count as settled, and a run's Memory Log step does not wait for another run's pass
- each outcome is written when its reflection is done, so an interrupted pass keeps what it finished
- any failure to price a decision is reported as a failed settlement, not taken for a window that has yet to trade

tests/test_cache_writes.py

@@ -93,12 +93,12 @@ def _fake_msvcrt(results):
 
     def locking(fd, mode, nbytes):
         calls.append(mode)
-        if mode == 2 and results:   # LK_LOCK; unlocking always succeeds
+        if mode in (1, 2) and results:   # LK_NBLCK, 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
+    return types.SimpleNamespace(LK_LOCK=2, LK_NBLCK=1, LK_UNLCK=0, locking=locking), calls
 
 
 @pytest.mark.unit
@@ -126,3 +126,39 @@ def test_on_windows_a_lock_that_cannot_be_taken_is_reported_not_retried_forever(
     with pytest.raises(OSError, match="bad handle"), open(tmp_path / "log.md.lock", "a+b") as handle, \
             files._hold_windows(handle):
         pass
+
+
+@pytest.mark.unit
+def test_a_lock_held_elsewhere_is_reported_when_not_waiting(tmp_path):
+    import threading
+
+    path = tmp_path / "log.md"
+    held, release = threading.Event(), threading.Event()
+
+    def holder():
+        with files.locked(path):
+            held.set()
+            release.wait(5)
+
+    worker = threading.Thread(target=holder)
+    worker.start()
+    held.wait(5)
+    with files.locked(path, wait=False) as acquired:
+        assert acquired is False
+    release.set()
+    worker.join()
+    with files.locked(path, wait=False) as acquired:
+        assert acquired is True
+
+
+@pytest.mark.unit
+def test_on_windows_a_held_lock_is_reported_when_not_waiting(tmp_path, monkeypatch):
+    import errno
+    import sys
+
+    fake, calls = _fake_msvcrt([OSError(errno.EACCES, "held")])
+    monkeypatch.setitem(sys.modules, "msvcrt", fake)
+
+    with open(tmp_path / "log.md.lock", "a+b") as handle, files._hold_windows(handle, wait=False) as acquired:
+        assert acquired is False
+    assert calls == [1]                # one try, no unlock of a lock it never held

tests/test_memory_log.py

@@ -1183,8 +1183,118 @@ def test_the_graph_settles_every_ticker_under_its_own_config(tmp_path, monkeypat
     graph.memory_log, graph.reflector = make_log(tmp_path), object()
     seen = []
     monkeypatch.setattr(settlement, "settle_all_pending",
-                        lambda log, reflector, config: seen.append(get_config()["output_language"])
+                        lambda log, reflector, config, wait=True: seen.append(get_config()["output_language"])
                         or settlement.Settlement(settled=[("AAPL", "2026-01-05")]))
 
     assert graph.settle_all_pending().settled == [("AAPL", "2026-01-05")]
     assert seen == ["Deutsch"]
+
+
+@pytest.mark.unit
+class TestSettlementOnceAndDurably:
+    """Settlement is paid for in reflection calls, so it runs once per decision and
+    keeps what it finished; what it could not do is reported (#1428, #1445)."""
+
+    def _log(self, tmp_path, monkeypatch, *dates):
+        log = make_log(tmp_path)
+        for date in dates:
+            log.store_decision("AAPL", date, DECISION_BUY)
+        series = pd.Series([100.0 + i for i in range(40)], index=pd.bdate_range("2026-01-02", periods=40))
+        monkeypatch.setattr(settlement, "get_closes", _closes({"AAPL": series, "SPY": series}))
+        return log
+
+    def test_two_runs_settling_at_once_reflect_once(self, tmp_path, monkeypatch):
+        import threading
+        import time
+
+        self._log(tmp_path, monkeypatch, "2026-01-05")
+        calls = []
+
+        class _Slow:
+            def reflect_on_final_decision(self, **kwargs):
+                calls.append(1)
+                time.sleep(0.2)
+                return "Held up."
+
+        results = []
+        threads = [threading.Thread(target=lambda: results.append(
+            settlement.settle_pending("AAPL", make_log(tmp_path), _Slow(), {"holding_period_days": 5})))
+            for _ in range(2)]
+        [t.start() for t in threads]
+        [t.join() for t in threads]
+
+        assert len(calls) == 1
+        assert sorted(len(r.settled) for r in results) == [0, 1]
+
+    def test_an_interrupted_pass_keeps_the_outcomes_it_finished(self, tmp_path, monkeypatch):
+        log = self._log(tmp_path, monkeypatch, "2026-01-05", "2026-01-12")
+
+        class _Interrupted:
+            calls = 0
+
+            def reflect_on_final_decision(self, **kwargs):
+                _Interrupted.calls += 1
+                if _Interrupted.calls == 2:
+                    raise KeyboardInterrupt
+                return "Held up."
+
+        with pytest.raises(KeyboardInterrupt):
+            settlement.settle_pending("AAPL", log, _Interrupted(), {"holding_period_days": 5})
+
+        assert [e["date"] for e in log.get_pending_entries()] == ["2026-01-12"]
+
+    def test_a_price_request_that_fails_is_reported_not_taken_for_not_due(self, tmp_path, monkeypatch):
+        from tradingagents.dataflows.errors import VendorUnavailableError
+
+        log = self._log(tmp_path, monkeypatch, "2026-01-05")
+
+        def down(*args, **kwargs):
+            raise VendorUnavailableError("Yahoo Finance request failed: TimeoutError")
+
+        monkeypatch.setattr(settlement, "get_closes", down)
+
+        result = settlement.settle_pending("AAPL", log, object(), {"holding_period_days": 5})
+
+        assert result.settled == [] and "prices unavailable" in result.failed[0][2]
+
+
+@pytest.mark.unit
+def test_any_failure_to_price_a_decision_is_reported(tmp_path, monkeypatch):
+    log = make_log(tmp_path)
+    log.store_decision("AAPL", "2026-01-05", DECISION_BUY)
+
+    def broken(*args, **kwargs):
+        raise RuntimeError("vendor outage")
+
+    monkeypatch.setattr(settlement, "get_closes", broken)
+
+    result = settlement.settle_pending("AAPL", log, object(), {"holding_period_days": 5})
+
+    assert result.settled == [] and "vendor outage" in result.failed[0][2]
+
+
+@pytest.mark.unit
+def test_a_run_does_not_wait_on_another_settlement_pass(tmp_path):
+    """The other pass settles the same log; a run's step goes on without waiting."""
+    import threading
+    import time
+
+    log = make_log(tmp_path)
+    log.store_decision("AAPL", "2026-01-05", DECISION_BUY)
+    held, release = threading.Event(), threading.Event()
+
+    def other_pass():
+        with make_log(tmp_path).settling():
+            held.set()
+            release.wait(5)
+
+    worker = threading.Thread(target=other_pass)
+    worker.start()
+    held.wait(5)
+    started = time.monotonic()
+    result = settlement.settle_all_pending(log, object(), {"holding_period_days": 5}, wait=False)
+    elapsed = time.monotonic() - started
+    release.set()
+    worker.join()
+
+    assert elapsed < 1 and result.settled == [] and result.failed == []

tradingagents/dataflows/files.py

@@ -34,46 +34,55 @@ def replace_file(path, write: Callable[[str], None]) -> None:
 
 
 @contextmanager
-def locked(path) -> Iterator[None]:
+def locked(path, wait: bool = True) -> Iterator[bool]:
     """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``.
+    lock is taken on a ``.lock`` file beside ``path``. Yields whether it is
+    held: with ``wait=False`` a lock someone else holds is not waited for, and
+    the block runs holding nothing.
     """
     with open(f"{path}.lock", "a+b") as handle:
         hold = _hold_windows if os.name == "nt" else _hold_posix
-        with hold(handle):
-            yield
+        with hold(handle, wait) as acquired:
+            yield acquired
 
 
 @contextmanager
-def _hold_posix(handle) -> Iterator[None]:
+def _hold_posix(handle, wait: bool = True) -> Iterator[bool]:
     import fcntl
 
-    fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
     try:
-        yield
+        fcntl.flock(handle.fileno(), fcntl.LOCK_EX if wait else fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        yield False
+        return
+    try:
+        yield True
     finally:
         fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
 
 
 @contextmanager
-def _hold_windows(handle) -> Iterator[None]:
+def _hold_windows(handle, wait: bool = True) -> Iterator[bool]:
     import msvcrt
 
     handle.seek(0)
     while True:
         try:
-            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK if wait else msvcrt.LK_NBLCK, 1)
             break
         except OSError as exc:
+            if not wait and exc.errno in (errno.EACCES, errno.EDEADLOCK):
+                yield False   # another writer holds it
+                return
             # 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
+        yield True
     finally:
         handle.seek(0)
         msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)

tradingagents/graph/trading_graph.py

@@ -317,7 +317,8 @@ class TradingAgentsGraph:
         """
         note = ""
         try:
-            done = self.settle_all_pending()
+            # Another run settling the same log does the work; this one goes on.
+            done = self.settle_all_pending(wait=False)
             if done.failed:
                 note = (f"{len(done.failed)} past decision(s) could not be settled this run "
                         "and stay pending.")
@@ -344,14 +345,15 @@ class TradingAgentsGraph:
         with run_config(self.config):
             return settlement.settle_pending(company_name, self.memory_log, self.reflector, self.config)
 
-    def settle_all_pending(self) -> settlement.Settlement:
+    def settle_all_pending(self, wait: bool = True) -> settlement.Settlement:
         """Settle every ticker's decisions whose holding window has now traded (#1445).
 
         For a scheduler whose tickers rotate: a ticker it stops analysing would
         otherwise keep its decisions pending, and their lessons out of later runs.
+        With ``wait=False`` a pass already running on the same log is not waited for.
         """
         with run_config(self.config):
-            return settlement.settle_all_pending(self.memory_log, self.reflector, self.config)
+            return settlement.settle_all_pending(self.memory_log, self.reflector, self.config, wait=wait)
 
     def record_decision(self, company_name, trade_date, final_state):
         """Record a finished run: its state log, and its decision in the memory log

tradingagents/memory/log.py

@@ -1,5 +1,6 @@
 """The memory log: an append-only markdown record of each decision and, once settled, its outcome."""
 
+import contextlib
 import re
 from pathlib import Path
 
@@ -58,6 +59,18 @@ class TradingMemoryLog:
             with open(self._log_path, "a", encoding="utf-8") as f:
                 f.write(entry)
 
+    def settling(self, wait: bool = True):
+        """Hold the log for one settlement pass at a time, across threads and processes.
+
+        A pass reads the pending entries, asks the model for a reflection on
+        each and writes it; two passes at once would pay for the same
+        reflections. A separate lock from the one each write takes. Yields
+        whether it is held; with ``wait=False`` another pass is not waited for.
+        """
+        if not self._log_path:
+            return contextlib.nullcontext(True)
+        return locked(f"{self._log_path}.settle", wait=wait)
+
     # --- Read ---
 
     def load_entries(self) -> list[dict]:
@@ -127,16 +140,19 @@ class TradingMemoryLog:
         holding_days: int,
         reflection: str,
         resolution_date: str | None = None,
-    ) -> None:
+    ) -> bool:
         """Replace pending tag and append REFLECTION section using atomic write.
 
+        Returns whether an entry was updated: False when the decision was no
+        longer pending (another pass settled it first).
+
         Finds the first pending entry matching (trade_date, ticker), updates
         its tag with return figures (and the ``resolution_date`` the outcome
         became known), and appends a REFLECTION section.  Uses a temp-file +
         os.replace() so a crash mid-write never corrupts the log.
         """
         if not self._log_path or not self._log_path.exists():
-            return
+            return False
         with locked(self._log_path):
             text = self._log_path.read_text(encoding="utf-8")
             blocks = text.split(self._SEPARATOR)
@@ -175,13 +191,14 @@ class TradingMemoryLog:
                     new_blocks.append(block)
 
             if not updated:
-                return
+                return False
 
             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)
+            return True
 
     def batch_update_with_outcomes(self, updates: list[dict]) -> None:
         """Apply multiple outcome updates in a single read + atomic write.

tradingagents/memory/settlement.py

@@ -59,50 +59,43 @@ def fetch_returns(
     caller via ``resolve_benchmark``). Returns ``(raw_return, alpha_return,
     holding_days, resolution_date)`` — where ``resolution_date`` is the date
     of the last price bar used, i.e. when the outcome became known (#1251) —
-    or ``(None, None, None, None)`` when the outcome cannot be settled yet:
-    the full holding window has not traded (#1169), or the symbol is delisted
-    or unreachable.
+    or ``(None, None, None, None)`` when the full holding window has not traded
+    yet (#1169). A failure to get or read the prices is raised, for the caller
+    to report.
     """
-    try:
-        start = datetime.strptime(trade_date, "%Y-%m-%d")
-        # holding_days counts trading days, so ask for the calendar span they
-        # occupy (about 7 for every 5) plus a week for holidays.
-        end = start + timedelta(days=round(holding_days * 7 / 5) + 7)
-        end_str = end.strftime("%Y-%m-%d")
+    start = datetime.strptime(trade_date, "%Y-%m-%d")
+    # holding_days counts trading days, so ask for the calendar span they
+    # occupy (about 7 for every 5) plus a week for holidays.
+    end = start + timedelta(days=round(holding_days * 7 / 5) + 7)
+    end_str = end.strftime("%Y-%m-%d")
 
-        # Closes for the instrument the analysis priced (XAUUSD -> GC=F, #984).
-        stock = _by_day(get_closes(ticker, trade_date, end_str))
-        # From a week earlier, so the benchmark has a close on or before entry.
-        bench_start = (start - timedelta(days=7)).strftime("%Y-%m-%d")
-        bench = _by_day(get_closes(benchmark, bench_start, end_str))
+    # Closes for the instrument the analysis priced (XAUUSD -> GC=F, #984).
+    stock = _by_day(get_closes(ticker, trade_date, end_str))
+    # From a week earlier, so the benchmark has a close on or before entry.
+    bench_start = (start - timedelta(days=7)).strftime("%Y-%m-%d")
+    bench = _by_day(get_closes(benchmark, bench_start, end_str))
 
-        # Require the full holding window to have traded. A rerun before it has
-        # leaves the entry pending to retry next run, rather than settling on a
-        # premature partial return (#1169).
-        if len(stock) <= holding_days:
-            return None, None, None, None
-        entry, exit_ = stock.index[0], stock.index[holding_days]
-        # The window is the stock's own sessions; the benchmark is valued at its
-        # last close by each end, so both returns span the same dates even when
-        # the calendars differ (a coin trades at weekends, an index does not).
-        # It must have traded through the exit, or its close there may yet move.
-        if bench.empty or bench.index[0] > entry or bench.index[-1] < exit_:
-            return None, None, None, None
-
-        raw = float((stock.iloc[holding_days] - stock.iloc[0]) / stock.iloc[0])
-        bench_entry, bench_exit = bench.asof(entry), bench.asof(exit_)
-        bench_ret = float((bench_exit - bench_entry) / bench_entry)
-        alpha = raw - bench_ret
-        # Every close used is known by the exit: the point-in-time cutoff for
-        # injecting the lesson (#1251).
-        resolution_date = exit_.strftime("%Y-%m-%d")
-        return raw, alpha, holding_days, resolution_date
-    except Exception as e:
-        logger.warning(
-            "Could not resolve outcome for %s on %s vs %s (will retry next run): %s",
-            ticker, trade_date, benchmark, e,
-        )
+    # Require the full holding window to have traded. A rerun before it has
+    # leaves the entry pending to retry next run, rather than settling on a
+    # premature partial return (#1169).
+    if len(stock) <= holding_days:
         return None, None, None, None
+    entry, exit_ = stock.index[0], stock.index[holding_days]
+    # The window is the stock's own sessions; the benchmark is valued at its
+    # last close by each end, so both returns span the same dates even when
+    # the calendars differ (a coin trades at weekends, an index does not).
+    # It must have traded through the exit, or its close there may yet move.
+    if bench.empty or bench.index[0] > entry or bench.index[-1] < exit_:
+        return None, None, None, None
+
+    raw = float((stock.iloc[holding_days] - stock.iloc[0]) / stock.iloc[0])
+    bench_entry, bench_exit = bench.asof(entry), bench.asof(exit_)
+    bench_ret = float((bench_exit - bench_entry) / bench_entry)
+    alpha = raw - bench_ret
+    # Every close used is known by the exit: the point-in-time cutoff for
+    # injecting the lesson (#1251).
+    resolution_date = exit_.strftime("%Y-%m-%d")
+    return raw, alpha, holding_days, resolution_date
 
 
 @dataclass
@@ -113,59 +106,56 @@ class Settlement:
     failed: list[tuple[str, str, str]] = field(default_factory=list)    # (ticker, date, reason)
 
 
-def settle_pending(ticker: str, memory_log, reflector, config: dict) -> Settlement:
+def settle_pending(ticker: str, memory_log, reflector, config: dict, wait: bool = True) -> Settlement:
     """Settle ``ticker``'s pending decisions whose holding window has traded.
 
-    Fetches returns for each same-ticker pending entry, generates reflections,
-    then writes all updates in a single atomic batch write to avoid redundant I/O.
-    Skips entries whose price data is not yet available (too recent or delisted);
-    an entry whose reflection fails stays pending and is reported as failed.
+    One pass at a time per log (``memory_log.settling()``), so runs sharing a log
+    never pay for the same reflection twice. Each outcome is written as soon as
+    its reflection is done, so an interrupted pass keeps what it finished.
+    Entries whose window has not traded yet are skipped; an entry whose prices
+    or reflection could not be had stays pending and is reported as failed.
+    With ``wait=False`` a pass already running elsewhere is left to do the work.
     """
     result = Settlement()
-    pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker]
-    if not pending:
-        return result
-
-    benchmark = resolve_benchmark(ticker, config)
-    updates = []
-    for entry in pending:
-        raw, alpha, days, resolution_date = fetch_returns(
-            ticker, entry["date"], config.get("holding_period_days", 5),
-            benchmark=benchmark,
-        )
-        if raw is None:
-            continue  # price not available yet — try again next run
-        try:
-            reflection = reflector.reflect_on_final_decision(
-                final_decision=entry.get("decision", ""),
-                raw_return=raw,
-                alpha_return=alpha,
-                benchmark_name=benchmark,
-                holding_days=days,
-            )
-        except Exception as exc:
-            # Reflection calls a provider: a transient failure leaves the entry
-            # pending for the next pass rather than stopping the analysis.
-            logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc)
-            result.failed.append((ticker, entry["date"], f"reflection failed: {exc}"))
-            continue
-        updates.append({
-            "ticker": ticker,
-            "trade_date": entry["date"],
-            "raw_return": raw,
-            "alpha_return": alpha,
-            "holding_days": days,
-            "reflection": reflection,
-            "resolution_date": resolution_date,
-        })
-
-    if updates:
-        memory_log.batch_update_with_outcomes(updates)
-        result.settled.extend((ticker, u["trade_date"]) for u in updates)
+    with memory_log.settling(wait=wait) as held:
+        if not held:
+            return result
+        pending = [e for e in memory_log.get_pending_entries() if e["ticker"] == ticker]
+        if not pending:
+            return result
+        benchmark = resolve_benchmark(ticker, config)
+        for entry in pending:
+            try:
+                raw, alpha, days, resolution_date = fetch_returns(
+                    ticker, entry["date"], config.get("holding_period_days", 5), benchmark=benchmark,
+                )
+            except Exception as exc:
+                logger.warning("Prices for settling %s on %s were unavailable: %s", ticker, entry["date"], exc)
+                result.failed.append((ticker, entry["date"], f"prices unavailable: {exc}"))
+                continue
+            if raw is None:
+                continue  # the holding window has not traded yet: try again next run
+            try:
+                reflection = reflector.reflect_on_final_decision(
+                    final_decision=entry.get("decision", ""),
+                    raw_return=raw,
+                    alpha_return=alpha,
+                    benchmark_name=benchmark,
+                    holding_days=days,
+                )
+            except Exception as exc:
+                # Reflection calls a provider: a transient failure leaves the entry
+                # pending for the next pass rather than stopping the analysis.
+                logger.warning("Reflection failed for %s on %s: %s", ticker, entry["date"], exc)
+                result.failed.append((ticker, entry["date"], f"reflection failed: {exc}"))
+                continue
+            if memory_log.update_with_outcome(ticker, entry["date"], raw, alpha, days, reflection,
+                                              resolution_date=resolution_date):
+                result.settled.append((ticker, entry["date"]))
     return result
 
 
-def settle_all_pending(memory_log, reflector, config: dict) -> Settlement:
+def settle_all_pending(memory_log, reflector, config: dict, wait: bool = True) -> Settlement:
     """Settle every ticker's pending decisions whose holding window has traded (#1445).
 
     A run settles its own ticker; a ticker no longer analysed would otherwise
@@ -173,7 +163,7 @@ def settle_all_pending(memory_log, reflector, config: dict) -> Settlement:
     """
     result = Settlement()
     for ticker in dict.fromkeys(e["ticker"] for e in memory_log.get_pending_entries()):
-        done = settle_pending(ticker, memory_log, reflector, config)
+        done = settle_pending(ticker, memory_log, reflector, config, wait=wait)
         result.settled.extend(done.settled)
         result.failed.extend(done.failed)
     return result