Merge PR #2063 into stage-338

fix: keep explicit forks out of lineage report
by @dso2ng

api/agent_sessions.py

@@ -199,6 +199,8 @@ def _is_continuation_session(parent: dict | None, child: dict | None) -> bool:
     """
     if not parent or not child:
         return False
+    if str(child.get('session_source') or '').strip().lower() == 'fork':
+        return False
     parent_source = str(parent.get('source') or '').strip().lower()
     child_source = str(child.get('source') or '').strip().lower()
     if parent_source and child_source and parent_source != child_source:
@@ -379,6 +381,7 @@ def read_importable_agent_session_rows(
             return []
 
         parent_expr = _optional_col('parent_session_id', session_cols)
+        session_source_expr = _optional_col('session_source', session_cols)
         ended_expr = _optional_col('ended_at', session_cols)
         end_reason_expr = _optional_col('end_reason', session_cols)
         user_id_expr = _optional_col('user_id', session_cols)
@@ -408,6 +411,7 @@ def read_importable_agent_session_rows(
             f"""
             SELECT s.id, s.title, s.model, s.message_count,
                    s.started_at, s.source,
+                   {session_source_expr},
                    {user_id_expr},
                    {chat_id_expr},
                    {chat_type_expr},
@@ -496,6 +500,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops:
                 return _empty_lineage_report(sid)
 
             source_expr = _optional_col('source', session_cols)
+            session_source_expr = _optional_col('session_source', session_cols)
             title_expr = _optional_col('title', session_cols)
             started_expr = _optional_col('started_at', session_cols, '0')
             ended_expr = _optional_col('ended_at', session_cols)
@@ -509,6 +514,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops:
                     f"""
                     SELECT s.id,
                            {source_expr},
+                           {session_source_expr},
                            {title_expr},
                            {started_expr},
                            {parent_expr},
@@ -551,6 +557,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops:
                     f"""
                     SELECT s.id,
                            {source_expr},
+                           {session_source_expr},
                            {title_expr},
                            {started_expr},
                            {parent_expr},
@@ -620,6 +627,7 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
             session_cols = {row[1] for row in cur.fetchall()}
             if 'parent_session_id' not in session_cols or 'end_reason' not in session_cols:
                 return {}
+            session_source_expr = _optional_col('session_source', session_cols)
             # Scoped fetch via PRIMARY KEY + idx_sessions_parent rather than a
             # full table scan. The sessions table grows unbounded over time
             # (1000+ rows is normal, 10000+ for power users), and this function
@@ -653,9 +661,9 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
                     placeholders = ','.join('?' * len(chunk))
                     cur.execute(
                         f"""
-                        SELECT id, source, title, started_at, parent_session_id, ended_at, end_reason
-                        FROM sessions
-                        WHERE id IN ({placeholders})
+                        SELECT s.id, s.source, {session_source_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason
+                        FROM sessions s
+                        WHERE s.id IN ({placeholders})
                         """,
                         chunk,
                     )

tests/test_session_lineage_metadata_api.py

