From 756f7d3e34c6591607c80dcb75c26a6fcc45cd01 Mon Sep 17 00:00:00 2001 From: Ray Tien Date: Mon, 14 Sep 2026 15:13:19 +0800 Subject: [PATCH 1/2] fix(apodex): offload per-turn session persist off the event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _on_turn fires after every agent turn and called _persist() inline — a synchronous open()+json.dump() over the full history/display_history/ workflow_turns. As a session grows this write grows with it, and being awaited directly in run_agent_loop it stalls the event loop on every single turn (TUI freezes, no other coroutine gets to run). Move it to asyncio.to_thread, awaited so writes stay ordered turn to turn and the resume checkpoint can't be overwritten out of order. Benchmarked with a 2000-message history + a concurrent heartbeat coroutine: sync persist blocks the loop for ~139ms with 0 heartbeat ticks; to_thread lets ~4558 ticks through with a 0.85ms max stall. --- apodex/session.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/apodex/session.py b/apodex/session.py index d58fcf0..31a34ac 100644 --- a/apodex/session.py +++ b/apodex/session.py @@ -476,7 +476,10 @@ async def _on_turn(self, turn: int, messages: list, metadata: dict) -> None: after each completed turn — keep history current and persist.""" self.history = list(messages) self.display_history = list(messages) - self._persist() + # _persist() does synchronous file I/O over the full history; run it + # off the event loop so long sessions don't stall on every turn. + # Awaited (not fire-and-forget) so writes stay ordered turn-to-turn. + await asyncio.to_thread(self._persist) # ── persistence (interrupt-safe resume) ─────────────────────────────── def _enrich_task(self, task: str) -> str: From 9f9ca17a015858316cace81b861e75dd4df72883 Mon Sep 17 00:00:00 2001 From: Ray Tien Date: Mon, 21 Sep 2026 20:51:48 +0800 Subject: [PATCH 2/2] Fix concurrent checkpoint writes by locking and writing atomically _persist() now runs on both the main thread (start_new_session, rename_session) and a to_thread worker (_on_turn), so overlapping writers could interleave and corrupt session.json. Serialize writes with a threading.Lock and write via tmp file + os.replace so a torn write is never observable on disk. --- apodex/session.py | 79 ++++++++++++++++++++++++++++------------------- 1 file changed, 47 insertions(+), 32 deletions(-) diff --git a/apodex/session.py b/apodex/session.py index 31a34ac..a3d2384 100644 --- a/apodex/session.py +++ b/apodex/session.py @@ -11,6 +11,7 @@ import asyncio import json import os +import threading from pathlib import Path from typing import Any @@ -189,6 +190,10 @@ def __init__( # plugins.tools._path_auth._authorized_local_path). Without this they # only allow a few default dirs and deny the user's repo. self._authorize_workspace(cwd) + # _persist() now runs both on the main thread (start_new_session, + # rename_session) and off-thread (_on_turn's asyncio.to_thread), so + # concurrent writers must serialize on the same checkpoint file. + self._persist_lock = threading.Lock() @staticmethod def _active_spill_workspace() -> Path | None: @@ -594,7 +599,13 @@ def replay_history(self) -> list[Message]: def _persist(self) -> None: """Checkpoint session state so ``--resume `` can continue it. - Best-effort; a failed write never disrupts the session.""" + Best-effort; a failed write never disrupts the session. + + Serialized via ``_persist_lock`` and written atomically (tmp file + + ``os.replace``) because this runs from both the main thread + (``start_new_session`` / ``rename_session``) and a worker thread + (``_on_turn``'s ``asyncio.to_thread``) — without both, concurrent + writers can interleave and corrupt the checkpoint file.""" try: import json @@ -606,37 +617,41 @@ def _persist(self) -> None: self.tui_state = raw_tui_state if isinstance(raw_tui_state, dict) else {} path = _session_state_path(self.session_id) - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w", encoding="utf-8") as f: - json.dump({ - "session_id": self.session_id, - "created_at": self.created_at, - "local_timezone": self.local_timezone, - "name": self.session_name, - "mode": self.mode, - "cwd": self.cwd, - "model": self.cfg.model, - # Native messages are plain OpenAI-wire dicts — already - # JSON-serializable, so they round-trip verbatim (no - # langchain messages_to_dict / messages_from_dict needed). - "history": list(self.history), - "display_history": list(self.display_history), - "workflow_turns": list(self.workflow_turns), - "usage": self.usage.to_dict(), - "tui": dict(self.tui_state), - "outputs": { - "agent_root": os.environ.get("FRONTIER_AGENT_OUTPUTS_DIR", ""), - "host_root": os.environ.get("APODEX_HOST_OUTPUTS_DIR", ""), - }, - "journal": self.journal.to_dict(), - "journal_observed": self.journal.observed_paths(), - "journal_revert_base": self.journal.revert_bases(), - "plan_active": bool(self.plan_state.active), - "todos": [ - {"content": item.content, "status": item.status} - for item in get_todos() - ], - }, f, ensure_ascii=False) + payload = { + "session_id": self.session_id, + "created_at": self.created_at, + "local_timezone": self.local_timezone, + "name": self.session_name, + "mode": self.mode, + "cwd": self.cwd, + "model": self.cfg.model, + # Native messages are plain OpenAI-wire dicts — already + # JSON-serializable, so they round-trip verbatim (no + # langchain messages_to_dict / messages_from_dict needed). + "history": list(self.history), + "display_history": list(self.display_history), + "workflow_turns": list(self.workflow_turns), + "usage": self.usage.to_dict(), + "tui": dict(self.tui_state), + "outputs": { + "agent_root": os.environ.get("FRONTIER_AGENT_OUTPUTS_DIR", ""), + "host_root": os.environ.get("APODEX_HOST_OUTPUTS_DIR", ""), + }, + "journal": self.journal.to_dict(), + "journal_observed": self.journal.observed_paths(), + "journal_revert_base": self.journal.revert_bases(), + "plan_active": bool(self.plan_state.active), + "todos": [ + {"content": item.content, "status": item.status} + for item in get_todos() + ], + } + with self._persist_lock: + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp_path = f"{path}.{os.getpid()}.{threading.get_ident()}.tmp" + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False) + os.replace(tmp_path, path) except Exception: pass