Conversation
There was a problem hiding this comment.
Pull request overview
Fixes intermittent RuntimeError: Event loop is closed caused by running memory updates on short-lived asyncio.run() loops that interact poorly with langchain providers’ globally cached httpx.AsyncClient pools.
Changes:
- Introduce
_MemoryLoopRunner, which hosts a single persistent asyncio event loop in a daemon thread for memory updates. - Switch
_run_async_update_syncto submit coroutines onto the persistent loop viarun_coroutine_threadsafe. - Update unit tests to patch the new loop-runner failure mode (
_ensure_loopreturningNone) instead of executor submission failures.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
backend/packages/harness/deerflow/agents/memory/updater.py |
Replaces per-call asyncio.run() / executor bridging with a persistent daemon-thread event loop runner for memory updates. |
backend/tests/test_memory_updater.py |
Updates tests to align with the new loop-runner-based sync execution path and failure handling. |
| self._loop = asyncio.new_event_loop() | ||
| asyncio.set_event_loop(self._loop) | ||
| self._ready.set() | ||
| try: | ||
| self._loop.run_forever() | ||
| finally: | ||
| pending = asyncio.all_tasks(self._loop) | ||
| for task in pending: | ||
| task.cancel() | ||
| if pending: | ||
| self._loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) | ||
| self._loop.run_until_complete(self._loop.shutdown_asyncgens()) | ||
| self._loop.run_until_complete(self._loop.shutdown_default_executor()) | ||
| self._loop.close() | ||
|
|
||
| def _ensure_loop(self) -> asyncio.AbstractEventLoop | None: | ||
| """Return the persistent loop, starting it if needed.""" | ||
| if self._loop is not None and self._thread is not None and self._thread.is_alive(): | ||
| return self._loop | ||
|
|
||
| with self._lock: | ||
| if self._loop is not None and self._thread is not None and self._thread.is_alive(): | ||
| return self._loop | ||
|
|
||
| self._ready.clear() |
There was a problem hiding this comment.
_ensure_loop() can briefly return a stale/closed loop if the runner is restarted: when a new thread is created, self._thread becomes alive before _run() overwrites self._loop, so the fast-path check may return the previous loop value. Consider clearing self._loop (and/or checking not self._loop.is_closed()) before starting the new thread, so callers can’t get an old loop during startup/restart races.
| self._loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(self._loop) | |
| self._ready.set() | |
| try: | |
| self._loop.run_forever() | |
| finally: | |
| pending = asyncio.all_tasks(self._loop) | |
| for task in pending: | |
| task.cancel() | |
| if pending: | |
| self._loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) | |
| self._loop.run_until_complete(self._loop.shutdown_asyncgens()) | |
| self._loop.run_until_complete(self._loop.shutdown_default_executor()) | |
| self._loop.close() | |
| def _ensure_loop(self) -> asyncio.AbstractEventLoop | None: | |
| """Return the persistent loop, starting it if needed.""" | |
| if self._loop is not None and self._thread is not None and self._thread.is_alive(): | |
| return self._loop | |
| with self._lock: | |
| if self._loop is not None and self._thread is not None and self._thread.is_alive(): | |
| return self._loop | |
| self._ready.clear() | |
| loop = asyncio.new_event_loop() | |
| self._loop = loop | |
| asyncio.set_event_loop(loop) | |
| self._ready.set() | |
| try: | |
| loop.run_forever() | |
| finally: | |
| pending = asyncio.all_tasks(loop) | |
| for task in pending: | |
| task.cancel() | |
| if pending: | |
| loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) | |
| loop.run_until_complete(loop.shutdown_asyncgens()) | |
| loop.run_until_complete(loop.shutdown_default_executor()) | |
| loop.close() | |
| with self._lock: | |
| if self._loop is loop: | |
| self._loop = None | |
| if self._thread is threading.current_thread(): | |
| self._thread = None | |
| def _ensure_loop(self) -> asyncio.AbstractEventLoop | None: | |
| """Return the persistent loop, starting it if needed.""" | |
| loop = self._loop | |
| thread = self._thread | |
| if loop is not None and not loop.is_closed() and thread is not None and thread.is_alive(): | |
| return loop | |
| with self._lock: | |
| loop = self._loop | |
| thread = self._thread | |
| if loop is not None and not loop.is_closed() and thread is not None and thread.is_alive(): | |
| return loop | |
| self._ready.clear() | |
| self._loop = None |
| future = asyncio.run_coroutine_threadsafe(coro, loop) | ||
| try: | ||
| return future.result(timeout=300) | ||
| except Exception: |
There was a problem hiding this comment.
_MemoryLoopRunner.run() calls asyncio.run_coroutine_threadsafe() outside the try/except. If scheduling fails (e.g., loop closed/stopping, wrong awaitable type), the exception will propagate and can crash callers; previously _run_async_update_sync always caught exceptions and returned False. Wrap the submission in the try/except and, on submission failure, close the coroutine (if possible) to avoid 'coroutine was never awaited' warnings.
| future = asyncio.run_coroutine_threadsafe(coro, loop) | |
| try: | |
| return future.result(timeout=300) | |
| except Exception: | |
| try: | |
| future = asyncio.run_coroutine_threadsafe(coro, loop) | |
| return future.result(timeout=300) | |
| except Exception: | |
| close = getattr(coro, "close", None) | |
| if callable(close): | |
| try: | |
| close() | |
| except Exception: | |
| pass |
| return future.result(timeout=300) | ||
| except Exception: | ||
| logger.exception("Memory update coroutine failed in persistent loop") | ||
| return False |
There was a problem hiding this comment.
The 300s timeout on future.result() changes behavior from the prior implementation (which waited indefinitely). On timeout, the coroutine will typically keep running in the background, but this method returns False, which can lead to confusing/incorrect state. Consider handling TimeoutError explicitly by cancelling the scheduled task (and logging a timeout-specific message), and/or making the timeout configurable if you need a bound here.
| future = asyncio.run_coroutine_threadsafe(coro, loop) | ||
| try: | ||
| return future.result(timeout=300) | ||
| except Exception: | ||
| logger.exception("Memory update coroutine failed in persistent loop") | ||
| return False |
There was a problem hiding this comment.
New loop-runner behavior isn’t directly covered by tests (e.g., timeout handling/cancellation, and submission errors from run_coroutine_threadsafe being handled without raising or leaking un-awaited coroutine warnings). Adding focused unit tests around _MemoryLoopRunner.run() would help prevent regressions in this cross-thread scheduling path.
|
@copilot apply changes based on the comments in this thread |
Applied all changes from the review thread in commits ca0835a and 27ddca2:
Warning Firewall rules blocked me from connecting to one or more addresses (expand for details)I tried to connect to the following addresses, but was blocked by firewall rules:
If you need me to access, download, or install something from one of these locations, you can either:
|
rayhpeng
left a comment
There was a problem hiding this comment.
Thanks for working on this. I think this needs another pass before merge.
The persistent loop reduces one symptom, but it does not fully address the root cause in #2615. The problem is not only that the memory updater closes its event loop; it is that langchain providers can keep a process-global cached AsyncClient / connection pool, and connections created by the memory updater are still created on a different event loop from the lead agent. Keeping the memory loop alive turns the failure mode from “closed loop” into “foreign live loop”, but the shared pool can still hand a memory-loop transport back to the main agent loop.
I think the safer fix is to isolate the memory updater from the main provider client pool, disable/restrict keepalive for the memory updater client path, or otherwise guarantee that connections created on the memory loop cannot be reused by the lead agent. It would also be good to add a regression test that exercises: memory update performs an async provider call, then the lead agent performs another async provider call, and the provider/client cache does not reuse a connection across loops.
There is also a smaller behavior change around _UPDATE_TIMEOUT. The previous implementation waited for the update to finish. With the new timeout path, future.cancel() is best-effort and returns immediately; if the coroutine is already inside provider I/O or asyncio.to_thread(_finalize_update), it may still complete and mutate memory storage after run() has returned False. The queue can then continue to the next update, causing overlapping writes that did not exist before. I would either remove this timeout, wait until cancellation has actually settled before returning, or make the timeout configurable and add a test proving a timed-out update cannot later write stale data.
rayhpeng
left a comment
There was a problem hiding this comment.
Re-reviewed the latest head. The persistent side-loop and timeout issues from the earlier implementation look addressed by the switch to the sync model.invoke() path, but I still see two issues that should be fixed before merge.
[P1] user_id is still dropped by memory updates
MemoryUpdateQueue captures context.user_id and passes it into update_memory(..., user_id=context.user_id), but this rewritten sync path never forwards user_id into _do_update_memory_sync, _prepare_update_prompt, or _finalize_update. Those methods still call get_memory_data(agent_name) and save(updated_memory, agent_name) without the user scope, so queued per-user memory updates can read/write the unscoped memory store. Since this PR rewrites update_memory, please thread user_id through the new sync path and add a regression test that verifies load/save receive the provided user_id.
[P2] async memory API still uses the unsafe provider path
The production update_memory() path now avoids the async provider client pool, but aupdate_memory() remains public and still calls model.ainvoke(). That preserves an entry point that can recreate the exact cross-loop async-client problem from #2615 if any caller uses MemoryUpdater().aupdate_memory(...) directly or if future queue code switches back to it. Consider making aupdate_memory() delegate to the sync implementation via asyncio.to_thread(...), or clearly removing/internalizing this async path so all memory updater entry points use the isolated sync client path.
Nit: backend/packages/harness/deerflow/agents/memory/updater.py has a duplicated # Matches sentences... comment around the upload regex; it can be removed while touching this file.
…loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops.
P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230.
rayhpeng
left a comment
There was a problem hiding this comment.
Re-reviewed the latest head. The previous blocking issues are addressed in this version: the persistent side loop and timeout are gone, user_id is now threaded through load/save, and aupdate_memory() delegates to the sync model.invoke() path.
One non-blocking test coverage issue remains:
[P3] Cache-isolation test no longer exercises finalize/save path
TestFinalizeCacheIsolation.test_deepcopy_prevents_cache_corruption_on_save_failure still builds mock_model = AsyncMock() and only configures mock_model.ainvoke, but update_memory() now calls the sync model.invoke() path. As written, the call can fail before _finalize_update() / save_mock are exercised, while the final assertion still passes because original_memory was never touched.
Please set mock_model.invoke = MagicMock(return_value=mock_response) and assert save_mock / saved_objects were actually hit, so the deepcopy-on-save-failure regression remains covered after the sync-path rewrite.
|
fixed #2647 |
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
…loop (bytedance#2627) * fix(memory): replace short-lived asyncio.run() with persistent event loop to prevent zombie httpx connections The memory updater used asyncio.run() inside daemon threads, creating and destroying short-lived event loops on every update. Langchain providers (e.g. langchain-anthropic) cache httpx AsyncClient instances globally via @lru_cache, so SSL connections created on a loop that is subsequently destroyed become zombie connections in the shared pool. When the main agent's lead run later reuses one of these connections, httpx/anyio triggers RuntimeError: Event loop is closed during connection cleanup. Replace the ThreadPoolExecutor + asyncio.run() pattern with a _MemoryLoopRunner that maintains a single persistent event loop in a daemon thread for the process lifetime. Since the loop never closes, connections bound to it never become invalid. The _run_async_update_sync function now submits coroutines to this persistent loop via run_coroutine_threadsafe instead of creating throwaway loops. * update the code to address the review comments * Fix the review comments of 2615 P1 — user_id forwarded through sync path: Added user_id parameter to _prepare_update_prompt, _finalize_update, and _do_update_memory_sync, and forwarded it to get_memory_data(agent_name, user_id=user_id) and save(..., user_id=user_id). The update_memory() entry point now passes user_id through both the executor.submit path and the direct call path. Added TestUserIdForwarding with two regression tests (sync + async) verifying get_memory_data and save receive the correct user_id. P2 — aupdate_memory() delegates to sync: Replaced the model.ainvoke() call with asyncio.to_thread(self._do_update_memory_sync, ...). This eliminates the unsafe async provider client path entirely — all memory updater entry points now use the isolated sync model.invoke() path. Updated the test from asserting ainvoke is awaited to asserting invoke is called and ainvoke is not. Nit — duplicate comment removed: Removed the duplicated # Matches sentences... comment on line 230. * Chore(test): update the code of test_memory_updater --------- Co-authored-by: rayhpeng <rayhpeng@gmail.com>
to prevent zombie httpx connections
The memory updater used asyncio.run() inside daemon threads, creating
and destroying short-lived event loops on every update. Langchain
providers (e.g. langchain-anthropic) cache httpx AsyncClient instances
globally via @lru_cache, so SSL connections created on a loop that is
subsequently destroyed become zombie connections in the shared pool.
When the main agent's lead run later reuses one of these connections,
httpx/anyio triggers RuntimeError: Event loop is closed during
connection cleanup.
Replace the ThreadPoolExecutor + asyncio.run() pattern with a
_MemoryLoopRunner that maintains a single persistent event loop in a
daemon thread for the process lifetime. Since the loop never closes,
connections bound to it never become invalid. The _run_async_update_sync
function now submits coroutines to this persistent loop via
run_coroutine_threadsafe instead of creating throwaway loops.
Fixes #2615