diff --git a/api/models.py b/api/models.py index 3ee491fba77..aa4558711e5 100644 --- a/api/models.py +++ b/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 diff --git a/tests/test_session_index.py b/tests/test_session_index.py index 12d37efa10e..1dfda2db982 100644 --- a/tests/test_session_index.py +++ b/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"]