diff --git a/api/agent_sessions.py b/api/agent_sessions.py index 97086795..f682c8f3 100644 --- a/api/agent_sessions.py +++ b/api/agent_sessions.py @@ -1,4 +1,5 @@ """Shared helpers for reading Hermes Agent sessions from state.db.""" +import json import logging import sqlite3 from contextlib import closing @@ -65,6 +66,15 @@ MESSAGING_SOURCES = { CLI_MIN_UNTITLED_MESSAGE_COUNT = 6 CLI_MIN_UNTITLED_USER_MESSAGE_COUNT = 2 +# Sub-second scheduling/write-order races during a compression/cli_close +# handoff can persist the continuation row with ``started_at`` a few +# milliseconds BEFORE the parent's ``ended_at`` lands (#6931). The tolerance +# below lets those rows still classify as the next segment; the fork / +# model_config branch-marker / cross-source / end_reason guards in +# ``_is_continuation_session`` already rule out unrelated rows, so a small +# window cannot collapse genuinely concurrently started children. +CONTINUATION_STARTED_AT_TOLERANCE_SECONDS = 2.0 + SOURCE_LABELS = { 'acp': 'ACP', 'api_server': 'API', @@ -318,6 +328,30 @@ def is_cli_session_row_visible(row: dict) -> bool: return _count_user_turns(row) >= CLI_MIN_UNTITLED_USER_MESSAGE_COUNT +def _branch_markers(row: dict | None) -> tuple[str | None, str | None]: + """Return ``(_branched_from, _delegate_from)`` from a row's ``model_config``. + + Hermes Agent stamps explicit branches and delegate/subagent runs in the + ``model_config`` JSON column (``_branched_from`` / ``_delegate_from``); + ``session_source='fork'`` alone only covers WebUI-created forks. Missing, + empty, or unparsable ``model_config`` degrades to ``(None, None)``. + """ + if not row: + return None, None + raw = row.get('model_config') + if isinstance(raw, dict): + config = raw + elif isinstance(raw, str) and raw.strip(): + try: + parsed = json.loads(raw) + except (TypeError, ValueError): + parsed = None + config = parsed if isinstance(parsed, dict) else {} + else: + config = {} + return config.get('_branched_from'), config.get('_delegate_from') + + def _is_continuation_session(parent: dict | None, child: dict | None) -> bool: """Return True when ``child`` is the next segment of the same conversation. @@ -325,17 +359,36 @@ def _is_continuation_session(parent: dict | None, child: dict | None) -> bool: by ``hermes -c`` also records a new child session; for sidebar projection it should continue the same visible conversation rather than becoming a separate child-session row. Plain parent/child links that started before the - parent's ended boundary remain child sessions. + parent's ended boundary remain child sessions. A small scheduling/write-order + overlap between the handoff boundary timestamps is tolerated (#6931). Do not collapse lineage across raw sources. A WebUI session that continues from a Telegram/CLI/etc. parent must remain visible as its own surface-owned conversation; otherwise the tip inherits the root's title/source metadata and can disappear under messaging/sidebar policies. + + Explicit branches/delegates never continue: a child whose ``model_config`` + ``_branched_from`` / ``_delegate_from`` points at this parent is a fork of + the conversation and must stay visible. Beyond the bounded tolerance window + there is no reliable continuation signal — titles are user-controlled and + non-unique, so they are never used to widen the window (#7021 re-gate). """ if not parent or not child: return False if str(child.get('session_source') or '').strip().lower() == 'fork': return False + # Real Agent branches/delegates are marked in model_config (not + # session_source, which only WebUI-created forks carry). The marker must + # reference THIS parent: compression continuations inherit the rotated + # agent's model_config verbatim, so presence alone (a delegate's + # continuation still carries the delegate's own ``_delegate_from``) would + # misclassify real continuations. + child_branched_from, child_delegate_from = _branch_markers(child) + parent_id = parent.get('id') + if parent_id and ( + child_branched_from == parent_id or child_delegate_from == parent_id + ): + 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: @@ -349,9 +402,16 @@ def _is_continuation_session(parent: dict | None, child: dict | None) -> bool: # continuations when no boundary timestamp is available. return True try: - return float(child.get('started_at') or 0) >= float(ended_at) + child_started = float(child.get('started_at') or 0) + parent_ended = float(ended_at) except (TypeError, ValueError): return False + if child_started >= parent_ended - CONTINUATION_STARTED_AT_TOLERANCE_SECONDS: + return True + # Beyond the bounded early-side window the child is a genuine concurrent + # session — an exact title match is not evidence (titles are user-controlled + # and non-unique, so it cannot extend the tolerance). + return False def _continuation_root_id(rows_by_id: dict[str, dict], session_id: str | None) -> str | None: @@ -588,6 +648,7 @@ def read_importable_agent_session_rows( parent_expr = _optional_col('parent_session_id', session_cols) session_source_expr = _optional_col('session_source', session_cols) + model_config_expr = _optional_col('model_config', 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) @@ -709,6 +770,7 @@ def read_importable_agent_session_rows( SELECT s.id, s.title, s.model, s.message_count, s.started_at, s.source, {session_source_expr}, + {model_config_expr}, {user_id_expr}, {chat_id_expr}, {chat_type_expr}, @@ -791,6 +853,11 @@ def read_importable_agent_session_rows( params, ) projected = _project_agent_session_rows([dict(row) for row in cur.fetchall()]) + # model_config is selected only so the continuation classifier can + # read the real _branched_from/_delegate_from markers; it is internal + # agent state and must not leak into the /api/sessions response. + for row in projected: + row.pop('model_config', None) projected = [_with_normalized_source(row) for row in projected] projected = [row for row in projected if is_cli_session_row_visible(row)] if limit is None: @@ -893,6 +960,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops: source_expr = _optional_col('source', session_cols) session_source_expr = _optional_col('session_source', session_cols) + model_config_expr = _optional_col('model_config', 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) @@ -907,6 +975,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops: SELECT s.id, {source_expr}, {session_source_expr}, + {model_config_expr}, {title_expr}, {started_expr}, {parent_expr}, @@ -953,6 +1022,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops: SELECT s.id, {source_expr}, {session_source_expr}, + {model_config_expr}, {title_expr}, {started_expr}, {parent_expr}, @@ -1030,6 +1100,7 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st return {} session_source_expr = _optional_col('session_source', session_cols) source_expr = _optional_col('source', session_cols) + model_config_expr = _optional_col('model_config', 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 @@ -1066,7 +1137,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, {source_expr}, {session_source_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason, {message_count_expr} + SELECT s.id, {source_expr}, {session_source_expr}, {model_config_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}) """, @@ -1095,7 +1166,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, {source_expr}, {session_source_expr}, s.title, s.started_at, s.parent_session_id, s.ended_at, s.end_reason, {message_count_expr} + SELECT s.id, {source_expr}, {session_source_expr}, {model_config_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}) """, diff --git a/api/models.py b/api/models.py index d3bcca28..8f333353 100644 --- a/api/models.py +++ b/api/models.py @@ -8581,9 +8581,26 @@ def get_state_db_session_messages( cur.execute("PRAGMA table_info(sessions)") session_cols = {str(row['name']) for row in cur.fetchall()} if {'parent_session_id', 'end_reason', 'started_at', 'source'}.issubset(session_cols): + # session_source/model_config carry the fork/delegate boundary + # identity consumed by _is_continuation_session (#6931/#7021). + # Select them when present so the branch-marker guard applies + # to the stitch path too; older schemas degrade to NULL. + identity_cols = [] + if 'session_source' in session_cols: + identity_cols.append('session_source') + else: + identity_cols.append('NULL AS session_source') + if 'model_config' in session_cols: + identity_cols.append('model_config') + else: + identity_cols.append('NULL AS model_config') + lineage_select = ( + "id, source, started_at, parent_session_id, ended_at, end_reason" + + ", " + ", ".join(identity_cols) + ) cur.execute( - """ - SELECT id, source, started_at, parent_session_id, ended_at, end_reason + f""" + SELECT {lineage_select} FROM sessions WHERE id = ? """, @@ -8601,8 +8618,8 @@ def get_state_db_session_messages( if not parent_id or parent_id in seen: break cur.execute( - """ - SELECT id, source, started_at, parent_session_id, ended_at, end_reason + f""" + SELECT {lineage_select} FROM sessions WHERE id = ? """, diff --git a/tests/test_issue5455_listing_readonly_connection.py b/tests/test_issue5455_listing_readonly_connection.py index fc2cb263..9502fae1 100644 --- a/tests/test_issue5455_listing_readonly_connection.py +++ b/tests/test_issue5455_listing_readonly_connection.py @@ -74,6 +74,77 @@ def test_listing_opens_read_only_and_returns_rows(tmp_path, monkeypatch): assert "mode=ro" in calls[0]["target"] +def test_listing_keeps_model_config_branch_visible_and_never_leaks_it(tmp_path, monkeypatch): + """#7021 re-gate: a CLI/agent branch whose _branched_from marker lives in + model_config must stay visible even when it starts inside the 2s + compression tolerance window, and model_config must not leak into the + returned rows.""" + db = tmp_path / "state.db" + conn = sqlite3.connect(str(db)) + conn.execute( + """ + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, title TEXT, model TEXT, message_count INTEGER, + started_at REAL, source TEXT, session_source TEXT, + parent_session_id TEXT, ended_at REAL, end_reason TEXT, + model_config TEXT + ) + """ + ) + conn.execute( + """ + INSERT INTO sessions + (id, title, model, message_count, started_at, source, session_source, + parent_session_id, ended_at, end_reason, model_config) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "cli-parent", "Parent conversation", "gpt", 0, 1000.0, "cli", "cli", + None, 1100.0, "compression", None, + ), + ) + # An explicit Agent branch starting 1.5s BEFORE the parent's compression + # ended_at. No session_source — the fork identity lives in model_config. + conn.execute( + """ + INSERT INTO sessions + (id, title, model, message_count, started_at, source, session_source, + parent_session_id, ended_at, end_reason, model_config) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "cli-branch", "Branch conversation", "gpt", 0, 1098.5, "cli", "cli", + "cli-parent", None, None, '{"_branched_from": "cli-parent"}', + ), + ) + conn.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, timestamp REAL)" + ) + conn.executemany( + "INSERT INTO messages (session_id, role, timestamp) VALUES (?,?,?)", + [ + ("cli-parent", "user", 1001.0), + ("cli-parent", "assistant", 1002.0), + ("cli-branch", "user", 1099.0), + ("cli-branch", "assistant", 1099.5), + ], + ) + conn.commit() + conn.close() + _record_connects(monkeypatch) + + out = read_importable_agent_session_rows(db, exclude_sources=None) + out_ids = {r["id"] for r in out} + + # The branch survives the tolerance window: it is NOT collapsed into the + # parent lineage. + assert "cli-branch" in out_ids + assert "cli-parent" in out_ids + # model_config is internal agent state — never exposed on returned rows. + for row in out: + assert "model_config" not in row + + def test_listing_read_only_uri_encodes_special_path_chars(tmp_path, monkeypatch): db_dir = tmp_path / "state dir #1" db_dir.mkdir() diff --git a/tests/test_session_lineage_metadata_api.py b/tests/test_session_lineage_metadata_api.py index 22635e57..6e35bc1b 100644 --- a/tests/test_session_lineage_metadata_api.py +++ b/tests/test_session_lineage_metadata_api.py @@ -1,5 +1,6 @@ """Regression tests for /api/sessions lineage metadata used by sidebar collapse.""" +import json import sqlite3 import time @@ -51,7 +52,8 @@ def _ensure_state_db(path): message_count INTEGER DEFAULT 0, parent_session_id TEXT, ended_at REAL, - end_reason TEXT + end_reason TEXT, + model_config TEXT ); """ ) @@ -62,6 +64,7 @@ def _ensure_messages_table(conn): conn.execute( """ CREATE TABLE messages ( + id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, @@ -79,14 +82,14 @@ def _insert_state_message(conn, sid, *, role, content, timestamp): conn.commit() -def _insert_state_row(conn, sid, *, title=None, parent=None, ended_at=None, end_reason=None, started_at=None, source='webui', session_source=None): +def _insert_state_row(conn, sid, *, title=None, parent=None, ended_at=None, end_reason=None, started_at=None, source='webui', session_source=None, model_config=None): conn.execute( """ INSERT INTO sessions - (id, source, session_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, model_config) + VALUES (?, ?, ?, ?, 'openai/gpt-5', ?, 2, ?, ?, ?, ?) """, - (sid, source, session_source, title or sid, started_at or time.time(), parent, ended_at, end_reason), + (sid, source, session_source, title or sid, started_at or time.time(), parent, ended_at, end_reason, model_config), ) conn.commit() @@ -693,3 +696,455 @@ def test_sessions_route_preserves_visible_child_lineage_when_archived_parent_fil ] finally: conn.close() + + +def test_compression_continuation_tolerates_sub_second_timestamp_overlap(_isolate): + """#6931: a compression continuation recorded a few ms BEFORE the parent's + ended_at (write-order race) is still one lineage in /api/sessions.""" + conn = _ensure_state_db(_isolate) + t0 = time.time() - 100 + try: + _save_webui_session("lineage_race_root", title="Shared conversation", updated_at=t0) + _save_webui_session("lineage_race_tip", title="Shared conversation", updated_at=t0 + 10) + _insert_state_row( + conn, + "lineage_race_root", + started_at=t0, + ended_at=t0 + 5, + end_reason="compression", + ) + # child.started_at lands 0.1s BEFORE parent.ended_at — the #6931 race. + _insert_state_row( + conn, + "lineage_race_tip", + parent="lineage_race_root", + started_at=t0 + 5 - 0.1, + ) + + rows = {row["session_id"]: row for row in all_sessions()} + + tip = rows["lineage_race_tip"] + assert tip.get("parent_session_id") == "lineage_race_root" + assert tip.get("_lineage_root_id") == "lineage_race_root" + assert tip.get("_lineage_tip_id") == "lineage_race_tip" + assert tip.get("_compression_segment_count") == 2 + assert tip.get("relationship_type") != "child_session" + finally: + conn.close() + + +def test_materially_overlapping_compression_child_stays_child_session(_isolate): + """#6931: a child that started long before the compression parent ended + (beyond tolerance, no matching title) must remain a separate child.""" + conn = _ensure_state_db(_isolate) + t0 = time.time() - 100 + try: + _save_webui_session("lineage_overlap_root", title="Root title", updated_at=t0) + _save_webui_session("lineage_overlap_child", title="Child title", updated_at=t0 + 10) + _insert_state_row( + conn, + "lineage_overlap_root", + started_at=t0, + ended_at=t0 + 60, + end_reason="compression", + ) + # child started 30s before the parent ended — a genuine concurrent child. + _insert_state_row( + conn, + "lineage_overlap_child", + parent="lineage_overlap_root", + started_at=t0 + 30, + ) + + rows = {row["session_id"]: row for row in all_sessions()} + + child = rows["lineage_overlap_child"] + assert child.get("relationship_type") == "child_session" + assert child.get("parent_session_id") == "lineage_overlap_root" + assert "_lineage_root_id" not in child + assert "_compression_segment_count" not in child + finally: + conn.close() + + +def test_same_title_independent_child_outside_tolerance_stays_visible(_isolate): + """#7021 re-gate: a same-title independent child started well outside the + tolerance window must NOT collapse on the title match — titles are + user-controlled and non-unique, so the only continuation evidence is the + bounded early-side timestamp window.""" + conn = _ensure_state_db(_isolate) + t0 = time.time() - 100 + try: + _save_webui_session("lineage_title_root", title="Shared conversation", updated_at=t0) + _save_webui_session("lineage_title_tip", title="Shared conversation", updated_at=t0 + 10) + _insert_state_row( + conn, + "lineage_title_root", + title="Shared conversation", + started_at=t0, + ended_at=t0 + 60, + end_reason="compression", + ) + # Same title, but started 50s before the parent ended — far outside any + # handoff race window. Must remain a visible child session. + _insert_state_row( + conn, + "lineage_title_tip", + title="Shared conversation", + parent="lineage_title_root", + started_at=t0 + 10, + ) + + rows = {row["session_id"]: row for row in all_sessions()} + + tip = rows["lineage_title_tip"] + assert tip.get("relationship_type") == "child_session" + assert tip.get("parent_session_id") == "lineage_title_root" + assert "_lineage_root_id" not in tip + assert "_compression_segment_count" not in tip + finally: + conn.close() + + +def test_model_config_branched_from_child_stays_visible_within_tolerance(_isolate): + """#7021 re-gate: an explicit Agent branch is marked in + model_config._branched_from (not session_source). Even when it starts + inside the 2s tolerance window of the parent's compression, the real + marker must keep it visible instead of collapsing it into the lineage.""" + conn = _ensure_state_db(_isolate) + t0 = time.time() - 100 + try: + _save_webui_session("lineage_branch_root", title="Shared conversation", updated_at=t0) + _save_webui_session("lineage_branch_tip", title="Shared conversation", updated_at=t0 + 10) + _insert_state_row( + conn, + "lineage_branch_root", + started_at=t0, + ended_at=t0 + 5, + end_reason="compression", + ) + # The reviewer's adversarial probe: an Agent branch starting 1.5s + # BEFORE the parent's compression ended_at. No session_source — the + # fork identity lives in model_config._branched_from. + _insert_state_row( + conn, + "lineage_branch_tip", + parent="lineage_branch_root", + started_at=t0 + 5 - 1.5, + model_config=json.dumps({"_branched_from": "lineage_branch_root"}), + ) + + rows = {row["session_id"]: row for row in all_sessions()} + + tip = rows["lineage_branch_tip"] + assert tip.get("relationship_type") == "child_session" + assert tip.get("parent_session_id") == "lineage_branch_root" + assert "_lineage_root_id" not in tip + assert "_compression_segment_count" not in tip + finally: + conn.close() + + +def test_model_config_delegate_from_child_stays_visible_within_tolerance(_isolate): + """#7021 re-gate: delegate/subagent runs are marked in + model_config._delegate_from. A delegate child starting inside the + tolerance window of the parent's compression must stay visible.""" + conn = _ensure_state_db(_isolate) + t0 = time.time() - 100 + try: + _save_webui_session("lineage_delegate_root", title="Shared conversation", updated_at=t0) + _save_webui_session("lineage_delegate_tip", title="Shared conversation", updated_at=t0 + 10) + _insert_state_row( + conn, + "lineage_delegate_root", + started_at=t0, + ended_at=t0 + 5, + end_reason="compression", + ) + _insert_state_row( + conn, + "lineage_delegate_tip", + parent="lineage_delegate_root", + started_at=t0 + 5 - 1.0, + model_config=json.dumps({"_delegate_from": "lineage_delegate_root"}), + ) + + rows = {row["session_id"]: row for row in all_sessions()} + + tip = rows["lineage_delegate_tip"] + assert tip.get("relationship_type") == "child_session" + assert tip.get("parent_session_id") == "lineage_delegate_root" + assert "_lineage_root_id" not in tip + assert "_compression_segment_count" not in tip + finally: + conn.close() + + +def test_continuation_classification_timestamp_tolerance_and_guards(): + """#6931/#7021 focused unit coverage of _is_continuation_session: bounded + tolerance, model_config branch markers, and preserved fork/cross-source/ + end_reason guards.""" + from api.agent_sessions import _is_continuation_session + + def make_parent(**over): + row = { + 'id': 'parent-1', + 'source': 'webui', + 'end_reason': 'compression', + 'ended_at': 1000.0, + 'title': 'Shared conversation', + } + row.update(over) + return row + + def make_child(**over): + row = { + 'id': 'child-1', + 'source': 'webui', + 'started_at': 1000.05, + 'title': 'Shared conversation', + } + row.update(over) + return row + + # Normal non-overlapping ordering: continuation. + assert _is_continuation_session(make_parent(), make_child()) + # Sub-second overlap (observed -0.06..-0.07s in #6931): continuation. + assert _is_continuation_session(make_parent(), make_child(started_at=999.95)) + # Overlap inside the 2s tolerance: continuation. + assert _is_continuation_session(make_parent(), make_child(started_at=998.5)) + # Overlap beyond tolerance: separate child, even with an exact title match + # (titles are user-controlled and non-unique — no title fallback). + assert not _is_continuation_session(make_parent(), make_child(started_at=950.0)) + assert not _is_continuation_session( + make_parent(), make_child(started_at=950.0, title='Another conversation') + ) + # model_config._branched_from pointing at the parent: never a + # continuation, regardless of timing. + assert not _is_continuation_session( + make_parent(), + make_child( + started_at=999.95, + model_config=json.dumps({'_branched_from': 'parent-1'}), + ), + ) + assert not _is_continuation_session( + make_parent(), + make_child( + started_at=998.5, + model_config={'_branched_from': 'parent-1'}, + ), + ) + # model_config._delegate_from pointing at the parent: never a + # continuation, regardless of timing. + assert not _is_continuation_session( + make_parent(), + make_child( + started_at=999.95, + model_config=json.dumps({'_delegate_from': 'parent-1'}), + ), + ) + # A marker pointing at a DIFFERENT session does not disqualify: compression + # continuations inherit the rotated agent's model_config verbatim, so a + # delegate's continuation still carries the delegate's own marker. + assert _is_continuation_session( + make_parent(), + make_child( + started_at=999.95, + model_config=json.dumps({'_delegate_from': 'some-other-session'}), + ), + ) + # Unparsable model_config degrades to no markers (no crash, no match). + assert _is_continuation_session( + make_parent(), + make_child(started_at=999.95, model_config='not-json'), + ) + assert not _is_continuation_session( + make_parent(), + make_child(started_at=950.0, model_config='not-json'), + ) + # Fork guard holds regardless of timing. + assert not _is_continuation_session( + make_parent(), make_child(started_at=999.95, session_source='fork') + ) + # Cross-source guard holds regardless of timing. + assert not _is_continuation_session( + make_parent(), make_child(started_at=999.95, source='telegram') + ) + # Non-compression/cli_close parents never continue. + assert not _is_continuation_session( + make_parent(end_reason='user_stop'), make_child(started_at=999.95) + ) + # Missing/unparsable boundary timestamps degrade to False (no crash). + assert not _is_continuation_session(make_parent(ended_at='not-a-number'), make_child()) + assert not _is_continuation_session(make_parent(), make_child(started_at='not-a-number')) + + +def test_state_db_stitch_keeps_branched_child_out_of_parent_transcript(_isolate): + """#7021 r2: the open/import transcript stitcher (get_state_db_session_messages + with stitch_continuations=True) must apply the same model_config branch-marker + guard as the listing classifier. A child whose model_config._branched_from / + _delegate_from points at the parent must NOT have its messages stitched into + the parent transcript even when it starts inside the tolerance window. + """ + conn = _ensure_state_db(_isolate) + _ensure_messages_table(conn) + t0 = time.time() - 100 + try: + # Compression parent whose child starts INSIDE the 2s tolerance window. + _insert_state_row( + conn, + "stitch_branch_parent", + started_at=t0, + ended_at=t0 + 5, + end_reason="compression", + ) + _insert_state_message( + conn, "stitch_branch_parent", role="user", content="parent turn", timestamp=t0 + 1 + ) + # Explicit Agent branch: fork identity lives in model_config, not + # session_source. Starting 1.5s before the parent's ended_at — inside + # the tolerance — it must still stay out of the parent's transcript. + _insert_state_row( + conn, + "stitch_branch_child", + parent="stitch_branch_parent", + started_at=t0 + 5 - 1.5, + model_config=json.dumps({"_branched_from": "stitch_branch_parent"}), + ) + _insert_state_message( + conn, "stitch_branch_child", role="user", content="branch turn", timestamp=t0 + 6 + ) + + msgs = models.get_state_db_session_messages( + "stitch_branch_child", stitch_continuations=True + ) + contents = [m["content"] for m in msgs] + assert "branch turn" in contents + assert "parent turn" not in contents + + # Control: a genuine compression continuation (no branch marker) starting + # inside the same window IS stitched into the parent transcript. + _insert_state_row( + conn, + "stitch_continuation_child", + parent="stitch_branch_parent", + started_at=t0 + 5 - 0.5, + ) + _insert_state_message( + conn, + "stitch_continuation_child", + role="user", + content="continuation turn", + timestamp=t0 + 7, + ) + msgs = models.get_state_db_session_messages( + "stitch_continuation_child", stitch_continuations=True + ) + contents = [m["content"] for m in msgs] + assert "continuation turn" in contents + assert "parent turn" in contents + finally: + conn.close() + + +def test_state_db_stitch_keeps_delegate_child_out_of_parent_transcript(_isolate): + """#7021 r2: same guard for model_config._delegate_from (subagent runs) in + the open/import transcript stitcher.""" + conn = _ensure_state_db(_isolate) + _ensure_messages_table(conn) + t0 = time.time() - 100 + try: + _insert_state_row( + conn, + "stitch_delegate_parent", + started_at=t0, + ended_at=t0 + 5, + end_reason="compression", + ) + _insert_state_message( + conn, "stitch_delegate_parent", role="user", content="parent turn", timestamp=t0 + 1 + ) + _insert_state_row( + conn, + "stitch_delegate_child", + parent="stitch_delegate_parent", + started_at=t0 + 5 - 1.0, + model_config=json.dumps({"_delegate_from": "stitch_delegate_parent"}), + ) + _insert_state_message( + conn, "stitch_delegate_child", role="user", content="delegate turn", timestamp=t0 + 6 + ) + + msgs = models.get_state_db_session_messages( + "stitch_delegate_child", stitch_continuations=True + ) + contents = [m["content"] for m in msgs] + assert "delegate turn" in contents + assert "parent turn" not in contents + finally: + conn.close() + + +def test_state_db_stitch_old_schema_without_identity_columns_still_stitches(_isolate): + """#7021 r2: older state.db schemas lacking session_source/model_config degrade + to NULL and keep the pre-fix behavior — a genuine compression continuation + within the tolerance is still stitched, without crashing.""" + conn = sqlite3.connect(str(_isolate)) + conn.executescript( + """ + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + source TEXT, + title TEXT, + started_at REAL NOT NULL, + message_count INTEGER DEFAULT 0, + parent_session_id TEXT, + ended_at REAL, + end_reason TEXT + ); + """ + ) + _ensure_messages_table(conn) + t0 = time.time() - 100 + try: + conn.execute( + """ + INSERT INTO sessions + (id, source, title, started_at, message_count, parent_session_id, ended_at, end_reason) + VALUES (?, ?, ?, ?, 2, ?, ?, ?) + """, + ( + "stitch_old_parent", + "webui", + "stitch_old_parent", + t0, + None, + t0 + 5, + "compression", + ), + ) + conn.execute( + """ + INSERT INTO sessions + (id, source, title, started_at, message_count, parent_session_id, ended_at, end_reason) + VALUES (?, ?, ?, ?, 2, ?, NULL, NULL) + """, + ("stitch_old_child", "webui", "stitch_old_child", t0 + 4.5, "stitch_old_parent"), + ) + conn.commit() + _insert_state_message( + conn, "stitch_old_parent", role="user", content="parent turn", timestamp=t0 + 1 + ) + _insert_state_message( + conn, "stitch_old_child", role="user", content="old child turn", timestamp=t0 + 6 + ) + + msgs = models.get_state_db_session_messages( + "stitch_old_child", stitch_continuations=True + ) + contents = [m["content"] for m in msgs] + assert "old child turn" in contents + assert "parent turn" in contents + finally: + conn.close()