fix(sidebar): keep manual forks out of compaction snapshot lineage swaps (#3799)

api/models.py

@@ -59,6 +59,7 @@
 _INDEX_WRITE_LOCK = threading.RLock()
 _SESSION_INDEX_REBUILD_LOCK = threading.Lock()
 _SESSION_INDEX_REBUILD_THREAD = None
+_SESSION_INDEX_REBUILD_THREAD_TARGET: tuple[Path, Path] | None = None
 
 # Path-safety contract for session IDs.  Accept alphanumerics, underscore, and
 # hyphen so API/gateway-issued ids (``api-*``, ``reachy-voice-*``) round-trip
@@ -146,26 +147,46 @@ def _session_dir_has_persisted_session_files() -> bool:
         return False
 
 
-def _rebuild_session_index_background() -> None:
+def _rebuild_session_index_background(expected_session_dir: Path, expected_index_file: Path) -> None:
+    global _SESSION_INDEX_REBUILD_THREAD, _SESSION_INDEX_REBUILD_THREAD_TARGET
     try:
-        _write_session_index(updates=None)
+        with _SESSION_INDEX_REBUILD_LOCK:
+            if SESSION_DIR != expected_session_dir or SESSION_INDEX_FILE != expected_index_file:
+                return
+        _write_session_index(
+            updates=None,
+            session_dir=expected_session_dir,
+            session_index_file=expected_index_file,
+        )
     except Exception:
         logger.debug("Background session-index rebuild failed", exc_info=True)
+    finally:
+        with _SESSION_INDEX_REBUILD_LOCK:
+            if _SESSION_INDEX_REBUILD_THREAD_TARGET == (
+                expected_session_dir,
+                expected_index_file,
+            ):
+                _SESSION_INDEX_REBUILD_THREAD = None
+                _SESSION_INDEX_REBUILD_THREAD_TARGET = None
 
 
 def _start_session_index_rebuild_thread() -> None:
     """Start one background full-index rebuild if the index is missing."""
-    global _SESSION_INDEX_REBUILD_THREAD
+    global _SESSION_INDEX_REBUILD_THREAD, _SESSION_INDEX_REBUILD_THREAD_TARGET
+    target = (SESSION_DIR, SESSION_INDEX_FILE)
     with _SESSION_INDEX_REBUILD_LOCK:
         if SESSION_INDEX_FILE.exists():
             return
         if (
             _SESSION_INDEX_REBUILD_THREAD is not None
             and _SESSION_INDEX_REBUILD_THREAD.is_alive()
+            and _SESSION_INDEX_REBUILD_THREAD_TARGET == target
         ):
             return
+        _SESSION_INDEX_REBUILD_THREAD_TARGET = target
         _SESSION_INDEX_REBUILD_THREAD = threading.Thread(
             target=_rebuild_session_index_background,
+            args=target,
             name="session-index-rebuild",
             daemon=True,
         )
@@ -191,7 +212,7 @@ def _index_entry_exists(session_id: str, in_memory_ids=None) -> bool:
     return p.exists()
 
 
-def _write_session_index(updates=None):
+def _write_session_index(updates=None, *, session_dir: Path | None = None, session_index_file: Path | None = None):
     """Update the session index file.
 
     When *updates* is provided (a list of Session objects whose compact
@@ -202,18 +223,20 @@ def _write_session_index(updates=None):
     LOCK protects in-memory state snapshots and payload construction only;
     disk I/O (write/flush/fsync/replace) always runs outside LOCK.
     """
-    _tmp = SESSION_INDEX_FILE.with_suffix(f'.tmp.{os.getpid()}.{threading.current_thread().ident}')
+    session_dir = session_dir or SESSION_DIR
+    session_index_file = session_index_file or SESSION_INDEX_FILE
+    _tmp = session_index_file.with_suffix(f'.tmp.{os.getpid()}.{threading.current_thread().ident}')
 
     with _INDEX_WRITE_LOCK:
         # Lazy full-rebuild path — used when index doesn't exist yet.
-        if updates is None or not SESSION_INDEX_FILE.exists():
+        if updates is None or not session_index_file.exists():
             _cleanup_stale_tmp_files()  # best-effort sweep on startup / first call
             entry_map: dict[str, dict] = {}
-            for p in SESSION_DIR.glob('*.json'):
+            for p in session_dir.glob('*.json'):
                 if p.name.startswith('_'):
                     continue
                 try:
-                    s = Session.load(p.stem)
+                    s = _load_session_from_path(p)
                     if s:
                         c = s.compact()
                         sid = c.get('session_id')
@@ -243,7 +266,7 @@ def _write_session_index(updates=None):
                     f.write(_payload)
                     f.flush()
                     os.fsync(f.fileno())
-                os.replace(_tmp, SESSION_INDEX_FILE)
+                os.replace(_tmp, session_index_file)
             except Exception:
                 # Best-effort cleanup of stale tmp on failure
                 try:
@@ -261,7 +284,7 @@ def _write_session_index(updates=None):
             # on-disk IDs once before entering the critical section.
             on_disk_ids = _persisted_session_ids_snapshot()
             with LOCK:
-                existing = json.loads(SESSION_INDEX_FILE.read_text(encoding='utf-8'))
+                existing = json.loads(session_index_file.read_text(encoding='utf-8'))
                 in_memory_ids = set(SESSIONS.keys())
 
                 existing = [
@@ -289,7 +312,7 @@ def _write_session_index(updates=None):
                     f.write(_payload)
                     f.flush()
                     os.fsync(f.fileno())
-                os.replace(_tmp, SESSION_INDEX_FILE)
+                os.replace(_tmp, session_index_file)
             except Exception:
                 try:
                     _tmp.unlink(missing_ok=True)
@@ -511,6 +534,16 @@ def _read_metadata_json_prefix(path, max_prefix_bytes=65536):
     return None
 
 
+def _load_session_from_path(path: Path) -> "Session | None":
+    """Load a session from an explicit JSON path without consulting SESSION_DIR."""
+    try:
+        data = json.loads(path.read_text(encoding='utf-8'))
+    except Exception:
+        return None
+    data['messages'], _collapsed_partials = _collapse_adjacent_duplicate_partials(data.get('messages'))
+    return Session(**data)
+
+
 def _lookup_index_message_count(session_id):
     """Return the indexed message count without loading the full session file."""
     return _index_message_count_map().get(str(session_id))
@@ -2452,9 +2485,18 @@ def _sidebar_message_count(session: dict) -> int:
 
 def _sidebar_lineage_root_id(session: dict, sessions_by_id: dict[str, dict]) -> str:
     sid = str(session.get('session_id') or '')
+    explicit = str(session.get('_lineage_root_id') or '').strip()
+    if explicit:
+        return explicit
+    relationship_type = str(session.get('relationship_type') or '').strip().lower()
+    if relationship_type == 'child_session':
+        return sid
     root = sid
     parent = session.get('parent_session_id')
+    source = str(session.get('session_source') or '').strip().lower()
     seen = {sid}
+    if source == 'fork':
+        return root
     while parent and parent not in seen and parent in sessions_by_id:
         root = str(parent)
         seen.add(root)
@@ -3108,6 +3150,8 @@ def all_sessions(diag=None):
                 and not s.get('has_pending_user_message')
                 and not s.get('worktree_path')
             )]