@@ -32,6 +32,7 @@ def _ensure_state_db(path):
         CREATE TABLE sessions (
             id TEXT PRIMARY KEY,
             source TEXT,
+            session_source TEXT,
             title TEXT,
             model TEXT,
             started_at REAL NOT NULL,
@@ -45,14 +46,14 @@ def _ensure_state_db(path):
     return conn
 
 
-def _insert_state_row(conn, sid, *, parent=None, ended_at=None, end_reason=None, started_at=None, source='webui'):
+def _insert_state_row(conn, sid, *, parent=None, ended_at=None, end_reason=None, started_at=None, source='webui', session_source=None):
     conn.execute(
         """
         INSERT INTO sessions
-        (id, source, title, model, started_at, message_count, parent_session_id, ended_at, end_reason)
-        VALUES (?, ?, ?, 'openai/gpt-5', ?, 2, ?, ?, ?)
+        (id, source, session_source, title, model, started_at, message_count, parent_session_id, ended_at, end_reason)
+        VALUES (?, ?, ?, ?, 'openai/gpt-5', ?, 2, ?, ?, ?)
         """,
-        (sid, source, sid, started_at or time.time(), parent, ended_at, end_reason),
+        (sid, source, session_source, sid, started_at or time.time(), parent, ended_at, end_reason),
     )
     conn.commit()
 
@@ -99,6 +100,40 @@ def test_all_sessions_exposes_state_db_lineage_metadata_for_webui_json_sessions(
         conn.close()
 
 
+def test_all_sessions_keeps_explicit_forks_out_of_state_db_lineage_metadata(_isolate):
+    conn = _ensure_state_db(_isolate)
+    t0 = time.time() - 100
+    try:
+        _save_webui_session("lineage_api_root", title="Visible root", updated_at=t0)
+        _save_webui_session("lineage_api_fork", title="Explicit fork", updated_at=t0 + 10)
+        _insert_state_row(
+            conn,
+            "lineage_api_root",
+            started_at=t0,
+            ended_at=t0 + 5,
+            end_reason="compression",
+        )
+        _insert_state_row(
+            conn,
+            "lineage_api_fork",
+            parent="lineage_api_root",
+            started_at=t0 + 6,
+            session_source="fork",
+        )
+
+        rows = {row["session_id"]: row for row in all_sessions()}
+
+        fork = rows["lineage_api_fork"]
+        assert fork.get("parent_session_id") == "lineage_api_root"
+        assert fork.get("relationship_type") == "child_session"
+        assert fork.get("parent_title") == "lineage_api_root"
+        assert fork.get("_parent_lineage_root_id") == "lineage_api_root"
+        assert "_lineage_root_id" not in fork
+        assert "_compression_segment_count" not in fork
+    finally:
+        conn.close()
+
+
 def test_non_compression_state_db_parent_does_not_create_sidebar_lineage(_isolate):
     conn = _ensure_state_db(_isolate)
     t0 = time.time() - 100

tests/test_session_lineage_report.py

@@ -18,6 +18,7 @@ def _ensure_state_db(path):
         CREATE TABLE sessions (
             id TEXT PRIMARY KEY,
             source TEXT,
+            session_source TEXT,
             title TEXT,
             model TEXT,
             started_at REAL NOT NULL,
@@ -26,19 +27,34 @@ def _ensure_state_db(path):
             ended_at REAL,
             end_reason TEXT
         );
+        CREATE TABLE messages (
+            id TEXT PRIMARY KEY,
+            session_id TEXT,
+            role TEXT,
+            content TEXT,
+            timestamp REAL
+        );
         """
     )
     return conn
 
 
-def _insert_state_row(conn, sid, *, parent=None, ended_at=None, end_reason=None, started_at=None, source="webui"):
+def _insert_state_row(conn, sid, *, parent=None, ended_at=None, end_reason=None, started_at=None, source="webui", session_source=None):
     conn.execute(
         """
         INSERT INTO sessions
-        (id, source, title, model, started_at, message_count, parent_session_id, ended_at, end_reason)
-        VALUES (?, ?, ?, 'openai/gpt-5', ?, 2, ?, ?, ?)
+        (id, source, session_source, title, model, started_at, message_count, parent_session_id, ended_at, end_reason)
+        VALUES (?, ?, ?, ?, 'openai/gpt-5', ?, 2, ?, ?, ?)
         """,
-        (sid, source, sid.replace("_", " "), started_at or time.time(), parent, ended_at, end_reason),
+        (sid, source, session_source, sid.replace("_", " "), started_at or time.time(), parent, ended_at, end_reason),
+    )
+    conn.commit()
+
+
+def _insert_message(conn, sid, *, timestamp=None, role="user"):
+    conn.execute(
+        "INSERT INTO messages (id, session_id, role, content, timestamp) VALUES (?, ?, ?, 'hello', ?)",
+        (f"msg_{sid}_{role}", sid, role, timestamp or time.time()),
     )
     conn.commit()
 
@@ -104,6 +120,59 @@ def test_lineage_report_keeps_cross_surface_parent_out_of_hidden_segments(tmp_pa
         conn.close()
 
 
+def test_lineage_report_keeps_explicit_forks_out_of_hidden_segments(tmp_path):
+    conn = _ensure_state_db(tmp_path / "state.db")
+    t0 = time.time() - 100
+    try:
+        _insert_state_row(conn, "lineage_report_root", started_at=t0, ended_at=t0 + 5, end_reason="compression")
+        _insert_state_row(
+            conn,
+            "lineage_report_fork",
+            parent="lineage_report_root",
+            started_at=t0 + 6,
+            session_source="fork",
+        )
+
+        report = agent_sessions.read_session_lineage_report(tmp_path / "state.db", "lineage_report_fork")
+
+        assert report["lineage_key"] == "lineage_report_fork"
+        assert report["tip_session_id"] == "lineage_report_fork"
+        assert report["total_segments"] == 1
+        assert [s["session_id"] for s in report["segments"]] == ["lineage_report_fork"]
+        assert report["segments"][0]["role"] == "tip"
+        assert report["children"] == []
+        assert report["manual_review"] is False
+    finally:
+        conn.close()
+
+
+def test_importable_agent_projection_keeps_explicit_forks_out_of_compression_lineage(tmp_path):
+    conn = _ensure_state_db(tmp_path / "state.db")
+    t0 = time.time() - 100
+    try:
+        _insert_state_row(conn, "lineage_report_root", started_at=t0, ended_at=t0 + 5, end_reason="compression")
+        _insert_state_row(
+            conn,
+            "lineage_report_fork",
+            parent="lineage_report_root",
+            started_at=t0 + 6,
+            session_source="fork",
+        )
+        _insert_message(conn, "lineage_report_fork", timestamp=t0 + 7)
+
+        rows = agent_sessions.read_importable_agent_session_rows(tmp_path / "state.db", exclude_sources=())
+
+        assert [row["id"] for row in rows] == ["lineage_report_fork"]
+        fork = rows[0]
+        assert fork.get("relationship_type") == "child_session"
+        assert fork.get("parent_session_id") == "lineage_report_root"
+        assert fork.get("_parent_lineage_root_id") == "lineage_report_root"
+        assert "_lineage_root_id" not in fork
+        assert "_compression_segment_count" not in fork
+    finally:
+        conn.close()
+
+
 def test_lineage_report_surfaces_non_continuation_children_without_mutation(tmp_path):
     conn = _ensure_state_db(tmp_path / "state.db")
     t0 = time.time() - 100