jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge PR #7021 (contributor head 2e12075400)

Browse Source
This commit is contained in:
nesquena-hermes committed 2026-09-22 21:14:09 +00:00
2 parents 47f8e028cd 2e12075400
commit 0a152c548d
4 files changed
+627 -13
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/agent_sessions.py
+75 -4
View File
@@ -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})
""",
api/models.py
+21 -4
View File
@@ -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 = ?
""",
tests/test_issue5455_listing_readonly_connection.py
+71
View File
@@ -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()
tests/test_session_lineage_metadata_api.py
+460 -5
View File
@@ -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()