diff --git a/CHANGELOG.md b/CHANGELOG.md index a6cffdd8..a12ed882 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/api/agent_sessions.py b/api/agent_sessions.py index 1da2625d..bc39649f 100644 --- a/api/agent_sessions.py +++ b/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 diff --git a/tests/test_gateway_sync.py b/tests/test_gateway_sync.py index 3571d375..3081ed15 100644 --- a/tests/test_gateway_sync.py +++ b/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() diff --git a/tests/test_pr1370_lineage_metadata_perf_and_orphan.py b/tests/test_pr1370_lineage_metadata_perf_and_orphan.py index cdab67b4..02f999a6 100644 --- a/tests/test_pr1370_lineage_metadata_perf_and_orphan.py +++ b/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