Reduce session index global lock window

CHANGELOG.md

@@ -3,6 +3,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- **Streaming-time session index updates now hold the global session lock for less work.** `_write_session_index()` now snapshots in-memory session rows under `LOCK`, but reads/parses the existing `_index.json`, sorts, serializes, and writes the updated index outside that lock. This reduces avoidable `/api/session` and `/api/sessions` contention while a running stream saves session metadata, addressing the lock-window half of #4918 without changing the active-stream lifecycle.
+
 ## [v0.51.653] — 2026-06-25 — Release XI (gateway approval failures stay actionable instead of dead-ending)
 
 ### Fixed

api/models.py

@@ -260,8 +260,9 @@ def _write_session_index(updates=None, *, session_dir: Path | None = None, sessi
     the existing index — O(1) for single-session changes.  When *updates*
     is None, a full rebuild is performed (used on startup / first call).
 
-    LOCK protects in-memory state snapshots and payload construction only;
-    disk I/O (write/flush/fsync/replace) always runs outside LOCK.
+    LOCK protects only in-memory session snapshots.  JSON parsing, payload
+    construction, and disk I/O run outside LOCK so active-stream saves do not
+    block ordinary session reads longer than necessary.
     """
     session_dir = session_dir or SESSION_DIR
     session_index_file = session_index_file or SESSION_INDEX_FILE
@@ -293,13 +294,16 @@ def _write_session_index(updates=None, *, session_dir: Path | None = None, sessi
                     logger.debug("Failed to load session from %s", p)
             entries = list(entry_map.values())
 
+            existing_ids = set(entry_map.keys())
             with LOCK:
-                existing_ids = set(entry_map.keys())
-                for s in SESSIONS.values():
-                    if s.session_id not in existing_ids:
-                        entries.append(s.compact())
-                entries.sort(key=lambda s: s.get('updated_at', 0), reverse=True)
-                _payload = json.dumps(entries, ensure_ascii=False, indent=2)
+                in_memory_entries = [
+                    s.compact()
+                    for s in SESSIONS.values()
+                    if s.session_id not in existing_ids
+                ]
+            entries.extend(in_memory_entries)
+            entries.sort(key=lambda s: s.get('updated_at', 0), reverse=True)
+            _payload = json.dumps(entries, ensure_ascii=False, indent=2)
 
             try:
                 with open(_tmp, 'w', encoding='utf-8') as f:
@@ -323,29 +327,30 @@ def _write_session_index(updates=None, *, session_dir: Path | None = None, sessi
             # Avoid N filesystem exists() checks under LOCK by collecting
             # on-disk IDs once before entering the critical section.
             on_disk_ids = _persisted_session_ids_snapshot()
+            existing = json.loads(session_index_file.read_text(encoding='utf-8'))
+            if not isinstance(existing, list):
+                raise ValueError("session index must be a list")
             with LOCK:
-                existing = json.loads(session_index_file.read_text(encoding='utf-8'))
                 in_memory_ids = set(SESSIONS.keys())
-
-                existing = [
-                    e for e in existing
-                    if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids)
-                ]
-
-                # Build lookup of updated entries
                 updated_map = {s.session_id: s.compact() for s in updates}
-                existing_ids = {e.get('session_id') for e in existing}
-                # Add any updated entries not yet in the index
-                for sid, entry in updated_map.items():
-                    if sid not in existing_ids:
-                        existing.append(entry)
-                # Replace matching entries in-place
-                for i, e in enumerate(existing):
-                    sid = e.get('session_id')
-                    if sid in updated_map:
-                        existing[i] = updated_map[sid]
-                existing.sort(key=lambda s: s.get('updated_at', 0), reverse=True)
-                _payload = json.dumps(existing, ensure_ascii=False, indent=2)
+
+            existing = [
+                e for e in existing
+                if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids)
+            ]
+
+            existing_ids = {e.get('session_id') for e in existing}
+            # Add any updated entries not yet in the index.
+            for sid, entry in updated_map.items():
+                if sid not in existing_ids:
+                    existing.append(entry)
+            # Replace matching entries in-place.
+            for i, e in enumerate(existing):
+                sid = e.get('session_id')
+                if sid in updated_map:
+                    existing[i] = updated_map[sid]
+            existing.sort(key=lambda s: s.get('updated_at', 0), reverse=True)
+            _payload = json.dumps(existing, ensure_ascii=False, indent=2)
 
             try:
                 with open(_tmp, 'w', encoding='utf-8') as f:

tests/test_session_index_lock_window.py

@@ -0,0 +1,71 @@
+import collections
+import json
+import threading
+
+
+class RecordingLock:
+    def __init__(self):
+        self._lock = threading.RLock()
+        self.depth = 0
+
+    def __enter__(self):
+        self._lock.acquire()
+        self.depth += 1
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.depth -= 1
+        self._lock.release()
+
+    @property
+    def held(self):
+        return self.depth > 0
+
+
+def test_session_index_fast_path_keeps_json_work_outside_global_lock(monkeypatch, tmp_path):
+    import api.models as models
+
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    index_file = session_dir / "_index.json"
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
+
+    old = models.Session(session_id="idx_old", title="Old", updated_at=1.0)
+    updated = models.Session(session_id="idx_updated", title="Updated", updated_at=20.0)
+    (session_dir / "idx_old.json").write_text("{}", encoding="utf-8")
+    (session_dir / "idx_updated.json").write_text("{}", encoding="utf-8")
+    index_file.write_text(
+        json.dumps(
+            [
+                old.compact(),
+                models.Session(session_id="idx_updated", title="Stale", updated_at=2.0).compact(),
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    lock = RecordingLock()
+    monkeypatch.setattr(models, "LOCK", lock)
+    monkeypatch.setattr(models, "SESSIONS", collections.OrderedDict())
+
+    original_loads = models.json.loads
+    original_dumps = models.json.dumps
+
+    def loads_outside_lock(*args, **kwargs):
+        assert not lock.held
+        return original_loads(*args, **kwargs)
+
+    def dumps_outside_lock(*args, **kwargs):
+        assert not lock.held
+        return original_dumps(*args, **kwargs)
+
+    monkeypatch.setattr(models.json, "loads", loads_outside_lock)
+    monkeypatch.setattr(models.json, "dumps", dumps_outside_lock)
+
+    models._write_session_index(updates=[updated])
+
+    rows = json.loads(index_file.read_text(encoding="utf-8"))
+    assert [row["session_id"] for row in rows] == ["idx_updated", "idx_old"]
+    assert rows[0]["title"] == "Updated"
+    assert rows[0]["updated_at"] == 20.0