diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index b16765ea9b40..25e73f48fa21 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -543,7 +543,33 @@ def _release_lock() -> None: todo_snapshot = agent._todo_store.format_for_injection() if todo_snapshot: - compressed.append({"role": "user", "content": todo_snapshot}) + # Inject the preserved task list so the model keeps its plan across the + # compaction boundary. When the compressed transcript already ends with + # a user message, fold the snapshot into that message (blank-line + # separated) instead of appending a second standalone user message — + # consecutive user/user turns are a content-ordering violation some + # providers reject, and a merged turn keeps the latest user content + # carrying both the original text and the task list. Only plain string + # content is mergeable; an empty transcript, a non-user tail, or + # structured (list) content falls back to the append path so image/tool + # parts aren't corrupted. + _merged_into_tail = False + if compressed: + _tail = compressed[-1] + if ( + isinstance(_tail, dict) + and _tail.get("role") == "user" + and isinstance(_tail.get("content"), str) + ): + _tail_content = _tail["content"] + _tail["content"] = ( + f"{_tail_content}\n\n{todo_snapshot}" + if _tail_content + else todo_snapshot + ) + _merged_into_tail = True + if not _merged_into_tail: + compressed.append({"role": "user", "content": todo_snapshot}) agent._invalidate_system_prompt() new_system_prompt = agent._build_system_prompt(system_message) diff --git a/cron/scheduler.py b/cron/scheduler.py index 410e9d7dc777..260b303bb26f 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1983,9 +1983,13 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: # stdout to telegram" watchdog pattern. The agent path is skipped # entirely: no AIAgent, no prompt, no tool loop, no token spend. # - # We check this BEFORE importing run_agent / constructing SessionDB so - # a pure-script tick never pays for the agent machinery it isn't going - # to use. Keep this block self-contained. + # We check this BEFORE importing run_agent so a pure-script tick never + # pays for the agent machinery it isn't going to use. The run is still + # recorded as a ``cron_{job_id}_{ts}`` session (a cheap sqlite write, + # same shape as agent runs) so it appears in the run-history endpoint + # (``/api/cron/jobs/{id}/runs``) and the desktop GUI — without it, + # no_agent runs are invisible after a manual trigger (#44080). Keep + # this block self-contained. # # Semantics: # - script stdout (trimmed) → delivered verbatim as the final message @@ -2001,6 +2005,63 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: logger.error("Job '%s': %s", job_id, err) return False, "", "", err + # Record this run as a session BEFORE executing the script: the run + # session's open/ended state is what the desktop run-history endpoint + # uses for its "running" indicator (``is_active``), and the session + # row is the run record itself. Best-effort — a missing/broken state + # store degrades to the old no-record behaviour, never blocks the run. + _session_db = None + _run_session_id = None + try: + from hermes_state import SessionDB + _session_db = SessionDB() + _run_session_id = f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}" + _session_db.create_session(_run_session_id, source="cron") + # First user message becomes the run's preview in run history. + _session_db.append_message( + _run_session_id, "user", f"no_agent script: {script_path}" + ) + except (Exception, KeyboardInterrupt) as e: + logger.debug( + "Job '%s': SQLite session store not available for no_agent run: %s", + job_id, e, + ) + _session_db = None + + def _record_run(run_doc: str, success: bool = True) -> None: + """Persist the outcome and close out this run's session record.""" + if not _session_db: + return + try: + _session_db.append_message(_run_session_id, "assistant", run_doc) + except (Exception, KeyboardInterrupt) as e: + logger.debug( + "Job '%s': failed to record no_agent run output: %s", job_id, e + ) + # Same titling scheme as the agent path so sidebars/history show a + # meaningful label; the run-time suffix keeps it unique against + # the sessions.title index across runs. + try: + _title_base = " ".join(job_name.split())[:60].strip() or f"cron {job_id}" + _session_db.set_session_title( + _run_session_id, + f"{_title_base} · {_hermes_now().strftime('%b %d %H:%M')}", + ) + except (Exception, KeyboardInterrupt) as e: + logger.debug("Job '%s': failed to set cron session title: %s", job_id, e) + try: + _session_db.end_session( + _run_session_id, "cron_complete" if success else "cron_failed" + ) + except (Exception, KeyboardInterrupt) as e: + logger.debug("Job '%s': failed to end session: %s", job_id, e) + try: + _session_db.close() + except (Exception, KeyboardInterrupt) as e: + logger.debug( + "Job '%s': failed to close SQLite session store: %s", job_id, e + ) + # Apply workdir if configured — lets scripts use predictable relative # paths. For no_agent jobs this is just the subprocess cwd (not an # agent TERMINAL_CWD bridge). @@ -2041,6 +2102,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: f"**Status:** script failed\n\n" f"{output}\n" ) + _record_run(doc, success=False) return False, doc, alert, output # Honour the wakeAgent gate as a silent signal — `wakeAgent: false` @@ -2056,6 +2118,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: f"**Mode:** no_agent (script)\n" f"**Status:** silent (wakeAgent=false)\n" ) + _record_run(silent_doc) return True, silent_doc, SILENT_MARKER, None if not output.strip(): @@ -2067,6 +2130,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: f"**Mode:** no_agent (script)\n" f"**Status:** silent (empty output)\n" ) + _record_run(silent_doc) return True, silent_doc, SILENT_MARKER, None doc = ( @@ -2077,6 +2141,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: f"---\n\n" f"{output}\n" ) + _record_run(doc) return True, doc, output, None # --------------------------------------------------------------- diff --git a/hermes_state.py b/hermes_state.py index a7938f7167f4..05081c351a2b 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -116,6 +116,75 @@ def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]: conn.execute(f"DELETE FROM sessions WHERE id IN ({ph})", ids) return ids + +# A child is a *compression continuation* of its parent (not a branch, +# subagent, or tool-spawned run) when the parent ended via 'compression' and +# the child carries no _branched_from / _delegate_from marker and is not a +# tool-source row. This is the same edge definition get_compression_tip and +# list_sessions_rich use to project roots forward to their tips, so a delete +# that walks this graph can never leave a row that resurfaces on refresh. +_COMPRESSION_CONTINUATION_PREDICATE = ( + "parent.end_reason = 'compression'" + " AND json_extract(COALESCE(child.model_config, '{}'), '$._branched_from') IS NULL" + " AND json_extract(COALESCE(child.model_config, '{}'), '$._delegate_from') IS NULL" + " AND COALESCE(child.source, '') != 'tool'" +) + + +def _compression_lineage_ids(conn, session_ids: List[str]) -> List[str]: + """All ids in the compression-continuation lineages touching *session_ids*. + + Walks both directions along the compression-continuation edge (parent ended + ``end_reason='compression'``, child is not a branch / delegate / tool row), + matching the projection that ``get_compression_tip`` and + ``list_sessions_rich`` apply. Deleting any member of a lineage must remove + the whole lineage, or the orphaned root (or a sibling continuation) + resurfaces in session pickers on the next refresh. + + The seeds themselves are always included. Returns ids in deterministic + order; callers orphan children before deleting the returned rows, so the + order is not used for FK safety. + """ + seeds = {sid for sid in session_ids if sid} + if not seeds: + return [] + + found: set[str] = set(seeds) + frontier: set[str] = set(seeds) + + # Walk up (ancestors) and down (descendants) in lockstep. A seed may sit in + # the middle of a chain, so both directions are needed to cover the whole + # lineage. Bounded by the reachable set, which is naturally finite. + while frontier: + ph = ",".join("?" * len(frontier)) + params = list(frontier) + rows = conn.execute( + f""" + SELECT child.id AS id + FROM sessions parent + JOIN sessions child ON child.parent_session_id = parent.id + WHERE parent.id IN ({ph}) + AND {_COMPRESSION_CONTINUATION_PREDICATE} + UNION ALL + SELECT parent.id AS id + FROM sessions child + JOIN sessions parent ON parent.id = child.parent_session_id + WHERE child.id IN ({ph}) + AND {_COMPRESSION_CONTINUATION_PREDICATE} + """, + params + params, + ).fetchall() + next_frontier: set[str] = set() + for row in rows: + rid = row["id"] + if rid and rid not in found: + found.add(rid) + next_frontier.add(rid) + frontier = next_frontier + + return sorted(found) + + T = TypeVar("T") DEFAULT_DB_PATH = get_hermes_home() / "state.db" @@ -4255,12 +4324,25 @@ def delete_session( Delegate subagent children (``model_config._delegate_from``) are cascade-deleted with the parent so they never resurface in session - pickers as orphaned rows. Branch / compression children are orphaned - (``parent_session_id → NULL``) so they remain accessible independently. + pickers as orphaned rows. Branch children (``_branched_from``) are + orphaned (``parent_session_id → NULL``) so they remain accessible + independently. + + The entire **compression lineage** of *session_id* is deleted with it: + every ancestor and descendant linked by a compression-continuation + edge (parent ended ``end_reason='compression'``, child is not a + branch / delegate / tool row — the same edge ``get_compression_tip`` + and ``list_sessions_rich`` use to project roots forward to their tips). + Otherwise deleting the visible tip orphans the old compression root, + which then resurfaces in session pickers on the next refresh as a + "deleted session resurrected". Deleting any member removes the whole + logical conversation. + When *sessions_dir* is provided, also removes on-disk transcript files (``.json`` / ``.jsonl`` / ``request_dump_*``) for every deleted session. Returns True if the session was found and deleted. """ + removed_lineage_ids: List[str] = [] removed_delegate_ids: List[str] = [] def _do(conn): @@ -4269,22 +4351,34 @@ def _do(conn): ) if cursor.fetchone()[0] == 0: return False - removed_delegate_ids.extend(_delete_delegate_children(conn, [session_id])) - # Orphan remaining child sessions (branches, etc.) so FK is satisfied. + # Expand to the whole compression lineage so neither the old root + # nor a sibling continuation row can resurface after the tip goes. + kill_ids = _compression_lineage_ids(conn, [session_id]) + removed_lineage_ids.extend(kill_ids) + kill_ph = ",".join("?" * len(kill_ids)) + removed_delegate_ids.extend(_delete_delegate_children(conn, kill_ids)) + # Orphan remaining children (branches, etc.) of every doomed row so + # the FK stays satisfied. Compression/delegate children are in the + # kill set already; branches must survive and stay accessible. conn.execute( - "UPDATE sessions SET parent_session_id = NULL " - "WHERE parent_session_id = ?", - (session_id,), + f"UPDATE sessions SET parent_session_id = NULL " + f"WHERE parent_session_id IN ({kill_ph})", + kill_ids, + ) + conn.execute( + f"DELETE FROM messages WHERE session_id IN ({kill_ph})", kill_ids + ) + conn.execute( + f"DELETE FROM sessions WHERE id IN ({kill_ph})", kill_ids ) - conn.execute("DELETE FROM messages WHERE session_id = ?", (session_id,)) - conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,)) return True deleted = self._execute_write(_do) if deleted: for delegate_id in removed_delegate_ids: self._remove_session_files(sessions_dir, delegate_id) - self._remove_session_files(sessions_dir, session_id) + for lineage_id in removed_lineage_ids: + self._remove_session_files(sessions_dir, lineage_id) return bool(deleted) def delete_session_if_empty( @@ -4343,6 +4437,9 @@ def delete_sessions( * Unknown IDs are silently skipped (no 404) — selection state in the UI can race against another tab's delete, and we'd rather succeed-on-the-rest than fail-the-whole-batch. + * The full compression lineage of every selected row is deleted so + visible continuation tips cannot leave old roots behind that later + resurface in session lists. * Delegate subagent children (``model_config._delegate_from``) are cascade-deleted with their parent; branch children are orphaned (``parent_session_id → NULL``) so they stay accessible. @@ -4382,8 +4479,9 @@ def _do(conn): if not existing: return 0 - existing_placeholders = ",".join("?" * len(existing)) - removed_delegate_ids.extend(_delete_delegate_children(conn, existing)) + kill_ids = _compression_lineage_ids(conn, existing) + kill_placeholders = ",".join("?" * len(kill_ids)) + removed_delegate_ids.extend(_delete_delegate_children(conn, kill_ids)) # Orphan remaining children whose parent is in the kill list so the # FK constraint stays satisfied. Pin children whose parent # is itself in the kill list rather than NULL-ing parents @@ -4391,18 +4489,18 @@ def _do(conn): # exactly this. conn.execute( f"UPDATE sessions SET parent_session_id = NULL " - f"WHERE parent_session_id IN ({existing_placeholders})", - existing, + f"WHERE parent_session_id IN ({kill_placeholders})", + kill_ids, ) conn.execute( - f"DELETE FROM messages WHERE session_id IN ({existing_placeholders})", - existing, + f"DELETE FROM messages WHERE session_id IN ({kill_placeholders})", + kill_ids, ) conn.execute( - f"DELETE FROM sessions WHERE id IN ({existing_placeholders})", - existing, + f"DELETE FROM sessions WHERE id IN ({kill_placeholders})", + kill_ids, ) - removed_ids.extend(existing) + removed_ids.extend(kill_ids) return len(existing) count = self._execute_write(_do) diff --git a/tests/agent/test_compression_rotation_state.py b/tests/agent/test_compression_rotation_state.py index 83ab63e2a699..3513a3b42c0d 100644 --- a/tests/agent/test_compression_rotation_state.py +++ b/tests/agent/test_compression_rotation_state.py @@ -130,3 +130,103 @@ def test_on_session_start_receives_platform(self, tmp_path: Path): kwargs = calls[-1].kwargs assert kwargs.get("platform") == "telegram" assert kwargs.get("boundary_reason") == "compression" + + +class TestTodoSnapshotMergedNotDuplicated: + """Regression: the post-compression todo snapshot must not produce a + second standalone user message when the compressed transcript already + ends with a user message — that yielded consecutive user/user turns + (a content-ordering violation some providers reject). Instead the + snapshot is merged into the trailing user content with a blank-line + separator, so the latest user turn keeps both the original text and + the preserved task list. + """ + + def test_snapshot_merges_into_trailing_user(self, tmp_path: Path): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_TODO_MERGE" + db.create_session(parent, source="cli") + agent = _build_agent_with_db(db, parent) + # Use a compressor transcript whose tail is a user message preceded by + # a NON-user (assistant) message. The pre-existing shared mock ends + # with two user messages, which would mask whether the snapshot is the + # source of any consecutive user/user — here the ONLY way a + # user/user pair can appear is if the snapshot appends a second + # standalone user message next to the tail. + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "earlier assistant turn"}, + {"role": "user", "content": "tail"}, + ] + baseline_len = len(agent.context_compressor.compress.return_value) + + # Populate the todo store the way a real session would: an active + # (pending) item, so format_for_injection() yields non-empty text. + agent._todo_store.write( + [{"id": "1", "content": "ship the migration", "status": "pending"}], + merge=False, + ) + snapshot = agent._todo_store.format_for_injection() + assert snapshot # sanity: the snapshot is non-empty + + compressed, _ = agent._compress_context(_msgs(), "sys", approx_tokens=120_000) + child = agent.session_id + assert child != parent # rotation happened + + # The RETURNED compressed messages: the snapshot must NOT append a + # second standalone user message (length unchanged) — it merges into + # the trailing user content, which then carries both the original + # tail text and the snapshot. No consecutive user/user is introduced + # by the snapshot. In rotation mode the gateway caller persists this + # returned transcript into the continuation session. + assert len(compressed) == baseline_len, ( + "todo snapshot appended a standalone user message instead of merging" + ) + last = compressed[-1] + assert last["role"] == "user" + assert "tail" in last["content"] + assert snapshot in last["content"] + for a, b in zip(compressed, compressed[1:]): + assert not (a["role"] == "user" and b["role"] == "user"), ( + "consecutive user/user messages in compressed transcript " + "caused by the todo snapshot" + ) + + def test_snapshot_merge_is_persisted_in_place(self, tmp_path: Path): + db = SessionDB(db_path=tmp_path / "state.db") + session = "PARENT_TODO_IN_PLACE" + db.create_session(session, source="cli") + agent = _build_agent_with_db(db, session) + agent.compression_in_place = True + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "earlier assistant turn"}, + {"role": "user", "content": "tail"}, + ] + + agent._todo_store.write( + [{"id": "1", "content": "ship the migration", "status": "pending"}], + merge=False, + ) + snapshot = agent._todo_store.format_for_injection() + assert snapshot + + compressed, _ = agent._compress_context(_msgs(), "sys", approx_tokens=120_000) + assert agent.session_id == session # in-place compaction kept the id + + # In-place mode writes the compacted transcript directly via + # archive_and_compact(). The live DB transcript should therefore carry + # the same merged tail and never persist a user/user pair caused by the + # todo snapshot. + live = db.get_messages(session) + assert [(m["role"], m["content"]) for m in live] == [ + (m["role"], m["content"]) for m in compressed + ] + persisted_roles = [m["role"] for m in live] + for a, b in zip(persisted_roles, persisted_roles[1:]): + assert not (a == "user" and b == "user"), ( + "consecutive user/user messages persisted for live transcript " + "caused by the todo snapshot" + ) + persisted_last = live[-1] + assert persisted_last["role"] == "user" + assert "tail" in persisted_last["content"] + assert snapshot in persisted_last["content"] diff --git a/tests/cron/test_cron_no_agent.py b/tests/cron/test_cron_no_agent.py index af94713868be..ff37b4675db6 100644 --- a/tests/cron/test_cron_no_agent.py +++ b/tests/cron/test_cron_no_agent.py @@ -31,6 +31,8 @@ def hermes_env(tmp_path, monkeypatch): import importlib import hermes_constants importlib.reload(hermes_constants) + import hermes_state + importlib.reload(hermes_state) # DEFAULT_DB_PATH binds at import time import cron.jobs importlib.reload(cron.jobs) import cron.scheduler @@ -329,3 +331,126 @@ def test_run_job_script_path_traversal_still_blocked(hermes_env): ok, output = _run_job_script("/etc/passwd") assert ok is False assert "Blocked" in output or "outside" in output + + +# --------------------------------------------------------------------------- +# run_job: no_agent runs are recorded as run-history sessions (#44080) +# --------------------------------------------------------------------------- + + +def _job_runs(job_id): + from hermes_state import SessionDB + + db = SessionDB() + try: + return db.list_cron_job_runs(job_id) + finally: + db.close() + + +def _session_messages(session_id): + from hermes_state import SessionDB + + db = SessionDB() + try: + return db.get_messages(session_id) + finally: + db.close() + + +def test_run_job_no_agent_success_records_run_session(hermes_env): + """A successful no_agent run must appear in the job's run history.""" + from cron.jobs import create_job + from cron.scheduler import run_job + + script_path = hermes_env / "scripts" / "alert.sh" + script_path.write_text("#!/bin/bash\necho 'RAM 92% on host'\n") + + job = create_job( + prompt=None, schedule="every 5m", script="alert.sh", no_agent=True, deliver="local" + ) + success, _, _, _ = run_job(job) + assert success is True + + runs = _job_runs(job["id"]) + assert len(runs) == 1 + run = runs[0] + assert run["id"].startswith(f"cron_{job['id']}_") + assert run["source"] == "cron" + # The run finished, so the GUI must not show it as still active. + assert run["ended_at"] is not None + assert run["end_reason"] == "cron_complete" + + # Script output is persisted so the GUI can show what the run produced. + messages = _session_messages(run["id"]) + roles = [m["role"] for m in messages] + assert "user" in roles and "assistant" in roles + assistant_text = " ".join( + str(m.get("content") or "") for m in messages if m["role"] == "assistant" + ) + assert "RAM 92% on host" in assistant_text + + +def test_run_job_no_agent_failure_records_run_session(hermes_env): + """A failed script run must be visible in run history, not silent.""" + from cron.jobs import create_job + from cron.scheduler import run_job + + script_path = hermes_env / "scripts" / "broken.sh" + script_path.write_text("#!/bin/bash\necho oops >&2\nexit 3\n") + + job = create_job( + prompt=None, schedule="every 5m", script="broken.sh", no_agent=True, deliver="local" + ) + success, _, _, _ = run_job(job) + assert success is False + + runs = _job_runs(job["id"]) + assert len(runs) == 1 + run = runs[0] + assert run["ended_at"] is not None + assert run["end_reason"] == "cron_failed" + + messages = _session_messages(run["id"]) + assistant_text = " ".join( + str(m.get("content") or "") for m in messages if m["role"] == "assistant" + ) + assert "script failed" in assistant_text + + +def test_run_job_no_agent_silent_records_run_session(hermes_env): + """A silent run (empty stdout) still leaves a run record.""" + from cron.jobs import create_job + from cron.scheduler import run_job + + script_path = hermes_env / "scripts" / "quiet.sh" + script_path.write_text("#!/bin/bash\ntrue\n") + + job = create_job( + prompt=None, schedule="every 5m", script="quiet.sh", no_agent=True, deliver="local" + ) + success, _, _, _ = run_job(job) + assert success is True + + runs = _job_runs(job["id"]) + assert len(runs) == 1 + assert runs[0]["ended_at"] is not None + + +def test_run_job_no_agent_broken_session_store_does_not_break_run(hermes_env): + """A broken SessionDB must not prevent the script from running.""" + from cron.jobs import create_job + from cron.scheduler import run_job + + script_path = hermes_env / "scripts" / "ok.sh" + script_path.write_text("#!/bin/bash\necho fine\n") + + job = create_job( + prompt=None, schedule="every 5m", script="ok.sh", no_agent=True, deliver="local" + ) + + with patch("hermes_state.SessionDB", side_effect=RuntimeError("db locked")): + success, output, final_response, error = run_job(job) + + assert success is True + assert "fine" in output diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index b9af9f25aab2..1f90ce2698f1 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -1662,6 +1662,14 @@ def test_message_count_per_session(self, db): # Delete and export # ========================================================================= +def _make_compression_lineage(db): + db.create_session(session_id="root", source="cli") + db.append_message("root", role="user", content="before compression") + db.end_session("root", end_reason="compression") + db.create_session(session_id="tip", source="cli", parent_session_id="root") + db.append_message("tip", role="user", content="after compression") + + class TestDeleteAndExport: def test_delete_session(self, db): db.create_session(session_id="s1", source="cli") @@ -1674,6 +1682,37 @@ def test_delete_session(self, db): def test_delete_nonexistent(self, db): assert db.delete_session("nope") is False + def test_delete_compression_tip_removes_projected_root(self, db): + _make_compression_lineage(db) + assert [ + (s["id"], s.get("_lineage_root_id")) + for s in db.list_sessions_rich(min_message_count=1, order_by_last_active=True) + ] == [("tip", "root")] + + assert db.delete_session("tip") is True + + assert db.get_session("root") is None + assert db.get_session("tip") is None + assert db.list_sessions_rich(min_message_count=1, order_by_last_active=True) == [] + + def test_delete_compression_lineage_preserves_explicit_branch(self, db): + _make_compression_lineage(db) + db.create_session( + session_id="branch", + source="cli", + parent_session_id="root", + model_config={"_branched_from": "root"}, + ) + db.append_message("branch", role="user", content="branched work") + + assert db.delete_session("tip") is True + + assert db.get_session("root") is None + assert db.get_session("tip") is None + branch = db.get_session("branch") + assert branch is not None + assert branch["parent_session_id"] is None + def test_resolve_session_id_exact(self, db): db.create_session(session_id="20260315_092437_c9a6ff", source="cli") assert db.resolve_session_id("20260315_092437_c9a6ff") == "20260315_092437_c9a6ff" @@ -1975,6 +2014,20 @@ def test_cleans_up_transcript_files(self, db, tmp_path): assert not (tmp_path / "s1.jsonl").exists() assert not (tmp_path / "s2.json").exists() + def test_deleting_compression_tip_removes_projected_root(self, db): + _make_compression_lineage(db) + assert [ + (s["id"], s.get("_lineage_root_id")) + for s in db.list_sessions_rich(min_message_count=1, order_by_last_active=True) + ] == [("tip", "root")] + + deleted = db.delete_sessions(["tip"]) + + assert deleted == 1 + assert db.get_session("root") is None + assert db.get_session("tip") is None + assert db.list_sessions_rich(min_message_count=1, order_by_last_active=True) == [] + class TestDeleteEmptySessions: """``delete_empty_sessions`` sweeps every ended, non-archived session diff --git a/ui-tui/src/app/createGatewayEventHandler.ts b/ui-tui/src/app/createGatewayEventHandler.ts index 45532b2058d7..0ae8aa346515 100644 --- a/ui-tui/src/app/createGatewayEventHandler.ts +++ b/ui-tui/src/app/createGatewayEventHandler.ts @@ -404,7 +404,10 @@ export function createGatewayEventHandler(ctx: GatewayEventHandlerContext): (ev: return (ev: GatewayEvent) => { const sid = getUiState().sid - if (ev.session_id && sid && ev.session_id !== sid && !ev.type.startsWith('gateway.')) { + // Filter events by session id. When sid is null (during session switch / + // reset), drop ALL non-gateway session events — the null-sid window must + // not let another live session's events bleed into the view (#51058). + if (ev.session_id && (!sid || ev.session_id !== sid) && !ev.type.startsWith('gateway.')) { return }