+            _diag_stage(diag, "all_sessions.lineage_metadata")
+            _enrich_sidebar_lineage_metadata(result)
             result = _prefer_fuller_snapshots_for_sidebar(result)
             sidebar_candidates = result
             visible_result = [s for s in sidebar_candidates if not _hide_from_default_sidebar(s)]
@@ -3119,8 +3163,6 @@ def all_sessions(diag=None):
             for s in result:
                 if not s.get('profile'):
                     s['profile'] = 'default'
-            _diag_stage(diag, "all_sessions.lineage_metadata")
-            _enrich_sidebar_lineage_metadata(result)
             return result
         except Exception:
             logger.debug("Failed to load session index, falling back to full scan")
@@ -3149,6 +3191,8 @@ def all_sessions(diag=None):
         and not s.pending_user_message
         and not getattr(s, 'worktree_path', None)
     )]
+    _diag_stage(diag, "all_sessions.lineage_metadata")
+    _enrich_sidebar_lineage_metadata(result)
     result = _prefer_fuller_snapshots_for_sidebar(result)
     sidebar_candidates = result
     visible_result = [s for s in sidebar_candidates if not _hide_from_default_sidebar(s)]
@@ -3158,8 +3202,6 @@ def all_sessions(diag=None):
     for s in result:
         if not s.get('profile'):
             s['profile'] = 'default'
-    _diag_stage(diag, "all_sessions.lineage_metadata")
-    _enrich_sidebar_lineage_metadata(result)
     return result
 
 

tests/test_session_index.py

