Release v0.51.306 — Release JV (stage-a2 — branchy compression lineage freshest-tip) (#3761)

* fix(sessions): resolve branchy compression lineage to the freshest tip (#3751)

Compression parents can have multiple continuation-looking children when a stale
segment is resumed after a newer compressed branch already exists. The previous
projection followed the newest DIRECT child only, so it could hide the deeper
branch with the latest real activity and make a conversation look missing/stale
after compaction or session rotation.

- compression_tip() now walks ALL reachable continuation descendants (DFS) and
  selects the freshest importable (messageful) tip by (last_activity, depth).
- read_session_lineage_metadata() expands descendants from the materialized
  ancestors via the parent index (scoped, 20-hop cap), pulls per-session message
  stats from the messages table, and exposes a canonical _lineage_tip_id so the
  WebUI sidebar collapse picks the same tip as the projection.

Backward-compat hardening (two release-gate Codex findings, both fixed + tested):
The new message-stats / tip-scoring code must not raise on older/minimal or
non-standard state.db schemas, which previously collapsed the whole projection:
  - messages table with NO `timestamp` column → MAX(timestamp) raised in SQL →
    read_session_lineage_metadata returned {} (lost all lineage metadata).
  - ISO-8601 TEXT messages.timestamp → float()/raw comparison raised TypeError;
    in read_importable_agent_session_rows that propagated through compression_tip
    and get_cli_sessions() swallowed it, hiding ALL imported agent rows.
Fixes:
  - PRAGMA table_info(messages) detection: require session_id, only SELECT
    MAX(timestamp) when present (else NULL + COUNT only); fall back to message_count.
  - new _as_score() helper (first numerically-coercible value, else next
    candidate e.g. started_at) used at EVERY tip-scoring / sort site in both
    compression_tip() and freshest_continuation_tip(), plus the projection sort.
  - regression tests: lineage metadata survives REAL/absent/TEXT messages.timestamp,
    and read_importable_agent_session_rows survives a TEXT timestamp (no empty hide).

Co-authored-by: ai-ag2026 <[email protected]>

* docs(changelog): stamp v0.51.306 — Release JV (stage-a2 #3751)

---------

Co-authored-by: nesquena-hermes <[email protected]>

CHANGELOG.md

@@ -3,6 +3,11 @@
 
 ## [Unreleased]
 
+## [v0.51.306] — 2026-06-06 — Release JV (stage-a2 — branchy compression lineage resolves to the freshest tip)
+
+### Fixed
+- **A conversation no longer looks missing or stale after compaction when its compression lineage branched.** When a compression parent had multiple continuation-looking children — which happens when a stale segment is resumed after a newer compressed branch already exists — the sidebar projection followed only the newest *direct* child, so it could surface a dead-end branch and hide the deeper branch that actually has the latest activity. The projection now walks every reachable continuation descendant and selects the freshest *messageful* tip by `(last_activity, depth)`, and `read_session_lineage_metadata()` exposes the same canonical `_lineage_tip_id` so the sidebar collapse and the import projection agree on which branch is live. Older/minimal `state.db` files stay compatible — `source`, `message_count`, and the `messages` table (including schemas with no `timestamp` column or an ISO-8601 *text* timestamp) are all treated as optional and can no longer collapse the lineage metadata. (#3751, @ai-ag2026)
+
 ## [v0.51.305] — 2026-06-06 — Release JU (stage-p2b — dormant unified-SessionDB adapter groundwork)
 
 ### Changed

api/agent_sessions.py

@@ -120,6 +120,26 @@ def _as_positive_int(value) -> int:
         return 0
 
 
+def _as_score(*values) -> float:
+    """First numerically-coercible value as a float, else 0.0.
+
+    Used to score lineage tips by recency. ``last_message_at`` comes from
+    ``MAX(timestamp)`` and is normally a numeric epoch, but older/non-standard
+    state.db schemas can store an ISO-8601 *text* timestamp. Rather than letting
+    a non-numeric value raise ValueError (which previously escaped the DB
+    try-block and dropped all lineage metadata), fall through to the next
+    candidate (e.g. ``started_at``).
+    """
+    for value in values:
+        if value in (None, ""):
+            continue
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            continue
+    return 0.0
+
+
 def _count_user_turns(row: dict) -> int:
     user_turns = row.get("actual_user_message_count")
     if user_turns is None:
@@ -279,23 +299,50 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
         children.sort(key=lambda row: row.get('started_at') or 0, reverse=True)
 
     def compression_tip(row: dict) -> tuple[dict | None, int]:
-        current = row
-        seen = {row['id']}
+        """Return the freshest importable continuation descendant for ``row``.
+
+        Compression parents can have multiple continuation-looking children when
+        a stale segment is resumed after a newer compressed branch already
+        exists. Picking the newest *direct* child can hide the branch whose
+        deeper descendant has the actual latest activity. Walk all reachable
+        continuation descendants and select by real message activity instead.
+        """
         latest_importable = row if (row.get('actual_message_count') or 0) > 0 else None
-        segment_count = 1
-        for _ in range(len(rows_by_id) + 1):
-            candidates = [
-                child for child in children_by_parent.get(current['id'], [])
-                if child['id'] not in seen and _is_continuation_session(current, child)
-            ]
-            if not candidates:
-                return latest_importable, segment_count
-            current = candidates[0]
-            seen.add(current['id'])
+        segment_count = 0
+        best_depth = 1
+        best_score = (
+            _as_score(latest_importable.get('last_activity'), latest_importable.get('started_at'))
+            if latest_importable
+            else 0
+        )
+        stack: list[tuple[dict, int]] = [(row, 1)]
+        seen: set[str] = set()
+
+        while stack:
+            current, depth = stack.pop()
+            current_id = current.get('id')
+            if not current_id or current_id in seen:
+                continue
+            seen.add(current_id)
             segment_count += 1
-            if (current.get('actual_message_count') or 0) > 0:
+
+            current_score = _as_score(current.get('last_activity'), current.get('started_at'))
+            if (
+                (current.get('actual_message_count') or 0) > 0
+                and (current_score > best_score or (current_score == best_score and depth >= best_depth))
+            ):
                 latest_importable = current
-        return latest_importable, segment_count
+                best_depth = depth
+                best_score = current_score
+            for child in children_by_parent.get(current_id, []):
+                child_id = child.get('id')
+                if not child_id or child_id in seen:
+                    continue
+                if not _is_continuation_session(current, child):
+                    continue
+                stack.append((child, depth + 1))
+
+        return latest_importable, max(segment_count, 1)
 
     projected = []
     for row in rows:
@@ -338,7 +385,7 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
         projected.append(merged)
 
     projected.sort(
-        key=lambda row: row.get('last_activity') or row.get('started_at') or 0,
+        key=lambda row: _as_score(row.get('last_activity'), row.get('started_at')),
         reverse=True,
     )
     return projected
@@ -346,7 +393,7 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
 
 def read_importable_agent_session_rows(
     db_path: Path,
-    limit: int = 200,
+    limit: int | None = 200,
     log=None,
     exclude_sources: tuple[str, ...] | None = ("cron", "webui"),
 ) -> list[dict]:
@@ -684,6 +731,8 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
             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)
+            source_expr = _optional_col('source', session_cols)
+            message_count_expr = _optional_col('message_count', session_cols, '0')
             # 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
@@ -692,7 +741,9 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
             #
             # Fetch the wanted ids first, then chase parent_session_id chains
             # in batches until no new ids appear. Each batch hits PRIMARY KEY
-            # so it's effectively O(N) lookups.
+            # so it's effectively O(N) lookups. Then walk continuation children
+            # from the materialized ancestors so branchy compression lineages can
+            # mark the real freshest tip, not just the newest direct sibling.
             #
             # IN-clause is chunked to 500 to stay under SQLITE_MAX_VARIABLE_NUMBER
             # on older sqlite (Python 3.9 ships sqlite 3.31 which defaults to 999;
@@ -717,7 +768,7 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
                     placeholders = ','.join('?' * len(chunk))
                     cur.execute(
                         f"""
-                        SELECT s.id, s.source, {session_source_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason
+                        SELECT s.id, {source_expr}, {session_source_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason, {message_count_expr}
                         FROM sessions s
                         WHERE s.id IN ({placeholders})
                         """,
@@ -730,9 +781,137 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
                     parent_id = rows.get(sid, {}).get('parent_session_id')
                     if parent_id and parent_id not in rows and parent_id not in to_fetch:
                         to_fetch.add(parent_id)
+
+            # Fetch descendants from the discovered ancestors using the parent
+            # index. This keeps the sidebar read scoped while still giving the
+            # collapse metadata enough information to choose the active branch.
+            to_expand = set(rows)
+            expanded: set[str] = set()
+            for _hop in range(20):
+                frontier = [sid for sid in to_expand if sid not in expanded]
+                if not frontier:
+                    break
+                to_expand = set()
+                for i in range(0, len(frontier), IN_CHUNK):
+                    chunk = frontier[i:i + IN_CHUNK]
+                    placeholders = ','.join('?' * len(chunk))
+                    cur.execute(
+                        f"""
+                        SELECT s.id, {source_expr}, {session_source_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason, {message_count_expr}
+                        FROM sessions s
+                        WHERE s.parent_session_id IN ({placeholders})
+                        """,
+                        chunk,
+                    )
+                    for row in cur.fetchall():
+                        child = dict(row)
+                        rows[child['id']] = child
+                        parent_id = child.get('parent_session_id')
+                        parent = rows.get(str(parent_id)) if parent_id else None
+                        if parent and child['id'] not in expanded and _is_continuation_session(parent, child):
+                            to_expand.add(child['id'])
+                expanded.update(frontier)
+
+            message_stats: dict[str, dict] = {}
+            cur.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'messages'")
+            has_messages_table = cur.fetchone() is not None
+            # Older/minimal state.db schemas can have a `messages` table WITHOUT a
+            # `timestamp` column (or with a non-numeric one). Detect the columns
+            # rather than gating on table existence alone: require `session_id`,
+            # and only select MAX(timestamp) when that column is actually present
+            # so the query can't raise and collapse the whole lineage metadata.
+            messages_has_session_id = False
+            messages_has_timestamp = False
+            if has_messages_table:
+                cur.execute("PRAGMA table_info(messages)")
+                _message_cols = {row[1] for row in cur.fetchall()}
+                messages_has_session_id = 'session_id' in _message_cols
+                messages_has_timestamp = 'timestamp' in _message_cols
+            use_messages_query = has_messages_table and messages_has_session_id
+            row_ids = list(rows)
+            if use_messages_query:
+                last_at_expr = "MAX(timestamp) AS last_message_at" if messages_has_timestamp else "NULL AS last_message_at"
+                for i in range(0, len(row_ids), IN_CHUNK):
+                    chunk = row_ids[i:i + IN_CHUNK]
+                    placeholders = ','.join('?' * len(chunk))
+                    cur.execute(
+                        f"""
+                        SELECT session_id, COUNT(*) AS actual_message_count, {last_at_expr}
+                        FROM messages
+                        WHERE session_id IN ({placeholders})
+                        GROUP BY session_id
+                        """,
+                        chunk,
+                    )
+                    for row in cur.fetchall():
+                        message_stats[row['session_id']] = dict(row)
+            for sid, row in rows.items():
+                stats = message_stats.get(sid) or {}
+                if use_messages_query:
+                    row['actual_message_count'] = int(stats.get('actual_message_count') or 0)
+                else:
+                    row['actual_message_count'] = int(row.get('message_count') or 0)
+                row['last_message_at'] = stats.get('last_message_at')
     except Exception:
         return {}
 
+    children_by_parent: dict[str, list[dict]] = {}
+    for row in rows.values():
+        parent_id = row.get('parent_session_id')
+        if parent_id:
+            children_by_parent.setdefault(parent_id, []).append(row)
+
+    def continuation_root_and_depth(sid: str) -> tuple[str, int]:
+        root_id = sid
+        current_id = sid
+        depth = 1
+        seen = {sid}
+        while True:
+            current = rows.get(current_id)
+            raw_parent_id = current.get('parent_session_id') if current else None
+            parent_id = str(raw_parent_id) if raw_parent_id else ''
+            if not parent_id:
+                break
+            parent = rows.get(parent_id)
+            if not parent or parent_id in seen:
+                break
+            if not _is_continuation_session(parent, current):
+                break
+            root_id = parent_id
+            current_id = parent_id
+            seen.add(parent_id)
+            depth += 1
+        return root_id, depth
+
+    def freshest_continuation_tip(root_id: str) -> tuple[str, int]:
+        best_id = root_id
+        best_depth = 1
+        segment_count = 0
+        best_score = _as_score(rows.get(root_id, {}).get('last_message_at'), rows.get(root_id, {}).get('started_at'))
+        stack: list[tuple[str, int]] = [(root_id, 1)]
+        seen: set[str] = set()
+        while stack:
+            current_id, depth = stack.pop()
+            if current_id in seen:
+                continue
+            seen.add(current_id)
+            current = rows.get(current_id)
+            if not current:
+                continue
+            segment_count += 1
+            actual_count = int(current.get('actual_message_count') or 0)
+            score = _as_score(current.get('last_message_at'), current.get('started_at'))
+            if actual_count > 0 and (score > best_score or (score == best_score and depth >= best_depth)):
+                best_id = current_id
+                best_depth = depth
+                best_score = score
+            for child in children_by_parent.get(current_id, []):
+                if _is_continuation_session(current, child):
+                    stack.append((child['id'], depth + 1))
+
+        return best_id, max(segment_count, best_depth)
+
+    lineage_tip_cache: dict[str, tuple[str, int]] = {}
     metadata: dict[str, dict] = {}
     for sid in wanted:
         row = rows.get(sid)
@@ -770,26 +949,15 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
                     entry['_parent_lineage_root_id'] = parent_root
                 continue
 
-        root_id = sid
-        current_id = sid
-        segment_count = 1
-        seen = {sid}
-        while True:
-            current = rows.get(current_id)
-            parent_id = current.get('parent_session_id') if current else None
-            parent = rows.get(parent_id) if parent_id else None
-            if not parent or parent_id in seen:
-                break
-            if not _is_continuation_session(parent, current):
-                break
-            root_id = parent_id
-            current_id = parent_id
-            seen.add(parent_id)
-            segment_count += 1
+        root_id, segment_count = continuation_root_and_depth(sid)
 
         if root_id != sid:
             entry = metadata.setdefault(sid, {})
             entry['_lineage_root_id'] = root_id
-            entry['_compression_segment_count'] = segment_count
+            if root_id not in lineage_tip_cache:
+                lineage_tip_cache[root_id] = freshest_continuation_tip(root_id)
+            tip_id, tip_depth = lineage_tip_cache[root_id]
+            entry['_lineage_tip_id'] = tip_id
+            entry['_compression_segment_count'] = max(segment_count, tip_depth)
 
     return metadata

tests/test_gateway_sync.py

@@ -427,6 +427,144 @@ def test_compression_chain_collapses_to_latest_tip_in_sidebar():
         post('/api/settings', {'show_cli_sessions': False})
 
 
+def test_compression_lineage_prefers_freshest_descendant_over_newer_direct_sibling():
+    """A later-started stale sibling must not hide a deeper active branch."""
+    conn = _ensure_state_db()
+    ids_to_remove = (
+        'branch_root_001',
+        'branch_old_mid_001',
+        'branch_fresh_tip_001',
+        'branch_empty_stale_tip_001',
+        'branch_newer_direct_sibling_001',
+    )
+    t0 = time.time() - 800
+    try:
+        _insert_agent_session_row(
+            conn,
+            'branch_root_001',
+            title='Qwen Routing Audit',
+            started_at=t0,
+            ended_at=t0 + 100,
+            end_reason='compression',
+            messages=2,
+        )
+        _insert_agent_session_row(
+            conn,
+            'branch_old_mid_001',
+            title='Qwen Routing Audit #2',
+            started_at=t0 + 101,
+            parent_session_id='branch_root_001',
+            ended_at=t0 + 200,
+            end_reason='compression',
+            messages=2,
+        )
+        _insert_agent_session_row(
+            conn,
+            'branch_newer_direct_sibling_001',
+            title='Qwen Routing Audit #3 stale sibling',
+            started_at=t0 + 150,
+            parent_session_id='branch_root_001',
+            messages=2,
+        )
+        _insert_agent_session_row(
+            conn,
+            'branch_fresh_tip_001',
+            title='Qwen Routing Audit #4 freshest tip',
+            started_at=t0 + 500,
+            parent_session_id='branch_old_mid_001',
+            messages=2,
+        )
+        _insert_agent_session_row(
+            conn,
+            'branch_empty_stale_tip_001',
+            title='Qwen Routing Audit #5 empty stale tip',
+            started_at=t0 + 700,
+            parent_session_id='branch_old_mid_001',
+            messages=0,
+        )
+        conn.execute(
+            "UPDATE sessions SET message_count = 3 WHERE id = ?",
+            ('branch_empty_stale_tip_001',),
+        )
+        conn.commit()
+
+        post('/api/settings', {'show_cli_sessions': True})
+        data, status = get('/api/sessions')
+        assert status == 200
+        ids = {s.get('session_id') for s in data.get('sessions', [])}
+        tip = next((s for s in data.get('sessions', []) if s.get('session_id') == 'branch_fresh_tip_001'), None)
+
+        assert 'branch_fresh_tip_001' in ids
+        assert 'branch_newer_direct_sibling_001' not in ids
+        assert 'branch_empty_stale_tip_001' not in ids
+        assert tip is not None
+        assert tip.get('title') == 'Qwen Routing Audit'
+        assert abs(tip.get('updated_at') - (t0 + 501)) < 0.01
+        assert tip.get('_lineage_root_id') == 'branch_root_001'
+        assert tip.get('_lineage_tip_id') == 'branch_fresh_tip_001'
+        assert tip.get('_compression_segment_count') == 5
+
+        from api.agent_sessions import read_importable_agent_session_rows
+
+        rows = read_importable_agent_session_rows(_get_state_db_path(), limit=None)
+        projected_tip = next((row for row in rows if row.get('id') == 'branch_fresh_tip_001'), None)
+        assert projected_tip is not None
+        assert projected_tip.get('_lineage_root_id') == 'branch_root_001'
+        assert projected_tip.get('_lineage_tip_id') == 'branch_fresh_tip_001'
+        assert projected_tip.get('_compression_segment_count') == 5
+
+        from api.agent_sessions import read_session_lineage_metadata
+
+        metadata = read_session_lineage_metadata(
+            _get_state_db_path(),
+            {'branch_old_mid_001', 'branch_fresh_tip_001', 'branch_empty_stale_tip_001', 'branch_newer_direct_sibling_001'},
+        )
+        assert metadata['branch_old_mid_001'].get('_lineage_tip_id') == 'branch_fresh_tip_001'
+        assert metadata['branch_empty_stale_tip_001'].get('_lineage_tip_id') == 'branch_fresh_tip_001'
+        assert metadata['branch_newer_direct_sibling_001'].get('_lineage_tip_id') == 'branch_fresh_tip_001'
+        assert metadata['branch_newer_direct_sibling_001'].get('_compression_segment_count') == 5
+    finally:
+        try:
+            _remove_test_sessions(conn, *ids_to_remove)
+            conn.close()
+        except Exception:
+            pass
+        post('/api/settings', {'show_cli_sessions': False})
+
+
+def test_compression_projection_handles_deep_lineage_iteratively():
+    """Very deep compression chains should not depend on Python recursion depth."""
+    from api.agent_sessions import _project_agent_session_rows
+
+    rows = []
+    previous = None
+    for idx in range(1100):
+        sid = f'deep_chain_{idx:04d}'
+        started_at = float(idx * 2)
+        rows.append({
+            'id': sid,
+            'title': 'Deep Chain' if idx == 0 else f'Deep Chain #{idx + 1}',
+            'source': 'cli',
+            'started_at': started_at,
+            'parent_session_id': previous,
+            'ended_at': started_at + 1 if idx < 1099 else None,
+            'end_reason': 'compression' if idx < 1099 else None,
+            'actual_message_count': 1,
+            'actual_user_message_count': 1,
+            'message_count': 1,
+            'last_activity': started_at + 0.5,
+        })
+        previous = sid
+
+    projected = _project_agent_session_rows(rows)
+
+    assert len(projected) == 1
+    assert projected[0]['id'] == 'deep_chain_1099'
+    assert projected[0]['_lineage_root_id'] == 'deep_chain_0000'
+    assert projected[0]['_lineage_tip_id'] == 'deep_chain_1099'
+    assert projected[0]['_compression_segment_count'] == 1100
+
+
 def test_compression_chain_with_empty_latest_tip_falls_back_to_latest_importable_segment():
     """Empty latest tips should not make the whole conversation disappear."""
     conn = _ensure_state_db()

tests/test_pr1370_lineage_metadata_perf_and_orphan.py

@@ -132,6 +132,42 @@ def test_orphan_parent_reference_not_exposed_in_metadata(tmp_path):
     )
 
 
+def test_old_schema_without_source_or_messages_table_keeps_lineage(tmp_path):
+    """Old state.db schemas may lack optional source/message tables but still carry lineage."""
+    from api.agent_sessions import read_session_lineage_metadata
+
+    db = tmp_path / "state.db"
+    conn = sqlite3.connect(str(db))
+    conn.executescript("""
+    CREATE TABLE sessions (
+        id TEXT PRIMARY KEY,
+        title TEXT,
+        started_at REAL NOT NULL,
+        parent_session_id TEXT,
+        ended_at REAL,
+        end_reason TEXT
+    );
+    CREATE INDEX idx_sessions_parent ON sessions(parent_session_id);
+    """)
+    t0 = time.time() - 100
+    conn.execute(
+        "INSERT INTO sessions (id, title, started_at, ended_at, end_reason) VALUES (?, ?, ?, ?, ?)",
+        ("old_root", "old_root", t0, t0 + 5, "compression"),
+    )
+    conn.execute(
+        "INSERT INTO sessions (id, title, started_at, parent_session_id) VALUES (?, ?, ?, ?)",
+        ("old_tip", "old_tip", t0 + 6, "old_root"),
+    )
+    conn.commit()
+    conn.close()
+
+    result = read_session_lineage_metadata(db, ["old_tip"])
+
+    assert result["old_tip"]["parent_session_id"] == "old_root"
+    assert result["old_tip"]["_lineage_root_id"] == "old_root"
+    assert result["old_tip"]["_compression_segment_count"] == 2
+
+
 def test_cycle_in_parent_chain_terminates(tmp_path):
     """Pathological data with a parent cycle (A→B→A) must not infinite-loop."""
     from api.agent_sessions import read_session_lineage_metadata
@@ -282,3 +318,133 @@ def test_non_compression_parent_does_not_extend_lineage(tmp_path):
     # _lineage_root_id should NOT be set — chain doesn't span the boundary
     assert "_lineage_root_id" not in entry
     assert "_compression_segment_count" not in entry
+
+
+
+# ── #3751 backward-compat: messages-table schema variants must not collapse
+# the lineage metadata (regression for the gate finding on stage-a2/v0.51.306) ──
+
+def _make_db_with_messages(path, *, timestamp_type):
+    """Build a state.db with a compression lineage and a messages table whose
+    timestamp column is REAL, absent, or TEXT (ISO-8601)."""
+    conn = sqlite3.connect(str(path))
+    conn.executescript("""
+    CREATE TABLE sessions (
+        id TEXT PRIMARY KEY,
+        source TEXT,
+        title TEXT,
+        model TEXT,
+        started_at REAL NOT NULL,
+        message_count INTEGER DEFAULT 0,
+        parent_session_id TEXT,
+        ended_at REAL,
+        end_reason TEXT
+    );
+    CREATE INDEX idx_sessions_parent ON sessions(parent_session_id);
+    """)
+    if timestamp_type == "absent":
+        conn.execute("CREATE TABLE messages (session_id TEXT, role TEXT, content TEXT)")
+    else:
+        conn.execute(f"CREATE TABLE messages (session_id TEXT, role TEXT, content TEXT, timestamp {timestamp_type})")
+    # Compression chain: root --(compression)--> tip
+    now = time.time()
+    conn.execute(
+        "INSERT INTO sessions (id, source, title, model, started_at, parent_session_id, ended_at, end_reason) "
+        "VALUES ('root', 'webui', 'root', 'openai/gpt-5', ?, NULL, ?, 'compression')",
+        (now - 100, now - 50),
+    )
+    conn.execute(
+        "INSERT INTO sessions (id, source, title, model, started_at, parent_session_id, ended_at, end_reason) "
+        "VALUES ('tip', 'webui', 'tip', 'openai/gpt-5', ?, 'root', NULL, NULL)",
+        (now - 40,),
+    )
+    if timestamp_type == "absent":
+        conn.execute("INSERT INTO messages (session_id, role, content) VALUES ('tip', 'user', 'hi')")
+    elif timestamp_type == "TEXT":
+        conn.execute(
+            "INSERT INTO messages (session_id, role, content, timestamp) VALUES ('tip', 'user', 'hi', '2026-06-06T12:00:00Z')"
+        )
+    else:  # REAL
+        conn.execute(
+            "INSERT INTO messages (session_id, role, content, timestamp) VALUES ('tip', 'user', 'hi', ?)",
+            (now - 35,),
+        )
+    conn.commit()
+    conn.close()
+
+
+@pytest.mark.parametrize("timestamp_type", ["REAL", "absent", "TEXT"])
+def test_lineage_metadata_survives_messages_timestamp_schema_variants(tmp_path, timestamp_type):
+    """The branchy-lineage tip resolver pulls per-session message stats from the
+    messages table. Older/minimal state.db schemas can have a messages table with
+    NO timestamp column, or a non-numeric (ISO-8601 text) timestamp. Neither may
+    raise out of the DB block and collapse ALL lineage metadata to {} — which is
+    a silent regression vs. the prior behavior that returned metadata.
+    """
+    from api.agent_sessions import read_session_lineage_metadata
+
+    db = tmp_path / "state.db"
+    _make_db_with_messages(db, timestamp_type=timestamp_type)
+
+    result = read_session_lineage_metadata(db, ["tip"])
+    entry = result.get("tip", {})
+    # The compression lineage must still be reported (not silently dropped).
+    assert entry.get("_lineage_root_id") == "root", (
+        f"lineage metadata collapsed for messages.timestamp={timestamp_type}: {result!r}"
+    )
+    # And the canonical tip must resolve to the messageful continuation.
+    assert entry.get("_lineage_tip_id") == "tip"
+
+
+def test_importable_rows_survive_text_timestamp_in_messages(tmp_path):
+    """read_importable_agent_session_rows() builds a per-session MAX(timestamp)
+    and the compression_tip() DFS scores tips by last_activity. An older/
+    non-standard messages.timestamp stored as ISO-8601 TEXT must not raise a
+    TypeError out of the projection (get_cli_sessions() would swallow it and
+    return [] — silently hiding ALL imported agent rows). Sibling-path guard to
+    the read_session_lineage_metadata fix.
+    """
+    from api.agent_sessions import read_importable_agent_session_rows
+
+    db = tmp_path / "state.db"
+    conn = sqlite3.connect(str(db))
+    conn.executescript("""
+    CREATE TABLE sessions (
+        id TEXT PRIMARY KEY,
+        source TEXT,
+        title TEXT,
+        model TEXT,
+        started_at REAL NOT NULL,
+        message_count INTEGER DEFAULT 0,
+        parent_session_id TEXT,
+        ended_at REAL,
+        end_reason TEXT
+    );
+    CREATE INDEX idx_sessions_parent ON sessions(parent_session_id);
+    CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, timestamp TEXT);
+    """)
+    now = time.time()
+    # compression chain root --(compression)--> tip, tip has a messageful row
+    conn.execute(
+        "INSERT INTO sessions (id, source, title, model, started_at, message_count, parent_session_id, ended_at, end_reason) "
+        "VALUES ('root', 'cli', 'root', 'openai/gpt-5', ?, 0, NULL, ?, 'compression')",
+        (now - 100, now - 50),
+    )
+    conn.execute(
+        "INSERT INTO sessions (id, source, title, model, started_at, message_count, parent_session_id, ended_at, end_reason) "
+        "VALUES ('tip', 'cli', 'tip', 'openai/gpt-5', ?, 1, 'root', NULL, NULL)",
+        (now - 40,),
+    )
+    # ISO-8601 TEXT timestamp — the value MAX() would return as a string
+    conn.execute(
+        "INSERT INTO messages (session_id, role, content, timestamp) VALUES ('tip', 'user', 'hi', '2026-06-06T12:00:00Z')"
+    )
+    conn.commit()
+    conn.close()
+
+    # Must NOT raise; must surface the imported chain (cli source).
+    rows = read_importable_agent_session_rows(db, limit=None, exclude_sources=None)
+    ids = {r["id"] for r in rows}
+    assert ids, "import projection returned no rows on a TEXT messages.timestamp (would hide all CLI sessions)"
+    # the chain collapses to its tip
+    assert "tip" in ids