Fix/misc bug fixes

cron/scheduler.py

@@ -158,7 +158,9 @@ def _deliver_result(job: dict, content: str) -> None:
     # Run the async send in a fresh event loop (safe from any thread)
     try:
         result = asyncio.run(_send_to_platform(platform, pconfig, chat_id, content, thread_id=thread_id))
-    except RuntimeError:
+    except RuntimeError as e:
+        if "event loop" not in str(e).lower():
+            raise
         # asyncio.run() fails if there's already a running loop in this thread;
         # spin up a new thread to avoid that.
         import concurrent.futures
@@ -524,14 +526,15 @@ def tick(verbose: bool = True) -> int:
 
         return executed
     finally:
-        if fcntl:
-            fcntl.flock(lock_fd, fcntl.LOCK_UN)
-        elif msvcrt:
-            try:
-                msvcrt.locking(lock_fd.fileno(), msvcrt.LK_UNLCK, 1)
-            except (OSError, IOError):
-                pass
-        lock_fd.close()
+        if lock_fd is not None:
+            if fcntl:
+                fcntl.flock(lock_fd, fcntl.LOCK_UN)
+            elif msvcrt:
+                try:
+                    msvcrt.locking(lock_fd.fileno(), msvcrt.LK_UNLCK, 1)
+                except (OSError, IOError):
+                    pass
+            lock_fd.close()
 
 
 if __name__ == "__main__":

gateway/session.py

@@ -890,9 +890,12 @@ def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db
                 logger.debug("Session DB operation failed: %s", e)
 
         # Also write legacy JSONL (keeps existing tooling working during transition)
