test(sessions): cover branch-seed and compression timestamp round-trips

Follow-up to the salvaged #28840 change: forward original timestamps in
_persist_branch_seed (TUI first-turn branch persist), add behavioral
round-trip tests for branch copy and compression-style replace, TUI
protocol tests asserting session.branch and _persist_branch_seed forward
timestamps, and register the contributor email mapping.
This commit is contained in:
Teknium
2026-07-22 04:43:37 -07:00
parent 9d73006ade
commit 961b832c11
4 changed files with 164 additions and 1 deletions
@@ -0,0 +1 @@
yitang
+59
View File
@@ -1809,6 +1809,65 @@ class TestTimestampPreservation:
assert child_raw == timestamps
assert parent_raw == child_raw
def test_branch_copy_roundtrip_preserves_timestamps(self, db):
"""End-to-end branch copy: load the parent transcript via
get_messages_as_conversation (the dict shape the CLI/gateway/TUI
branch loops iterate) and re-append into a child forwarding
``msg.get("timestamp")`` — the copies must keep the originals
instead of being restamped with time.time() (#28841).
"""
timestamps = [1_600_000_000.0, 1_600_000_060.0, 1_600_000_120.0]
db.create_session(session_id="parent", source="cli")
for i, ts in enumerate(timestamps):
db.append_message(
"parent",
role="user" if i % 2 == 0 else "assistant",
content=f"msg-{i}",
timestamp=ts,
)
history = db.get_messages_as_conversation("parent")
assert [m.get("timestamp") for m in history] == timestamps
db.create_session(session_id="child", source="cli",
parent_session_id="parent")
# Mirrors the branch copy loops in gateway/slash_commands.py,
# hermes_cli/cli_commands_mixin.py and tui_gateway/server.py.
for msg in history:
db.append_message(
"child",
role=msg.get("role", "user"),
content=msg.get("content"),
timestamp=msg.get("timestamp"),
)
assert self._raw_timestamps(db, "child") == timestamps
def test_compression_replace_roundtrip_preserves_timestamps(self, db):
"""Compression-style rewrite: replace_messages with dicts loaded from
get_messages_as_conversation must keep the surviving messages'
original timestamps (#28841)."""
timestamps = [1_500_000_000.0, 1_500_000_100.0, 1_500_000_200.0]
db.create_session(session_id="s1", source="cli")
for i, ts in enumerate(timestamps):
db.append_message(
"s1",
role="user" if i % 2 == 0 else "assistant",
content=f"msg-{i}",
timestamp=ts,
)
history = db.get_messages_as_conversation("s1")
# Simulate a compression that keeps the last two turns verbatim and
# prepends a fresh summary message (no timestamp — falls back to now).
compressed = [{"role": "user", "content": "[summary]"}] + history[-2:]
db.replace_messages("s1", compressed)
raw = self._raw_timestamps(db, "s1")
assert len(raw) == 3
assert raw[1:] == timestamps[-2:]
assert raw[0] > timestamps[-1] # summary stamped with a current time
# =========================================================================
# FTS5 search
+95
View File
@@ -1289,6 +1289,101 @@ def test_session_branch_persists_branched_from_marker(server, monkeypatch):
assert kwargs["model_config"] == {"_branched_from": parent_key}
def test_session_branch_forwards_original_timestamps(server, monkeypatch):
"""TUI /branch must copy the parent's messages WITH their original
timestamps — append_message otherwise stamps time.time() at INSERT and
the branch's whole history silently appears authored "now" (#28841).
"""
append_calls = []
class _DB:
def get_session_title(self, _key):
return "parent-title"
def get_next_title_in_lineage(self, base):
return f"{base} 2"
def create_session(self, new_key, **kwargs):
return new_key
def append_message(self, **kwargs):
append_calls.append(kwargs)
return None
def set_session_title(self, _key, _title):
return None
monkeypatch.setattr(server, "_get_db", lambda: _DB())
monkeypatch.setattr(server, "_resolve_model", lambda: "test/model")
monkeypatch.setattr(server, "_new_session_key", lambda: "20260101_000001_child0")
monkeypatch.setattr(
server,
"_make_agent",
lambda _sid, key, session_id=None, session_db=None, **_kwargs: types.SimpleNamespace(
model="test/model", session_id=session_id or key
),
)
monkeypatch.setattr(server, "_init_session", lambda *_a, **_k: None)
monkeypatch.setattr(server, "_set_session_context", lambda *_a, **_k: [])
monkeypatch.setattr(server, "_clear_session_context", lambda *_a, **_k: None)
monkeypatch.setattr(server, "_session_cwd", lambda _s: "/tmp/branch-cwd")
original_ts = [1_700_000_000.0, 1_700_000_020.0]
parent_sid = "parent02"
server._sessions[parent_sid] = {
"session_key": "20260101_000000_parent",
"history": [
{"role": "user", "content": "hello", "timestamp": original_ts[0]},
{"role": "assistant", "content": "hi!", "timestamp": original_ts[1]},
],
"history_lock": threading.Lock(),
"cols": 80,
}
resp = server.handle_request(
{"id": "b2", "method": "session.branch", "params": {"session_id": parent_sid}}
)
assert "error" not in resp, resp
assert len(append_calls) == 2
assert [c.get("timestamp") for c in append_calls] == original_ts
def test_persist_branch_seed_forwards_original_timestamps(server, monkeypatch):
"""First-turn branch seed persist must carry each copied message's
original timestamp through to append_message (#28841)."""
import contextlib
append_calls = []
class _DB:
def append_message(self, **kwargs):
append_calls.append(kwargs)
return None
@contextlib.contextmanager
def _fake_session_db(_session):
yield _DB()
monkeypatch.setattr(server, "_session_db", _fake_session_db)
original_ts = [100.0, 200.0]
session = {
"session_key": "20260101_000002_seed00",
"parent_session_id": "20260101_000000_parent",
"history": [
{"role": "user", "content": "a", "timestamp": original_ts[0]},
{"role": "assistant", "content": "b", "timestamp": original_ts[1]},
],
"history_lock": threading.Lock(),
}
server._persist_branch_seed(session)
assert session.get("_branch_seed_persisted") is True
assert [c.get("timestamp") for c in append_calls] == original_ts
def test_make_agent_accepts_list_system_prompt(server, monkeypatch):
captured = {}
+9 -1
View File
@@ -2067,7 +2067,15 @@ def _persist_branch_seed(session: dict) -> None:
return
try:
for msg in seed:
db.append_message(session_id=key, role=msg.get("role", "user"), content=msg.get("content"))
db.append_message(
session_id=key,
role=msg.get("role", "user"),
content=msg.get("content"),
# Preserve the parent's original message timestamps —
# append_message would otherwise stamp time.time() and the
# branch's copied history would all appear authored "now".
timestamp=msg.get("timestamp"),
)
session["_branch_seed_persisted"] = True
except Exception:
logger.debug("branch seed persist failed", exc_info=True)