@@ -437,6 +437,76 @@ def test_pre_compression_snapshot_hidden_from_active_sidebar_but_file_remains(mo
     assert [row["session_id"] for row in rows] == ["new_sid"]
 
 
+def test_forked_child_of_snapshot_stays_visible_when_snapshot_is_fuller(monkeypatch):
+    """A manual fork should not be grouped into a snapshot's hidden continuation lineage.
+
+    Even when the parent snapshot has a fuller transcript and a newer
+    timestamp, a `/branch` fork is independently discoverable and should stay in
+    the active sidebar rows as its own root.
+    """
+    snapshot = Session(
+        session_id="snapshot_parent",
+        title="Long Conversation",
+        messages=[
+            {"role": "user", "content": "root"},
+            {"role": "assistant", "content": "compressed context"},
+            {"role": "user", "content": "old question"},
+            {"role": "assistant", "content": "old answer"},
+        ],
+        pre_compression_snapshot=True,
+        parent_session_id="snapshot_origin",
+        updated_at=300.0,
+        last_message_at=300.0,
+    )
+    fork = Session(
+        session_id="manual_fork_child",
+        title="Long Conversation",
+        messages=[
+            {"role": "user", "content": "new branch"},
+            {"role": "assistant", "content": "reply"},
+        ],
+        parent_session_id="snapshot_parent",
+        session_source="fork",
+        updated_at=200.0,
+        last_message_at=200.0,
+    )
+    snapshot.save(touch_updated_at=False)
+    fork.save(touch_updated_at=False)
+    monkeypatch.setattr(models, "_enrich_sidebar_lineage_metadata", lambda _sessions: None)
+
+    rows = models.all_sessions()
+
+    assert snapshot.path.exists(), "snapshot JSON must stay available for lineage traversal"
+    assert rows[0]["session_id"] == "manual_fork_child"
+    assert rows[0]["session_source"] == "fork"
+    assert models._sidebar_lineage_root_id(
+        {
+            "session_id": "lineage_child",
+            "_lineage_root_id": "lineage_root",
+            "parent_session_id": "snapshot_parent",
+        },
+        {
+            "snapshot_parent": {
+                "session_id": "snapshot_parent",
+                "parent_session_id": "snapshot_origin",
+            }
+        },
+    ) == "lineage_root"
+    assert models._sidebar_lineage_root_id(
+        {
+            "session_id": "child_session_sid",
+            "relationship_type": "child_session",
+            "parent_session_id": "snapshot_parent",
+        },
+        {
+            "snapshot_parent": {
+                "session_id": "snapshot_parent",
+                "parent_session_id": "snapshot_origin",
+            }
+        },
+    ) == "child_session_sid"
+
+
 def test_fuller_pre_compression_snapshot_replaces_shorter_visible_segment(monkeypatch):
     """If the hidden snapshot has the fuller transcript, keep it reachable.
 
@@ -1300,3 +1370,50 @@ def test_all_sessions_ignores_stale_index_entries():
     ids = {e["session_id"] for e in rows}
     assert "sess_a" in ids
     assert "ghost_sid" not in ids
+
+
+def test_background_index_rebuild_skips_after_session_dir_switch(tmp_path, monkeypatch):
+    """A delayed rebuild thread must not write into a newer isolated session dir."""
+    original_session_dir = models.SESSION_DIR
+    original_index_file = models.SESSION_INDEX_FILE
+    new_session_dir = tmp_path / "other-sessions"
+    new_session_dir.mkdir()
+    new_index_file = new_session_dir / "_index.json"
+
+    monkeypatch.setattr(models, "_SESSION_INDEX_REBUILD_THREAD", object())
+    monkeypatch.setattr(models, "_SESSION_INDEX_REBUILD_THREAD_TARGET", (
+        original_session_dir,
+        original_index_file,
+    ))
+    monkeypatch.setattr(models, "SESSION_DIR", new_session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", new_index_file)
+
+    models._rebuild_session_index_background(
+        original_session_dir,
+        original_index_file,
+    )
+
+    assert not new_index_file.exists()
+    monkeypatch.setattr(models, "SESSION_DIR", original_session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", original_index_file)
+    session = _make_session("late_switch_sid", "Late switch", updated_at=100.0)
+    session.save(skip_index=True)
+
+    original_write_session_index = models._write_session_index
+
+    def _switch_globals_then_write(*args, **kwargs):
+        monkeypatch.setattr(models, "SESSION_DIR", new_session_dir)
+        monkeypatch.setattr(models, "SESSION_INDEX_FILE", new_index_file)
+        return original_write_session_index(*args, **kwargs)
+
+    monkeypatch.setattr(models, "_write_session_index", _switch_globals_then_write)
+
+    models._rebuild_session_index_background(
+        original_session_dir,
+        original_index_file,
+    )
+
+    assert original_index_file.exists()
+    assert not new_index_file.exists()
+    rows = _read_index(original_index_file)
+    assert [row["session_id"] for row in rows] == ["late_switch_sid"]