-        transcript_path = self.get_transcript_path(session_id)
-        with open(transcript_path, "a", encoding="utf-8") as f:
-            f.write(json.dumps(message, ensure_ascii=False) + "\n")
+        try:
+            transcript_path = self.get_transcript_path(session_id)
+            with open(transcript_path, "a", encoding="utf-8") as f:
+                f.write(json.dumps(message, ensure_ascii=False) + "\n")
+        except OSError as e:
+            logger.debug("Failed to write JSONL transcript for session %s: %s", session_id, e)
 
     def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> None:
         """Replace the entire transcript for a session with new messages.

gateway/status.py

@@ -159,7 +159,10 @@ def _read_pid_record() -> Optional[dict]:
     if not pid_path.exists():
         return None
 
-    raw = pid_path.read_text().strip()
+    try:
+        raw = pid_path.read_text().strip()
+    except OSError:
+        return None
     if not raw:
         return None
 

hermes_state.py

@@ -148,13 +148,15 @@ def _init_schema(self):
                 except sqlite3.OperationalError:
                     pass  # Column already exists
                 cursor.execute("UPDATE schema_version SET version = 2")
+                self._conn.commit()
             if current_version < 3:
                 # v3: add title column to sessions
                 try:
                     cursor.execute("ALTER TABLE sessions ADD COLUMN title TEXT")
                 except sqlite3.OperationalError:
                     pass  # Column already exists
                 cursor.execute("UPDATE schema_version SET version = 3")
+                self._conn.commit()
             if current_version < 4:
                 # v4: add unique index on title (NULLs allowed, only non-NULL must be unique)
                 try:
@@ -165,6 +167,7 @@ def _init_schema(self):
                 except sqlite3.OperationalError:
                     pass  # Index already exists
                 cursor.execute("UPDATE schema_version SET version = 4")
+                self._conn.commit()
             if current_version < 5:
                 new_columns = [
                     ("cache_read_tokens", "INTEGER DEFAULT 0"),
@@ -185,6 +188,7 @@ def _init_schema(self):
                     except sqlite3.OperationalError:
                         pass
                 cursor.execute("UPDATE schema_version SET version = 5")
+                self._conn.commit()
 
         # Unique title index — always ensure it exists (safe to run after migrations
         # since the title column is guaranteed to exist at this point)
@@ -851,23 +855,25 @@ def search_sessions(
 
     def session_count(self, source: str = None) -> int:
         """Count sessions, optionally filtered by source."""
-        if source:
-            cursor = self._conn.execute(
-                "SELECT COUNT(*) FROM sessions WHERE source = ?", (source,)
-            )
-        else:
-            cursor = self._conn.execute("SELECT COUNT(*) FROM sessions")
-        return cursor.fetchone()[0]
+        with self._lock:
+            if source:
+                cursor = self._conn.execute(
+                    "SELECT COUNT(*) FROM sessions WHERE source = ?", (source,)
+                )
+            else:
+                cursor = self._conn.execute("SELECT COUNT(*) FROM sessions")
+            return cursor.fetchone()[0]
 
     def message_count(self, session_id: str = None) -> int:
         """Count messages, optionally for a specific session."""
-        if session_id:
-            cursor = self._conn.execute(
-                "SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,)
-            )
-        else:
-            cursor = self._conn.execute("SELECT COUNT(*) FROM messages")
-        return cursor.fetchone()[0]
+        with self._lock:
+            if session_id:
+                cursor = self._conn.execute(
+                    "SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,)
+                )
+            else:
+                cursor = self._conn.execute("SELECT COUNT(*) FROM messages")
+            return cursor.fetchone()[0]
 
     # =========================================================================
     # Export and cleanup

run_agent.py

@@ -287,7 +287,10 @@ def _should_parallelize_tool_batch(tool_calls) -> bool:
         if tool_name in _PATH_SCOPED_TOOLS:
             scoped_path = _extract_parallel_scope_path(tool_name, function_args)
             if scoped_path is None:
-                return False
+                # Path couldn't be extracted; fall back to parallel-safe check
+                if tool_name not in _PARALLEL_SAFE_TOOLS:
+                    return False
+                continue
             if any(_paths_overlap(scoped_path, existing) for existing in reserved_paths):
                 return False
             reserved_paths.append(scoped_path)
@@ -308,8 +311,11 @@ def _extract_parallel_scope_path(tool_name: str, function_args: dict) -> Path |
     if not isinstance(raw_path, str) or not raw_path.strip():
         return None
 
-    # Avoid resolve(); the file may not exist yet.
-    return Path(raw_path).expanduser()
+    # Avoid resolve(); the file may not exist yet. But do collapse lexical
+    # aliases like "./" and "../" so same-target writes do not get
+    # misclassified as independent and run concurrently.
+    normalized = os.path.normpath(os.path.expanduser(raw_path.strip()))
+    return Path(normalized)
 
 
 def _paths_overlap(left: Path, right: Path) -> bool:
@@ -1715,7 +1721,6 @@ def _hydrate_todo_store(self, history: List[Dict[str, Any]]) -> None:
             self._todo_store.write(last_todo_response, merge=False)
             if not self.quiet_mode:
                 self._vprint(f"{self.log_prefix}📋 Restored {len(last_todo_response)} todo item(s) from history")
-        _set_interrupt(False)
 
     @property
     def is_interrupted(self) -> bool:
@@ -3552,8 +3557,16 @@ def _materialize_data_url_for_vision(image_url: str) -> tuple[str, Optional[Path
             "image/jpg": ".jpg",
         }.get(mime, ".jpg")
         tmp = tempfile.NamedTemporaryFile(prefix="anthropic_image_", suffix=suffix, delete=False)
-        with tmp:
-            tmp.write(base64.b64decode(data))
+        try:
+            with tmp:
+                tmp.write(base64.b64decode(data))
+        except Exception:
+            # Clean up the temp file on decode/write failure
+            try:
+                os.unlink(tmp.name)
+            except OSError:
+                pass
+            raise
         path = Path(tmp.name)
         return str(path), path
 

tests/test_run_agent.py

@@ -1086,6 +1086,27 @@ def test_one_empty_path_does_not_overlap(self):
         assert not _paths_overlap(Path("src/a.py"), Path(""))
 
 
+class TestParallelScopeNormalization:
+    def test_extract_parallel_scope_path_normalizes_dotdot_segments(self):
+        from run_agent import _extract_parallel_scope_path
+
+        assert _extract_parallel_scope_path("write_file", {"path": "subdir/../notes.txt"}) == Path("notes.txt")
+
+    def test_same_effective_path_disables_parallel_batch(self):
+        from run_agent import _should_parallelize_tool_batch
+
+        tool_calls = [
+            _mock_tool_call("write_file", json.dumps({"path": "notes.txt", "content": "a"})),
+            _mock_tool_call("patch", json.dumps({
+                "path": "subdir/../notes.txt",
+                "old_string": "a",
+                "new_string": "b",
+            })),
+        ]
+
+        assert _should_parallelize_tool_batch(tool_calls) is False
+
+
 class TestHandleMaxIterations:
     def test_returns_summary(self, agent):
         resp = _mock_response(content="Here is a summary of what I did.")

tools/memory_tool.py

@@ -23,7 +23,10 @@
 - Frozen snapshot pattern: system prompt is stable, tool responses show live state
 """
 
-import fcntl
+try:
+    import fcntl
+except ImportError:
+    fcntl = None  # Not available on Windows
 import json
 import logging
 import os
@@ -130,6 +133,10 @@ def _file_lock(path: Path):
         Uses a separate .lock file so the memory file itself can still be
         atomically replaced via os.replace().
         """
+        if fcntl is None:
+            # No file locking on Windows — yield without lock
+            yield
+            return
         lock_path = path.with_suffix(path.suffix + ".lock")
         lock_path.parent.mkdir(parents=True, exist_ok=True)
         fd = open(lock_path, "w")

tools/web_tools.py

@@ -47,7 +47,6 @@
 import asyncio
 from typing import List, Dict, Any, Optional
 import httpx
-from firecrawl import Firecrawl
 from agent.auxiliary_client import async_call_llm
 from tools.debug_helpers import DebugSession
 from tools.website_policy import check_website_access
@@ -115,6 +114,7 @@ def _get_firecrawl_client():
             kwargs["api_key"] = api_key
         if api_url:
             kwargs["api_url"] = api_url
+        from firecrawl import Firecrawl
         _firecrawl_client = Firecrawl(**kwargs)
     return _firecrawl_client