From c83efbbe863f862f7a2558b3275047f0258ccfb1 Mon Sep 17 00:00:00 2001 From: Pierce Brookins Date: Sun, 7 Jun 2026 11:34:41 -0500 Subject: [PATCH 1/2] Add session tree navigation commands --- code_puppy/plugins/session_tree/__init__.py | 1 + .../session_tree/register_callbacks.py | 286 ++++++++++++++++ code_puppy/plugins/session_tree/tree_menu.py | 268 +++++++++++++++ code_puppy/plugins/session_tree/tree_model.py | 163 +++++++++ tests/plugins/test_session_tree.py | 317 ++++++++++++++++++ 5 files changed, 1035 insertions(+) create mode 100644 code_puppy/plugins/session_tree/__init__.py create mode 100644 code_puppy/plugins/session_tree/register_callbacks.py create mode 100644 code_puppy/plugins/session_tree/tree_menu.py create mode 100644 code_puppy/plugins/session_tree/tree_model.py create mode 100644 tests/plugins/test_session_tree.py diff --git a/code_puppy/plugins/session_tree/__init__.py b/code_puppy/plugins/session_tree/__init__.py new file mode 100644 index 000000000..52230a858 --- /dev/null +++ b/code_puppy/plugins/session_tree/__init__.py @@ -0,0 +1 @@ +"""Session tree navigation plugin.""" diff --git a/code_puppy/plugins/session_tree/register_callbacks.py b/code_puppy/plugins/session_tree/register_callbacks.py new file mode 100644 index 000000000..2f566d0ec --- /dev/null +++ b/code_puppy/plugins/session_tree/register_callbacks.py @@ -0,0 +1,286 @@ +"""Session navigation commands for Code Puppy.""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +from datetime import datetime +from pathlib import Path +from typing import Any + +from code_puppy.callbacks import register_callback + +from .tree_menu import TreeSelectionMenu +from .tree_model import ( + HistoryNode, + build_nodes, + history_before, + history_through, + message_text, + resolve_node, + selectable_nodes, + user_nodes, +) + +_LABELS: dict[str, str] = {} + + +def _emit_error(message: Any) -> None: + from code_puppy.messaging import emit_error + + emit_error(message) + + +def _emit_info(message: Any) -> None: + from code_puppy.messaging import emit_info + + emit_info(message) + + +def _emit_success(message: Any) -> None: + from code_puppy.messaging import emit_success + + emit_success(message) + + +def _emit_warning(message: Any) -> None: + from code_puppy.messaging import emit_warning + + emit_warning(message) + + +def _current_history() -> tuple[Any, list[Any]]: + from code_puppy.agents.agent_manager import get_current_agent + + agent = get_current_agent() + return agent, list(agent.get_message_history()) + + +def _help_entries() -> list[tuple[str, str]]: + return [ + ( + "tree", + "Open a TUI to inspect, restore, fork, or summarize prior conversation points", + ), + ("fork", "Create a new branch/session from a previous user message"), + ("clone", "Duplicate the current active branch into a new autosave session"), + ] + + +def _render_nodes(title: str, nodes: list[HistoryNode]) -> None: + lines = [title] + for ordinal, node in enumerate(nodes, start=1): + marker = "*" if node.is_system else " " + label = f" [{node.label}]" if node.label else "" + lines.append( + f" {ordinal}. [{node.node_id}] {marker}{node.role}: {node.preview}{label}" + ) + _emit_info("\n".join(lines)) + + +def _tree_args(command: str) -> tuple[bool, str]: + parts = command.split(maxsplit=2) + if len(parts) >= 2 and parts[1].lower() in {"summary", "summarize"}: + return True, parts[2].strip() if len(parts) == 3 else "" + return False, parts[1].strip() if len(parts) >= 2 else "" + + +def _selection_from_command(command: str) -> str: + return _tree_args(command)[1] + + +def _markdown_result(prompt: str) -> Any: + from code_puppy.plugins.customizable_commands.register_callbacks import ( + MarkdownCommandResult, + ) + + return MarkdownCommandResult(prompt) + + +def _run_tree_menu(history: list[Any]) -> Any: + with concurrent.futures.ThreadPoolExecutor() as executor: + future = executor.submit( + lambda: asyncio.run( + TreeSelectionMenu(history=history, labels=_LABELS).run_async() + ) + ) + return future.result(timeout=300) + + +def _resolve_or_menu( + command: str, history: list[Any] +) -> tuple[HistoryNode | None, bool]: + summarize, selection = _tree_args(command) + if selection: + nodes = selectable_nodes(history) + node = resolve_node(selection, nodes) + if node is None: + _emit_warning(f"/tree: no history point matches '{selection}'") + return node, summarize + + result = _run_tree_menu(history) + if result.cancelled: + _emit_warning("/tree: selection cancelled") + return None, False + return result.node, result.summarize + + +def _summarize_through(history: list[Any], node: HistoryNode) -> list[Any]: + from code_puppy.agents._compaction import _run_summarization_core + + compacted, _summarized = _run_summarization_core( + history_through(history, node.index), + protected_tokens=0, + with_protection=False, + model_name=None, + ) + return list(compacted) + + +def _restore_summary(agent: Any, history: list[Any], node: HistoryNode) -> bool: + try: + summarized_history = _summarize_through(history, node) + agent.set_message_history(summarized_history) + except Exception as exc: # noqa: BLE001 + _emit_error(f"/tree summary: failed to summarize history - {exc}") + return True + + _emit_success( + f"Summarized conversation through {node.role} message {node.index} " + f"[{node.node_id}] into a fresh conversation ({len(summarized_history)} messages)." + ) + return True + + +def _handle_tree(command: str) -> bool | Any: + try: + agent, history = _current_history() + except Exception as exc: # noqa: BLE001 - commands must fail soft. + _emit_error(f"/tree: could not read current history - {exc}") + return True + + nodes = selectable_nodes(history) + if not nodes: + _emit_warning("/tree: conversation history is empty - nothing to restore") + return True + + try: + node, summarize = _resolve_or_menu(command, history) + except Exception as exc: # noqa: BLE001 + _emit_warning(f"/tree TUI failed, falling back to text view: {exc}") + _render_nodes("Conversation Tree:", build_nodes(history, _LABELS)) + return True + + if node is None: + return True + + if summarize: + return _restore_summary(agent, history, node) + + try: + if node.is_userish: + prompt = message_text(history[node.index]).strip() + agent.set_message_history(history_before(history, node.index)) + _emit_success( + f"Moved to parent of {node.role} message {node.index} [{node.node_id}]. " + "Edit the restored prompt to create a branch." + ) + return _markdown_result(prompt) if prompt else True + + agent.set_message_history(history_through(history, node.index)) + except Exception as exc: # noqa: BLE001 + _emit_error(f"/tree: failed to restore history - {exc}") + return True + + _emit_success( + f"Restored conversation to {node.role} message {node.index} [{node.node_id}]" + ) + return True + + +def _handle_fork(command: str) -> bool | Any: + try: + agent, history = _current_history() + except Exception as exc: # noqa: BLE001 + _emit_error(f"/fork: could not read current history - {exc}") + return True + + nodes = user_nodes(history) + if not nodes: + _emit_warning("/fork: no previous user messages are available to fork") + return True + + selection = _selection_from_command(command) + if not selection: + _render_nodes("Forkable User Messages:", nodes) + _emit_warning("Usage: /fork ") + return True + + node = resolve_node(selection, nodes) + if node is None: + _emit_warning(f"/fork: no user message matches '{selection}'") + return True + + prompt = message_text(history[node.index]).strip() + try: + agent.set_message_history(history_before(history, node.index)) + except Exception as exc: # noqa: BLE001 + _emit_error(f"/fork: failed to create branch history - {exc}") + return True + + _emit_success( + f"Created branch before user message {node.index} [{node.node_id}]. " + "Edit the restored prompt to continue from this branch." + ) + return _markdown_result(prompt) if prompt else True + + +def _handle_clone() -> bool: + try: + agent, history = _current_history() + if not history: + _emit_warning("/clone: no active branch to clone") + return True + + from code_puppy.config import AUTOSAVE_DIR, rotate_autosave_id + from code_puppy.session_storage import save_session + + session_id = rotate_autosave_id() + session_name = f"autosave_{session_id}" + metadata = save_session( + history=history, + session_name=session_name, + base_dir=Path(AUTOSAVE_DIR), + timestamp=datetime.now().isoformat(), + token_estimator=agent.estimate_tokens_for_message, + auto_saved=True, + ) + _emit_success( + f"Cloned current active branch to new autosave session {session_id} " + f"({metadata.message_count} messages)." + ) + except Exception as exc: # noqa: BLE001 + _emit_error(f"/clone: failed to clone active branch - {exc}") + return True + + +def _handle_custom_command(command: str, name: str) -> bool | str | None: + if name == "tree": + return _handle_tree(command) + if name == "fork": + return _handle_fork(command) + if name == "clone": + return _handle_clone() + return None + + +register_callback("custom_command_help", _help_entries) +register_callback("custom_command", _handle_custom_command) + + +__all__ = [ + "_handle_custom_command", + "_handle_fork", + "_handle_tree", + "_help_entries", +] diff --git a/code_puppy/plugins/session_tree/tree_menu.py b/code_puppy/plugins/session_tree/tree_menu.py new file mode 100644 index 000000000..689837031 --- /dev/null +++ b/code_puppy/plugins/session_tree/tree_menu.py @@ -0,0 +1,268 @@ +"""Prompt-toolkit TUI for session tree selection.""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from prompt_toolkit import Application +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.layout import Layout, Window +from prompt_toolkit.layout.controls import FormattedTextControl + +from code_puppy.command_line.pagination import ( + ensure_visible_page, + get_page_bounds, + get_page_for_index, + get_total_pages, +) + +from .tree_model import ( + FILTER_MODES, + FilterMode, + HistoryNode, + build_nodes, + visible_nodes, +) + +TREE_PAGE_SIZE = 14 + + +@dataclass(slots=True) +class TreeMenuResult: + node: HistoryNode | None = None + cancelled: bool = False + summarize: bool = False + custom_summary_focus: str = "" + + +@dataclass(slots=True) +class TreeSelectionMenu: + history: list[Any] + labels: dict[str, str] = field(default_factory=dict) + mode: FilterMode = FilterMode.DEFAULT + selected_index: int = 0 + page: int = 0 + page_size: int = TREE_PAGE_SIZE + result: TreeMenuResult = field(default_factory=TreeMenuResult) + show_label_timestamps: bool = True + + def __post_init__(self) -> None: + self._clamp_selection() + + @property + def nodes(self) -> list[HistoryNode]: + return build_nodes(self.history, self.labels) + + @property + def visible(self) -> list[HistoryNode]: + return visible_nodes(self.nodes, self.mode) + + @property + def total_pages(self) -> int: + return get_total_pages(len(self.visible), self.page_size) + + @property + def page_start(self) -> int: + start, _ = get_page_bounds(self.page, len(self.visible), self.page_size) + return start + + @property + def page_end(self) -> int: + _, end = get_page_bounds(self.page, len(self.visible), self.page_size) + return end + + @property + def page_nodes(self) -> list[HistoryNode]: + return self.visible[self.page_start : self.page_end] + + def selected_node(self) -> HistoryNode | None: + if 0 <= self.selected_index < len(self.visible): + return self.visible[self.selected_index] + return None + + def _clamp_selection(self) -> None: + if not self.visible: + self.selected_index = 0 + self.page = 0 + return + self.selected_index = max(0, min(self.selected_index, len(self.visible) - 1)) + self.page = ensure_visible_page( + self.selected_index, self.page, len(self.visible), self.page_size + ) + + def _move(self, delta: int) -> None: + self.selected_index += delta + self._clamp_selection() + + def _page(self, delta: int) -> None: + if not self.visible: + return + self.page = max(0, min(self.page + delta, self.total_pages - 1)) + self.selected_index = self.page_start + + def _cycle_filter(self) -> None: + modes = list(FILTER_MODES) + next_index = (modes.index(self.mode) + 1) % len(modes) + selected = self.selected_node() + self.mode = modes[next_index] + if selected and selected in self.visible: + self.selected_index = self.visible.index(selected) + else: + self.selected_index = 0 + self._clamp_selection() + + def _toggle_label_timestamp(self) -> None: + self.show_label_timestamps = not self.show_label_timestamps + + def _toggle_label(self) -> None: + selected = self.selected_node() + if selected is None: + return + if selected.node_id in self.labels: + del self.labels[selected.node_id] + return + suffix = ( + f" @ {datetime.now().strftime('%H:%M')}" + if self.show_label_timestamps + else "" + ) + self.labels[selected.node_id] = f"label{suffix}" + + def _branch_prefix(self, node: HistoryNode) -> str: + if node.index == len(self.history) - 1: + return "└─" + return "├─" + + def _render(self): + lines = [("bold cyan", " Session Tree")] + lines.append(("fg:ansibrightblack", f"\n Filter: {self.mode.value}")) + if self.total_pages > 1: + lines.append( + ("fg:ansibrightblack", f" Page {self.page + 1}/{self.total_pages}") + ) + lines.append(("", "\n")) + + if not self.visible: + lines.append(("fg:ansiyellow", "\n No entries match this filter.\n\n")) + for offset, node in enumerate(self.page_nodes): + absolute_index = self.page_start + offset + is_selected = absolute_index == self.selected_index + is_active = node.index == len(self.history) - 1 + cursor = " › " if is_selected else " " + style = "fg:ansiwhite bold" if is_selected else "fg:ansibrightblack" + label = f" [{node.label}]" if node.label else "" + active = " ← active" if is_active else "" + lines.append( + ( + style, + f'{cursor}{self._branch_prefix(node)} {node.role}: "{node.preview}"', + ) + ) + lines.append(("fg:ansiblue", f" #{node.node_id}{label}{active}\n")) + + lines.append(("", "\n")) + lines.append(("fg:ansibrightblack", " ↑/↓ ")) + lines.append(("", "Navigate\n")) + lines.append(("fg:ansibrightblack", " ←/→ PgUp/PgDn ")) + lines.append(("", "Page\n")) + lines.append(("fg:ansibrightblack", " Ctrl+O ")) + lines.append(("", "Cycle filter\n")) + lines.append(("fg:ansibrightblack", " L ")) + lines.append(("", "Set/clear label\n")) + lines.append(("fg:ansibrightblack", " T ")) + lines.append(("", "Toggle label timestamps\n")) + lines.append(("fg:ansigreen", " Enter ")) + lines.append(("", "Select/restore\n")) + lines.append(("fg:ansigreen", " S ")) + lines.append(("", "Summarize through selection as new conversation\n")) + lines.append(("fg:ansiyellow", " Esc/Ctrl+C ")) + lines.append(("", "Cancel\n")) + return lines + + async def run_async(self) -> TreeMenuResult: + control = FormattedTextControl(lambda: self._render()) + kb = KeyBindings() + + def refresh(event) -> None: + control.text = self._render() + event.app.invalidate() + + @kb.add("up") + @kb.add("c-p") + def _(event): + self._move(-1) + refresh(event) + + @kb.add("down") + @kb.add("c-n") + def _(event): + self._move(1) + refresh(event) + + @kb.add("pageup") + @kb.add("left") + def _(event): + self._page(-1) + refresh(event) + + @kb.add("pagedown") + @kb.add("right") + def _(event): + self._page(1) + refresh(event) + + @kb.add("c-o") + def _(event): + self._cycle_filter() + refresh(event) + + @kb.add("L") + @kb.add("l") + def _(event): + self._toggle_label() + refresh(event) + + @kb.add("T") + @kb.add("t") + def _(event): + self._toggle_label_timestamp() + refresh(event) + + @kb.add("enter") + def _(event): + self.result.node = self.selected_node() + event.app.exit() + + @kb.add("S") + @kb.add("s") + def _(event): + self.result.node = self.selected_node() + self.result.summarize = True + event.app.exit() + + @kb.add("escape") + @kb.add("c-c") + def _(event): + self.result.cancelled = True + event.app.exit() + + sys.stdout.write("\033[?1049h\033[2J\033[H") + sys.stdout.flush() + try: + app = Application( + layout=Layout(Window(content=control, wrap_lines=True)), + key_bindings=kb, + full_screen=False, + ) + await app.run_async() + finally: + sys.stdout.write("\033[?1049l") + sys.stdout.flush() + return self.result + + +def selected_index_to_page(index: int, page_size: int = TREE_PAGE_SIZE) -> int: + return get_page_for_index(index, page_size) diff --git a/code_puppy/plugins/session_tree/tree_model.py b/code_puppy/plugins/session_tree/tree_model.py new file mode 100644 index 000000000..0fa40840e --- /dev/null +++ b/code_puppy/plugins/session_tree/tree_model.py @@ -0,0 +1,163 @@ +"""Small helpers for navigating Code Puppy message history.""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from enum import StrEnum +from typing import Any, Iterable, Sequence + + +class FilterMode(StrEnum): + DEFAULT = "default" + NO_TOOLS = "no-tools" + USER_ONLY = "user-only" + LABELED_ONLY = "labeled-only" + ALL = "all" + + +FILTER_MODES = tuple(FilterMode) +TOOL_ROLES = {"tool", "toolresult", "tool_result", "bashexecution"} +USERISH_ROLES = {"user", "custom"} + + +@dataclass(frozen=True, slots=True) +class HistoryNode: + index: int + node_id: str + role: str + preview: str + is_system: bool = False + label: str = "" + + @property + def is_userish(self) -> bool: + return self.role.lower() in USERISH_ROLES + + @property + def is_toolish(self) -> bool: + return self.role.lower() in TOOL_ROLES + + +def message_role(message: Any) -> str: + role = getattr(message, "role", None) + if isinstance(role, str): + return role + name = type(message).__name__.lower() + if "response" in name: + return "assistant" + if "request" in name: + return "user" + return name or "message" + + +def message_text(message: Any) -> str: + content = getattr(message, "content", None) + if isinstance(content, str): + return content + + parts = getattr(message, "parts", None) + if parts is None: + parts = content if isinstance(content, list) else [] + + chunks: list[str] = [] + for part in parts or []: + text = getattr(part, "content", None) + if text is None: + text = getattr(part, "text", None) + if isinstance(part, dict): + text = part.get("content") or part.get("text") + if isinstance(text, str): + chunks.append(text) + + if chunks: + return " ".join(chunks) + return str(message) + + +def node_id_for(index: int, message: Any) -> str: + raw = f"{index}:{type(message).__name__}:{message_text(message)}".encode() + return hashlib.sha1(raw).hexdigest()[:8] + + +def _preview(text: str, max_len: int = 72) -> str: + clean = " ".join(text.split()) + if not clean: + clean = "(no text)" + if len(clean) <= max_len: + return clean + return clean[: max_len - 1] + "…" + + +def build_nodes( + history: Sequence[Any], labels: dict[str, str] | None = None +) -> list[HistoryNode]: + labels = labels or {} + nodes: list[HistoryNode] = [] + for index, message in enumerate(history): + role = message_role(message) + node_id = node_id_for(index, message) + nodes.append( + HistoryNode( + index=index, + node_id=node_id, + role=role, + preview=_preview(message_text(message)), + is_system=index == 0 and role == "system", + label=labels.get(node_id, ""), + ) + ) + return nodes + + +def selectable_nodes(history: Sequence[Any]) -> list[HistoryNode]: + return [node for node in build_nodes(history) if not node.is_system] + + +def user_nodes(history: Sequence[Any]) -> list[HistoryNode]: + return [node for node in selectable_nodes(history) if node.is_userish] + + +def visible_nodes(nodes: Sequence[HistoryNode], mode: FilterMode) -> list[HistoryNode]: + if mode == FilterMode.ALL: + return list(nodes) + if mode == FilterMode.NO_TOOLS: + return [node for node in nodes if not node.is_toolish] + if mode == FilterMode.USER_ONLY: + return [node for node in nodes if node.is_userish] + if mode == FilterMode.LABELED_ONLY: + return [node for node in nodes if node.label] + return [node for node in nodes if not node.is_system and not node.is_toolish] + + +def resolve_node(selection: str, nodes: Iterable[HistoryNode]) -> HistoryNode | None: + value = selection.strip() + if not value: + return None + + node_list = list(nodes) + if value.isdigit(): + ordinal = int(value) + if 1 <= ordinal <= len(node_list): + return node_list[ordinal - 1] + for node in node_list: + if node.index == ordinal: + return node + + value_lower = value.lower() + for node in node_list: + if node.node_id.startswith(value_lower): + return node + return None + + +def history_through(history: Sequence[Any], index: int) -> list[Any]: + if index < 0: + return [] + return list(history[: index + 1]) + + +def history_before(history: Sequence[Any], index: int) -> list[Any]: + if index <= 0: + return list(history[:1]) if history else [] + return list(history[:index]) diff --git a/tests/plugins/test_session_tree.py b/tests/plugins/test_session_tree.py new file mode 100644 index 000000000..21a861ab7 --- /dev/null +++ b/tests/plugins/test_session_tree.py @@ -0,0 +1,317 @@ +"""Tests for the /tree, /fork, and /clone session navigation plugin.""" + +from __future__ import annotations + +import importlib +import sys +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + + +class Message(SimpleNamespace): + pass + + +def _plugin_module(): + sys.modules.setdefault("dbos", MagicMock()) + return importlib.import_module("code_puppy.plugins.session_tree.register_callbacks") + + +def _tree_model(): + return importlib.import_module("code_puppy.plugins.session_tree.tree_model") + + +def _agent_manager_module(agent: MagicMock) -> SimpleNamespace: + return SimpleNamespace(get_current_agent=lambda: agent) + + +def _history() -> list[Message]: + return [ + Message(role="system", content="system prompt"), + Message(role="user", content="first task"), + Message(role="assistant", content="first answer"), + Message(role="toolResult", content="tool output"), + Message(role="user", content="second task"), + Message(role="assistant", content="second answer"), + ] + + +def test_build_nodes_creates_stable_tree_entries(): + nodes = _tree_model().build_nodes(_history()) + + assert len(nodes) == 6 + assert nodes[0].is_system is True + assert nodes[1].role == "user" + assert nodes[1].preview == "first task" + assert len(nodes[1].node_id) == 8 + + +def test_visible_nodes_supports_filter_modes_and_labels(): + model = _tree_model() + history = _history() + base_nodes = model.build_nodes(history) + labels = {base_nodes[2].node_id: "keeper"} + nodes = model.build_nodes(history, labels) + + assert all( + not node.is_toolish + for node in model.visible_nodes(nodes, model.FilterMode.NO_TOOLS) + ) + assert { + node.role for node in model.visible_nodes(nodes, model.FilterMode.USER_ONLY) + } == {"user"} + assert model.visible_nodes(nodes, model.FilterMode.LABELED_ONLY) == [nodes[2]] + assert len(model.visible_nodes(nodes, model.FilterMode.ALL)) == len(nodes) + + +def test_resolve_node_accepts_ordinal_index_and_id_prefix(): + model = _tree_model() + nodes = model.selectable_nodes(_history()) + + assert model.resolve_node("1", nodes).preview == "first task" + assert model.resolve_node("4", nodes).preview == "second task" + assert model.resolve_node(nodes[1].node_id[:4], nodes) == nodes[1] + assert model.resolve_node("nope", nodes) is None + + +def test_history_helpers_preserve_system_prompt_for_forks(): + model = _tree_model() + history = _history() + + assert model.history_through(history, 2) == history[:3] + assert model.history_before(history, 1) == history[:1] + + +def test_help_entries_include_tree_fork_clone_and_summary_language(): + entries = dict(_plugin_module()._help_entries()) + + assert "summarize" in entries["tree"] + assert entries["fork"] + assert entries["clone"] + + +def test_handle_custom_command_ignores_unknown_command(): + assert _plugin_module()._handle_custom_command("/nope", "nope") is None + + +def test_tree_restores_non_user_history_point_from_argument(): + history = _history() + agent = MagicMock() + agent.get_message_history.return_value = history + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + ): + result = _plugin_module()._handle_custom_command("/tree 2", "tree") + + assert result is True + agent.set_message_history.assert_called_once_with(history[:3]) + + +def test_tree_summary_selection_replaces_history_with_summarized_conversation(): + history = _history() + summarized = [history[0], Message(role="user", content="Summary: compacted")] + agent = MagicMock() + agent.get_message_history.return_value = history + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._summarize_through", + return_value=summarized, + ) as summarize, + patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + ): + result = _plugin_module()._handle_custom_command("/tree summary 2", "tree") + + assert result is True + summarize.assert_called_once() + agent.set_message_history.assert_called_once_with(summarized) + + +def test_tree_summary_from_tui_result_replaces_history(): + plugin = _plugin_module() + history = _history() + summarized = [history[0], Message(role="user", content="Summary: compacted")] + agent = MagicMock() + agent.get_message_history.return_value = history + node = _tree_model().selectable_nodes(history)[1] + result_obj = SimpleNamespace(node=node, cancelled=False, summarize=True) + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._run_tree_menu", + return_value=result_obj, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._summarize_through", + return_value=summarized, + ), + patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + ): + result = plugin._handle_custom_command("/tree", "tree") + + assert result is True + agent.set_message_history.assert_called_once_with(summarized) + + +def test_tree_summary_failure_does_not_mutate_history(): + history = _history() + agent = MagicMock() + agent.get_message_history.return_value = history + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._summarize_through", + side_effect=RuntimeError("no biscuits"), + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._emit_error" + ) as error, + ): + result = _plugin_module()._handle_custom_command("/tree summarize 2", "tree") + + assert result is True + agent.set_message_history.assert_not_called() + assert "failed to summarize" in str(error.call_args) + + +def test_tree_user_selection_returns_prompt_and_restores_parent(): + history = _history() + agent = MagicMock() + agent.get_message_history.return_value = history + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + ): + result = _plugin_module()._handle_custom_command("/tree 4", "tree") + + assert str(result) == "second task" + agent.set_message_history.assert_called_once_with(history[:4]) + + +def test_tree_warns_for_empty_or_system_only_history(): + agent = MagicMock() + agent.get_message_history.return_value = [Message(role="system", content="system")] + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._emit_warning" + ) as warning, + ): + result = _plugin_module()._handle_custom_command("/tree", "tree") + + assert result is True + agent.set_message_history.assert_not_called() + assert "nothing to restore" in str(warning.call_args) + + +def test_tree_warns_for_invalid_selection(): + history = _history() + agent = MagicMock() + agent.get_message_history.return_value = history + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._emit_warning" + ) as warning, + ): + result = _plugin_module()._handle_custom_command("/tree nope", "tree") + + assert result is True + agent.set_message_history.assert_not_called() + assert "no history point matches" in str(warning.call_args) + + +def test_fork_returns_selected_prompt_and_trims_history_before_it(): + history = _history() + agent = MagicMock() + agent.get_message_history.return_value = history + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + ): + result = _plugin_module()._handle_custom_command("/fork 2", "fork") + + assert str(result) == "second task" + agent.set_message_history.assert_called_once_with(history[:4]) + + +def test_fork_warns_when_no_user_messages_exist(): + agent = MagicMock() + agent.get_message_history.return_value = [ + Message(role="system", content="system"), + Message(role="assistant", content="hello"), + ] + + with ( + patch.dict( + sys.modules, + {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + ), + patch( + "code_puppy.plugins.session_tree.register_callbacks._emit_warning" + ) as warning, + ): + result = _plugin_module()._handle_custom_command("/fork", "fork") + + assert result is True + agent.set_message_history.assert_not_called() + assert "no previous user messages" in str(warning.call_args) + + +def test_clone_saves_active_branch_to_new_autosave_session(tmp_path): + history = _history() + agent = MagicMock() + agent.get_message_history.return_value = history + agent.estimate_tokens_for_message.return_value = 1 + + config = SimpleNamespace( + AUTOSAVE_DIR=str(tmp_path), rotate_autosave_id=lambda: "abc123" + ) + with ( + patch.dict( + sys.modules, + { + "code_puppy.agents.agent_manager": _agent_manager_module(agent), + "code_puppy.config": config, + }, + ), + patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + ): + result = _plugin_module()._handle_custom_command("/clone", "clone") + + assert result is True + assert (tmp_path / "autosave_abc123.pkl").exists() + assert (tmp_path / "autosave_abc123_meta.json").exists() From 0bfe6ef1753ca07f5136b52282aab127865d2c94 Mon Sep 17 00:00:00 2001 From: Pierce Brookins Date: Tue, 21 Jul 2026 14:19:51 -0400 Subject: [PATCH 2/2] Fix session tree persistence and branch navigation --- .../session_tree/register_callbacks.py | 260 ++++++------ code_puppy/plugins/session_tree/tree_menu.py | 71 +++- code_puppy/plugins/session_tree/tree_model.py | 163 ++++---- code_puppy/plugins/session_tree/tree_store.py | 135 ++++++ tests/plugins/test_session_tree.py | 386 ++++++++---------- 5 files changed, 550 insertions(+), 465 deletions(-) create mode 100644 code_puppy/plugins/session_tree/tree_store.py diff --git a/code_puppy/plugins/session_tree/register_callbacks.py b/code_puppy/plugins/session_tree/register_callbacks.py index 2f566d0ec..cbb382100 100644 --- a/code_puppy/plugins/session_tree/register_callbacks.py +++ b/code_puppy/plugins/session_tree/register_callbacks.py @@ -12,72 +12,52 @@ from .tree_menu import TreeSelectionMenu from .tree_model import ( + FilterMode, HistoryNode, - build_nodes, - history_before, - history_through, message_text, resolve_node, selectable_nodes, user_nodes, ) +from .tree_store import SessionTree, TreeStore -_LABELS: dict[str, str] = {} +def _emit(kind: str, message: str) -> None: + from code_puppy import messaging -def _emit_error(message: Any) -> None: - from code_puppy.messaging import emit_error + getattr(messaging, f"emit_{kind}")(message) - emit_error(message) +def _current_history() -> tuple[Any, list[Any]]: + from code_puppy.agents.agent_manager import get_current_agent -def _emit_info(message: Any) -> None: - from code_puppy.messaging import emit_info - - emit_info(message) - - -def _emit_success(message: Any) -> None: - from code_puppy.messaging import emit_success - - emit_success(message) - + agent = get_current_agent() + return agent, list(agent.get_message_history()) -def _emit_warning(message: Any) -> None: - from code_puppy.messaging import emit_warning - emit_warning(message) +def _tree_store() -> TreeStore: + from code_puppy.config import AUTOSAVE_DIR, get_current_autosave_id + path = Path(AUTOSAVE_DIR) / "session_trees" / f"{get_current_autosave_id()}.pkl" + return TreeStore(path) -def _current_history() -> tuple[Any, list[Any]]: - from code_puppy.agents.agent_manager import get_current_agent - agent = get_current_agent() - return agent, list(agent.get_message_history()) +def _load_tree(history: list[Any]) -> tuple[TreeStore, SessionTree]: + store = _tree_store() + tree = store.load() + tree.sync_history(history) + store.save(tree) + return store, tree def _help_entries() -> list[tuple[str, str]]: return [ - ( - "tree", - "Open a TUI to inspect, restore, fork, or summarize prior conversation points", - ), - ("fork", "Create a new branch/session from a previous user message"), - ("clone", "Duplicate the current active branch into a new autosave session"), + ("tree", "Navigate the persistent conversation tree and switch branches"), + ("fork", "Re-edit a previous user message to create a branch"), + ("clone", "Duplicate the active branch into a new autosave session"), ] -def _render_nodes(title: str, nodes: list[HistoryNode]) -> None: - lines = [title] - for ordinal, node in enumerate(nodes, start=1): - marker = "*" if node.is_system else " " - label = f" [{node.label}]" if node.label else "" - lines.append( - f" {ordinal}. [{node.node_id}] {marker}{node.role}: {node.preview}{label}" - ) - _emit_info("\n".join(lines)) - - def _tree_args(command: str) -> tuple[bool, str]: parts = command.split(maxsplit=2) if len(parts) >= 2 and parts[1].lower() in {"summary", "summarize"}: @@ -85,10 +65,6 @@ def _tree_args(command: str) -> tuple[bool, str]: return False, parts[1].strip() if len(parts) >= 2 else "" -def _selection_from_command(command: str) -> str: - return _tree_args(command)[1] - - def _markdown_result(prompt: str) -> Any: from code_puppy.plugins.customizable_commands.register_callbacks import ( MarkdownCommandResult, @@ -97,170 +73,168 @@ def _markdown_result(prompt: str) -> Any: return MarkdownCommandResult(prompt) -def _run_tree_menu(history: list[Any]) -> Any: +def _run_tree_menu(tree: SessionTree, *, user_only: bool = False) -> Any: + mode = FilterMode.USER_ONLY if user_only else FilterMode.DEFAULT with concurrent.futures.ThreadPoolExecutor() as executor: future = executor.submit( - lambda: asyncio.run( - TreeSelectionMenu(history=history, labels=_LABELS).run_async() - ) + lambda: asyncio.run(TreeSelectionMenu(tree=tree, mode=mode).run_async()) ) return future.result(timeout=300) def _resolve_or_menu( - command: str, history: list[Any] + command: str, tree: SessionTree, nodes: list[HistoryNode] ) -> tuple[HistoryNode | None, bool]: summarize, selection = _tree_args(command) if selection: - nodes = selectable_nodes(history) node = resolve_node(selection, nodes) if node is None: - _emit_warning(f"/tree: no history point matches '{selection}'") + _emit("warning", f"/tree: no unique history point matches '{selection}'") return node, summarize - - result = _run_tree_menu(history) + result = _run_tree_menu(tree) if result.cancelled: - _emit_warning("/tree: selection cancelled") return None, False return result.node, result.summarize -def _summarize_through(history: list[Any], node: HistoryNode) -> list[Any]: +def _summary_messages(history: list[Any], abandoned: list[Any]) -> list[Any]: + """Summarize only the abandoned tail and return generated context messages.""" from code_puppy.agents._compaction import _run_summarization_core - compacted, _summarized = _run_summarization_core( - history_through(history, node.index), + if not abandoned: + return [] + system = history[:1] + compacted, _ = _run_summarization_core( + [*system, *abandoned], protected_tokens=0, with_protection=False, model_name=None, ) - return list(compacted) + return list(compacted[len(system) :]) + + +def _navigate( + agent: Any, + store: TreeStore, + tree: SessionTree, + node: HistoryNode, + *, + summarize: bool, +) -> bool | Any: + old_leaf_id = tree.active_leaf_id + target_id = node.parent_id if node.is_userish else node.node_id + prompt = ( + message_text(tree.nodes[node.node_id].message).strip() + if node.is_userish + else "" + ) + target_history = tree.history_through(target_id) + if summarize: + try: + target_history.extend( + _summary_messages( + tree.history_through(old_leaf_id), + tree.abandoned_history(node.node_id), + ) + ) + except Exception as exc: # noqa: BLE001 - command boundary must fail soft + _emit("error", f"/tree: failed to summarize abandoned branch - {exc}") + return True -def _restore_summary(agent: Any, history: list[Any], node: HistoryNode) -> bool: try: - summarized_history = _summarize_through(history, node) - agent.set_message_history(summarized_history) + agent.set_message_history(target_history) + tree.sync_history(target_history) + store.save(tree) except Exception as exc: # noqa: BLE001 - _emit_error(f"/tree summary: failed to summarize history - {exc}") + _emit("error", f"/tree: failed to navigate - {exc}") return True - _emit_success( - f"Summarized conversation through {node.role} message {node.index} " - f"[{node.node_id}] into a fresh conversation ({len(summarized_history)} messages)." - ) + if node.is_userish: + _emit( + "success", "Moved before selected user message; edit it to create a branch." + ) + return _markdown_result(prompt) if prompt else True + _emit("success", f"Navigated to {node.role} message [{node.node_id}]") return True def _handle_tree(command: str) -> bool | Any: try: agent, history = _current_history() - except Exception as exc: # noqa: BLE001 - commands must fail soft. - _emit_error(f"/tree: could not read current history - {exc}") - return True - - nodes = selectable_nodes(history) - if not nodes: - _emit_warning("/tree: conversation history is empty - nothing to restore") - return True - - try: - node, summarize = _resolve_or_menu(command, history) + store, tree = _load_tree(history) except Exception as exc: # noqa: BLE001 - _emit_warning(f"/tree TUI failed, falling back to text view: {exc}") - _render_nodes("Conversation Tree:", build_nodes(history, _LABELS)) + _emit("error", f"/tree: could not read session tree - {exc}") return True - if node is None: + nodes = selectable_nodes(tree) + if not nodes: + _emit("warning", "/tree: conversation history is empty") return True - - if summarize: - return _restore_summary(agent, history, node) - try: - if node.is_userish: - prompt = message_text(history[node.index]).strip() - agent.set_message_history(history_before(history, node.index)) - _emit_success( - f"Moved to parent of {node.role} message {node.index} [{node.node_id}]. " - "Edit the restored prompt to create a branch." - ) - return _markdown_result(prompt) if prompt else True - - agent.set_message_history(history_through(history, node.index)) + node, summarize = _resolve_or_menu(command, tree, nodes) + # Label changes made in the selector survive selection and cancellation. + store.save(tree) except Exception as exc: # noqa: BLE001 - _emit_error(f"/tree: failed to restore history - {exc}") + _emit("error", f"/tree: selector failed - {exc}") return True - - _emit_success( - f"Restored conversation to {node.role} message {node.index} [{node.node_id}]" + return ( + True + if node is None + else _navigate(agent, store, tree, node, summarize=summarize) ) - return True def _handle_fork(command: str) -> bool | Any: try: agent, history = _current_history() + store, tree = _load_tree(history) except Exception as exc: # noqa: BLE001 - _emit_error(f"/fork: could not read current history - {exc}") + _emit("error", f"/fork: could not read session tree - {exc}") return True - - nodes = user_nodes(history) + nodes = user_nodes(tree) if not nodes: - _emit_warning("/fork: no previous user messages are available to fork") - return True - - selection = _selection_from_command(command) - if not selection: - _render_nodes("Forkable User Messages:", nodes) - _emit_warning("Usage: /fork ") - return True - - node = resolve_node(selection, nodes) - if node is None: - _emit_warning(f"/fork: no user message matches '{selection}'") + _emit("warning", "/fork: no previous user messages are available") return True - - prompt = message_text(history[node.index]).strip() - try: - agent.set_message_history(history_before(history, node.index)) - except Exception as exc: # noqa: BLE001 - _emit_error(f"/fork: failed to create branch history - {exc}") + selection = command.split(maxsplit=1)[1].strip() if " " in command else "" + if selection: + node = resolve_node(selection, nodes) + else: + try: + result = _run_tree_menu(tree, user_only=True) + node = result.node if not result.cancelled else None + except Exception as exc: # noqa: BLE001 + _emit("error", f"/fork: selector failed - {exc}") + return True + if node is None or not node.is_userish: + if selection: + _emit("warning", f"/fork: no unique user message matches '{selection}'") return True - - _emit_success( - f"Created branch before user message {node.index} [{node.node_id}]. " - "Edit the restored prompt to continue from this branch." - ) - return _markdown_result(prompt) if prompt else True + return _navigate(agent, store, tree, node, summarize=False) def _handle_clone() -> bool: try: agent, history = _current_history() if not history: - _emit_warning("/clone: no active branch to clone") + _emit("warning", "/clone: no active branch to clone") return True - - from code_puppy.config import AUTOSAVE_DIR, rotate_autosave_id + from code_puppy.config import AUTOSAVE_DIR from code_puppy.session_storage import save_session - session_id = rotate_autosave_id() - session_name = f"autosave_{session_id}" + # A clone is a new saved artifact, not a mutation of the current session identity. + clone_id = datetime.now().strftime("%Y%m%d_%H%M%S_%f") metadata = save_session( history=history, - session_name=session_name, + session_name=f"auto_session_{clone_id}", base_dir=Path(AUTOSAVE_DIR), timestamp=datetime.now().isoformat(), token_estimator=agent.estimate_tokens_for_message, auto_saved=True, ) - _emit_success( - f"Cloned current active branch to new autosave session {session_id} " - f"({metadata.message_count} messages)." - ) + _emit("success", f"Cloned active branch to {metadata.session_name}.") except Exception as exc: # noqa: BLE001 - _emit_error(f"/clone: failed to clone active branch - {exc}") + _emit("error", f"/clone: failed to clone active branch - {exc}") return True @@ -277,10 +251,4 @@ def _handle_custom_command(command: str, name: str) -> bool | str | None: register_callback("custom_command_help", _help_entries) register_callback("custom_command", _handle_custom_command) - -__all__ = [ - "_handle_custom_command", - "_handle_fork", - "_handle_tree", - "_help_entries", -] +__all__ = ["_handle_custom_command", "_handle_fork", "_handle_tree", "_help_entries"] diff --git a/code_puppy/plugins/session_tree/tree_menu.py b/code_puppy/plugins/session_tree/tree_menu.py index 689837031..0855cfd68 100644 --- a/code_puppy/plugins/session_tree/tree_menu.py +++ b/code_puppy/plugins/session_tree/tree_menu.py @@ -4,8 +4,6 @@ import sys from dataclasses import dataclass, field -from datetime import datetime -from typing import Any from prompt_toolkit import Application from prompt_toolkit.key_binding import KeyBindings @@ -26,6 +24,7 @@ build_nodes, visible_nodes, ) +from .tree_store import SessionTree TREE_PAGE_SIZE = 14 @@ -40,25 +39,25 @@ class TreeMenuResult: @dataclass(slots=True) class TreeSelectionMenu: - history: list[Any] - labels: dict[str, str] = field(default_factory=dict) + tree: SessionTree mode: FilterMode = FilterMode.DEFAULT selected_index: int = 0 page: int = 0 page_size: int = TREE_PAGE_SIZE result: TreeMenuResult = field(default_factory=TreeMenuResult) - show_label_timestamps: bool = True + show_label_timestamps: bool = False + search_query: str = "" def __post_init__(self) -> None: self._clamp_selection() @property def nodes(self) -> list[HistoryNode]: - return build_nodes(self.history, self.labels) + return build_nodes(self.tree) @property def visible(self) -> list[HistoryNode]: - return visible_nodes(self.nodes, self.mode) + return visible_nodes(self.nodes, self.mode, self.search_query) @property def total_pages(self) -> int: @@ -121,24 +120,25 @@ def _toggle_label(self) -> None: selected = self.selected_node() if selected is None: return - if selected.node_id in self.labels: - del self.labels[selected.node_id] - return - suffix = ( - f" @ {datetime.now().strftime('%H:%M')}" - if self.show_label_timestamps - else "" - ) - self.labels[selected.node_id] = f"label{suffix}" + self.tree.set_label(selected.node_id, None if selected.label else "label") def _branch_prefix(self, node: HistoryNode) -> str: - if node.index == len(self.history) - 1: - return "└─" - return "├─" + ancestors = "".join( + "│ " if has_next else " " for has_next in node.ancestor_has_next + ) + if node.depth == 0: + return "" + return f"{ancestors}{'└─ ' if node.is_last else '├─ '}" def _render(self): lines = [("bold cyan", " Session Tree")] lines.append(("fg:ansibrightblack", f"\n Filter: {self.mode.value}")) + lines.append( + ( + "fg:ansibrightblack", + f" Search: {self.search_query or '(type to search)'}", + ) + ) if self.total_pages > 1: lines.append( ("fg:ansibrightblack", f" Page {self.page + 1}/{self.total_pages}") @@ -150,15 +150,19 @@ def _render(self): for offset, node in enumerate(self.page_nodes): absolute_index = self.page_start + offset is_selected = absolute_index == self.selected_index - is_active = node.index == len(self.history) - 1 + is_active = node.is_active cursor = " › " if is_selected else " " style = "fg:ansiwhite bold" if is_selected else "fg:ansibrightblack" - label = f" [{node.label}]" if node.label else "" + timestamp = "" + if self.show_label_timestamps and node.label_timestamp: + timestamp = f" @{node.label_timestamp[11:16]}" + label = f" [{node.label}{timestamp}]" if node.label else "" active = " ← active" if is_active else "" + path_marker = "• " if node.is_on_active_path else " " lines.append( ( style, - f'{cursor}{self._branch_prefix(node)} {node.role}: "{node.preview}"', + f'{cursor}{self._branch_prefix(node)}{path_marker}{node.role}: "{node.preview}"', ) ) lines.append(("fg:ansiblue", f" #{node.node_id}{label}{active}\n")) @@ -243,9 +247,32 @@ def _(event): self.result.summarize = True event.app.exit() + @kb.add("backspace") + def _(event): + if self.search_query: + self.search_query = self.search_query[:-1] + self.selected_index = 0 + self._clamp_selection() + refresh(event) + + @kb.add("") + def _(event): + text = event.data + if text and text.isprintable(): + self.search_query += text + self.selected_index = 0 + self._clamp_selection() + refresh(event) + @kb.add("escape") @kb.add("c-c") def _(event): + if self.search_query: + self.search_query = "" + self.selected_index = 0 + self._clamp_selection() + refresh(event) + return self.result.cancelled = True event.app.exit() diff --git a/code_puppy/plugins/session_tree/tree_model.py b/code_puppy/plugins/session_tree/tree_model.py index 0fa40840e..6bd86090f 100644 --- a/code_puppy/plugins/session_tree/tree_model.py +++ b/code_puppy/plugins/session_tree/tree_model.py @@ -1,12 +1,13 @@ -"""Small helpers for navigating Code Puppy message history.""" +"""View-model helpers for a persistent conversation tree.""" from __future__ import annotations -import hashlib from dataclasses import dataclass from enum import StrEnum from typing import Any, Iterable, Sequence +from .tree_store import SessionTree, StoredNode + class FilterMode(StrEnum): DEFAULT = "default" @@ -23,12 +24,18 @@ class FilterMode(StrEnum): @dataclass(frozen=True, slots=True) class HistoryNode: - index: int node_id: str + parent_id: str | None role: str preview: str + depth: int + is_last: bool + ancestor_has_next: tuple[bool, ...] is_system: bool = False label: str = "" + label_timestamp: str | None = None + is_active: bool = False + is_on_active_path: bool = False @property def is_userish(self) -> bool: @@ -39,6 +46,14 @@ def is_toolish(self) -> bool: return self.role.lower() in TOOL_ROLES +@dataclass(frozen=True, slots=True) +class _FlatNode: + stored: StoredNode + depth: int + is_last: bool + ancestor_has_next: tuple[bool, ...] + + def message_role(message: Any) -> str: role = getattr(message, "role", None) if isinstance(role, str): @@ -55,11 +70,9 @@ def message_text(message: Any) -> str: content = getattr(message, "content", None) if isinstance(content, str): return content - parts = getattr(message, "parts", None) if parts is None: parts = content if isinstance(content, list) else [] - chunks: list[str] = [] for part in parts or []: text = getattr(part, "content", None) @@ -69,95 +82,97 @@ def message_text(message: Any) -> str: text = part.get("content") or part.get("text") if isinstance(text, str): chunks.append(text) - - if chunks: - return " ".join(chunks) - return str(message) - - -def node_id_for(index: int, message: Any) -> str: - raw = f"{index}:{type(message).__name__}:{message_text(message)}".encode() - return hashlib.sha1(raw).hexdigest()[:8] - - -def _preview(text: str, max_len: int = 72) -> str: - clean = " ".join(text.split()) - if not clean: - clean = "(no text)" - if len(clean) <= max_len: - return clean - return clean[: max_len - 1] + "…" + return " ".join(chunks) if chunks else str(message) + + +def _preview(text: str, max_len: int = 96) -> str: + clean = " ".join(text.split()) or "(no text)" + return clean if len(clean) <= max_len else f"{clean[: max_len - 1]}…" + + +def _flatten(tree: SessionTree) -> list[_FlatNode]: + result: list[_FlatNode] = [] + roots = tree.children(None) + stack: list[tuple[StoredNode, int, bool, tuple[bool, ...]]] = [] + for index in range(len(roots) - 1, -1, -1): + stack.append((roots[index], 0, index == len(roots) - 1, ())) + while stack: + node, depth, is_last, ancestors = stack.pop() + result.append(_FlatNode(node, depth, is_last, ancestors)) + children = tree.children(node.node_id) + for index in range(len(children) - 1, -1, -1): + child_is_last = index == len(children) - 1 + stack.append( + (children[index], depth + 1, child_is_last, (*ancestors, not is_last)) + ) + return result -def build_nodes( - history: Sequence[Any], labels: dict[str, str] | None = None -) -> list[HistoryNode]: - labels = labels or {} +def build_nodes(tree: SessionTree) -> list[HistoryNode]: + active_path = set(tree.path_ids(tree.active_leaf_id)) nodes: list[HistoryNode] = [] - for index, message in enumerate(history): - role = message_role(message) - node_id = node_id_for(index, message) + for item in _flatten(tree): + role = message_role(item.stored.message) nodes.append( HistoryNode( - index=index, - node_id=node_id, + node_id=item.stored.node_id, + parent_id=item.stored.parent_id, role=role, - preview=_preview(message_text(message)), - is_system=index == 0 and role == "system", - label=labels.get(node_id, ""), + preview=_preview(message_text(item.stored.message)), + depth=item.depth, + is_last=item.is_last, + ancestor_has_next=item.ancestor_has_next, + is_system=item.stored.parent_id is None and role == "system", + label=item.stored.label, + label_timestamp=item.stored.label_timestamp, + is_active=item.stored.node_id == tree.active_leaf_id, + is_on_active_path=item.stored.node_id in active_path, ) ) return nodes -def selectable_nodes(history: Sequence[Any]) -> list[HistoryNode]: - return [node for node in build_nodes(history) if not node.is_system] +def selectable_nodes(tree: SessionTree) -> list[HistoryNode]: + return [node for node in build_nodes(tree) if not node.is_system] -def user_nodes(history: Sequence[Any]) -> list[HistoryNode]: - return [node for node in selectable_nodes(history) if node.is_userish] +def user_nodes(tree: SessionTree) -> list[HistoryNode]: + return [node for node in selectable_nodes(tree) if node.is_userish] -def visible_nodes(nodes: Sequence[HistoryNode], mode: FilterMode) -> list[HistoryNode]: +def visible_nodes( + nodes: Sequence[HistoryNode], mode: FilterMode, query: str = "" +) -> list[HistoryNode]: if mode == FilterMode.ALL: - return list(nodes) - if mode == FilterMode.NO_TOOLS: - return [node for node in nodes if not node.is_toolish] - if mode == FilterMode.USER_ONLY: - return [node for node in nodes if node.is_userish] - if mode == FilterMode.LABELED_ONLY: - return [node for node in nodes if node.label] - return [node for node in nodes if not node.is_system and not node.is_toolish] + visible = list(nodes) + elif mode == FilterMode.NO_TOOLS: + visible = [node for node in nodes if not node.is_toolish] + elif mode == FilterMode.USER_ONLY: + visible = [node for node in nodes if node.is_userish] + elif mode == FilterMode.LABELED_ONLY: + visible = [node for node in nodes if node.label] + else: + visible = [node for node in nodes if not node.is_system and not node.is_toolish] + tokens = query.lower().split() + if not tokens: + return visible + return [ + node + for node in visible + if all( + token in f"{node.role} {node.preview} {node.label}".lower() + for token in tokens + ) + ] def resolve_node(selection: str, nodes: Iterable[HistoryNode]) -> HistoryNode | None: value = selection.strip() if not value: return None - node_list = list(nodes) - if value.isdigit(): - ordinal = int(value) - if 1 <= ordinal <= len(node_list): - return node_list[ordinal - 1] - for node in node_list: - if node.index == ordinal: - return node - - value_lower = value.lower() - for node in node_list: - if node.node_id.startswith(value_lower): - return node - return None - - -def history_through(history: Sequence[Any], index: int) -> list[Any]: - if index < 0: - return [] - return list(history[: index + 1]) - - -def history_before(history: Sequence[Any], index: int) -> list[Any]: - if index <= 0: - return list(history[:1]) if history else [] - return list(history[:index]) + if value.isdigit() and 1 <= int(value) <= len(node_list): + return node_list[int(value) - 1] + lowered = value.lower() + matches = [node for node in node_list if node.node_id.startswith(lowered)] + return matches[0] if len(matches) == 1 else None diff --git a/code_puppy/plugins/session_tree/tree_store.py b/code_puppy/plugins/session_tree/tree_store.py new file mode 100644 index 000000000..c9b9044b5 --- /dev/null +++ b/code_puppy/plugins/session_tree/tree_store.py @@ -0,0 +1,135 @@ +"""Persistent append-only conversation graph for the session tree plugin.""" + +from __future__ import annotations + +import hashlib +import pickle +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Sequence + + +def _message_fingerprint(message: Any) -> str: + """Return a stable-enough identity for matching a message on one path.""" + try: + payload = pickle.dumps(message) + except Exception: # pragma: no cover - exotic third-party messages + payload = repr(message).encode("utf-8", errors="replace") + return hashlib.sha256(payload).hexdigest() + + +@dataclass(slots=True) +class StoredNode: + node_id: str + parent_id: str | None + message: Any + fingerprint: str + sequence: int + label: str = "" + label_timestamp: str | None = None + + +@dataclass(slots=True) +class SessionTree: + """All known messages and branches for one Code Puppy autosave session.""" + + nodes: dict[str, StoredNode] = field(default_factory=dict) + active_leaf_id: str | None = None + next_sequence: int = 0 + + def sync_history(self, history: Sequence[Any]) -> str | None: + """Merge a live linear history into the graph and mark its final node active.""" + parent_id: str | None = None + for message in history: + fingerprint = _message_fingerprint(message) + matching = next( + ( + node + for node in self.nodes.values() + if node.parent_id == parent_id and node.fingerprint == fingerprint + ), + None, + ) + if matching is None: + matching = StoredNode( + node_id=uuid.uuid4().hex[:12], + parent_id=parent_id, + message=message, + fingerprint=fingerprint, + sequence=self.next_sequence, + ) + self.nodes[matching.node_id] = matching + self.next_sequence += 1 + else: + # Keep the newest object shape after dependency upgrades/deserialization. + matching.message = message + parent_id = matching.node_id + self.active_leaf_id = parent_id + return parent_id + + def children(self, parent_id: str | None) -> list[StoredNode]: + return sorted( + (node for node in self.nodes.values() if node.parent_id == parent_id), + key=lambda node: node.sequence, + ) + + def path_ids(self, node_id: str | None) -> list[str]: + path: list[str] = [] + seen: set[str] = set() + while node_id is not None and node_id not in seen: + seen.add(node_id) + node = self.nodes.get(node_id) + if node is None: + break + path.append(node_id) + node_id = node.parent_id + path.reverse() + return path + + def history_through(self, node_id: str | None) -> list[Any]: + return [self.nodes[item].message for item in self.path_ids(node_id)] + + def common_ancestor(self, left_id: str | None, right_id: str | None) -> str | None: + common: str | None = None + for left, right in zip(self.path_ids(left_id), self.path_ids(right_id)): + if left != right: + break + common = left + return common + + def abandoned_history(self, target_id: str) -> list[Any]: + ancestor = self.common_ancestor(self.active_leaf_id, target_id) + active_path = self.path_ids(self.active_leaf_id) + start = active_path.index(ancestor) + 1 if ancestor in active_path else 0 + return [self.nodes[item].message for item in active_path[start:]] + + def set_label(self, node_id: str, label: str | None) -> None: + node = self.nodes[node_id] + node.label = (label or "").strip() + node.label_timestamp = ( + datetime.now(timezone.utc).isoformat() if node.label else None + ) + + +class TreeStore: + """Atomic pickle persistence for a session graph.""" + + def __init__(self, path: Path) -> None: + self.path = path + + def load(self) -> SessionTree: + if not self.path.exists(): + return SessionTree() + try: + value = pickle.loads(self.path.read_bytes()) # noqa: S301 + except Exception: + return SessionTree() + return value if isinstance(value, SessionTree) else SessionTree() + + def save(self, tree: SessionTree) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + temporary = self.path.with_suffix(".tmp") + temporary.write_bytes(pickle.dumps(tree)) + temporary.replace(self.path) diff --git a/tests/plugins/test_session_tree.py b/tests/plugins/test_session_tree.py index 21a861ab7..d3a9fec87 100644 --- a/tests/plugins/test_session_tree.py +++ b/tests/plugins/test_session_tree.py @@ -1,12 +1,22 @@ -"""Tests for the /tree, /fork, and /clone session navigation plugin.""" +"""Tests for persistent /tree, /fork, and /clone navigation.""" from __future__ import annotations import importlib import sys +from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch +from code_puppy.plugins.session_tree.tree_model import ( + FilterMode, + build_nodes, + resolve_node, + selectable_nodes, + visible_nodes, +) +from code_puppy.plugins.session_tree.tree_store import SessionTree, TreeStore + class Message(SimpleNamespace): pass @@ -17,14 +27,6 @@ def _plugin_module(): return importlib.import_module("code_puppy.plugins.session_tree.register_callbacks") -def _tree_model(): - return importlib.import_module("code_puppy.plugins.session_tree.tree_model") - - -def _agent_manager_module(agent: MagicMock) -> SimpleNamespace: - return SimpleNamespace(get_current_agent=lambda: agent) - - def _history() -> list[Message]: return [ Message(role="system", content="system prompt"), @@ -36,282 +38,220 @@ def _history() -> list[Message]: ] -def test_build_nodes_creates_stable_tree_entries(): - nodes = _tree_model().build_nodes(_history()) - - assert len(nodes) == 6 - assert nodes[0].is_system is True - assert nodes[1].role == "user" - assert nodes[1].preview == "first task" - assert len(nodes[1].node_id) == 8 - +def _tree(history=None) -> SessionTree: + tree = SessionTree() + tree.sync_history(history or _history()) + return tree -def test_visible_nodes_supports_filter_modes_and_labels(): - model = _tree_model() - history = _history() - base_nodes = model.build_nodes(history) - labels = {base_nodes[2].node_id: "keeper"} - nodes = model.build_nodes(history, labels) - assert all( - not node.is_toolish - for node in model.visible_nodes(nodes, model.FilterMode.NO_TOOLS) +def _command_context(tmp_path: Path, history: list[Message]): + agent = MagicMock() + agent.get_message_history.return_value = history + manager = SimpleNamespace(get_current_agent=lambda: agent) + config = SimpleNamespace( + AUTOSAVE_DIR=str(tmp_path), get_current_autosave_id=lambda: "current" + ) + return agent, patch.dict( + sys.modules, + { + "code_puppy.agents.agent_manager": manager, + "code_puppy.config": config, + }, ) - assert { - node.role for node in model.visible_nodes(nodes, model.FilterMode.USER_ONLY) - } == {"user"} - assert model.visible_nodes(nodes, model.FilterMode.LABELED_ONLY) == [nodes[2]] - assert len(model.visible_nodes(nodes, model.FilterMode.ALL)) == len(nodes) - -def test_resolve_node_accepts_ordinal_index_and_id_prefix(): - model = _tree_model() - nodes = model.selectable_nodes(_history()) - assert model.resolve_node("1", nodes).preview == "first task" - assert model.resolve_node("4", nodes).preview == "second task" - assert model.resolve_node(nodes[1].node_id[:4], nodes) == nodes[1] - assert model.resolve_node("nope", nodes) is None +def test_sync_history_is_stable_and_builds_real_branches(): + history = _history() + tree = _tree(history) + original_ids = tree.path_ids(tree.active_leaf_id) + tree.sync_history(history) + assert tree.path_ids(tree.active_leaf_id) == original_ids -def test_history_helpers_preserve_system_prompt_for_forks(): - model = _tree_model() - history = _history() + tree.sync_history([*history[:4], Message(role="user", content="alternate")]) + branch_parent = original_ids[3] + assert len(tree.children(branch_parent)) == 2 + assert len(tree.nodes) == len(history) + 1 - assert model.history_through(history, 2) == history[:3] - assert model.history_before(history, 1) == history[:1] +def test_duplicate_messages_on_one_path_remain_distinct(): + repeated = Message(role="user", content="same") + tree = _tree([repeated, repeated]) + path = tree.path_ids(tree.active_leaf_id) -def test_help_entries_include_tree_fork_clone_and_summary_language(): - entries = dict(_plugin_module()._help_entries()) + assert len(path) == 2 + assert path[0] != path[1] - assert "summarize" in entries["tree"] - assert entries["fork"] - assert entries["clone"] +def test_store_persists_branches_labels_and_active_leaf(tmp_path): + store = TreeStore(tmp_path / "tree.pkl") + tree = _tree() + selected = selectable_nodes(tree)[0] + tree.set_label(selected.node_id, "keeper") + store.save(tree) -def test_handle_custom_command_ignores_unknown_command(): - assert _plugin_module()._handle_custom_command("/nope", "nope") is None + restored = store.load() + assert restored.nodes[selected.node_id].label == "keeper" + assert restored.active_leaf_id == tree.active_leaf_id -def test_tree_restores_non_user_history_point_from_argument(): +def test_model_marks_active_path_and_renders_branch_depth(): history = _history() - agent = MagicMock() - agent.get_message_history.return_value = history + tree = _tree(history) + first_path = tree.path_ids(tree.active_leaf_id) + tree.sync_history([*history[:3], Message(role="user", content="alternate")]) + nodes = build_nodes(tree) - with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), - ): - result = _plugin_module()._handle_custom_command("/tree 2", "tree") + assert sum(node.is_active for node in nodes) == 1 + assert all( + node.is_on_active_path + for node in nodes + if node.node_id in tree.path_ids(tree.active_leaf_id) + ) + assert any(node.depth > 0 and not node.is_on_active_path for node in nodes) + assert first_path[-1] in tree.nodes - assert result is True - agent.set_message_history.assert_called_once_with(history[:3]) +def test_filters_search_and_unique_resolution(): + tree = _tree() + nodes = build_nodes(tree) + assistant = next(node for node in nodes if node.role == "assistant") + tree.set_label(assistant.node_id, "keeper") + nodes = build_nodes(tree) -def test_tree_summary_selection_replaces_history_with_summarized_conversation(): - history = _history() - summarized = [history[0], Message(role="user", content="Summary: compacted")] - agent = MagicMock() - agent.get_message_history.return_value = history + assert {node.role for node in visible_nodes(nodes, FilterMode.USER_ONLY)} == { + "user" + } + assert all( + not node.is_toolish for node in visible_nodes(nodes, FilterMode.NO_TOOLS) + ) + assert [node.node_id for node in visible_nodes(nodes, FilterMode.LABELED_ONLY)] == [ + assistant.node_id + ] + assert ( + visible_nodes(nodes, FilterMode.ALL, "first answer")[0].node_id + == assistant.node_id + ) + selectable = selectable_nodes(tree) + assert resolve_node("1", selectable) == selectable[0] + assert resolve_node(assistant.node_id[:6], selectable).node_id == assistant.node_id - with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._summarize_through", - return_value=summarized, - ) as summarize, - patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), - ): - result = _plugin_module()._handle_custom_command("/tree summary 2", "tree") - assert result is True - summarize.assert_called_once() - agent.set_message_history.assert_called_once_with(summarized) +def test_abandoned_history_starts_after_common_ancestor(): + history = _history() + tree = _tree(history) + old_leaf = tree.active_leaf_id + target = tree.path_ids(old_leaf)[2] + abandoned = tree.abandoned_history(target) + assert abandoned == history[3:] -def test_tree_summary_from_tui_result_replaces_history(): + +def test_tree_user_selection_restores_parent_and_preserves_old_branch(tmp_path): plugin = _plugin_module() history = _history() - summarized = [history[0], Message(role="user", content="Summary: compacted")] - agent = MagicMock() - agent.get_message_history.return_value = history - node = _tree_model().selectable_nodes(history)[1] - result_obj = SimpleNamespace(node=node, cancelled=False, summarize=True) + agent, context = _command_context(tmp_path, history) + with context, patch.object(plugin, "_emit"): + result = plugin._handle_custom_command("/tree 4", "tree") - with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._run_tree_menu", - return_value=result_obj, - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._summarize_through", - return_value=summarized, - ), - patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), - ): - result = plugin._handle_custom_command("/tree", "tree") - - assert result is True - agent.set_message_history.assert_called_once_with(summarized) + assert str(result) == "second task" + agent.set_message_history.assert_called_once_with(history[:4]) + stored = TreeStore(tmp_path / "session_trees/current.pkl").load() + assert len(stored.nodes) == len(history) + assert stored.history_through(stored.active_leaf_id) == history[:4] -def test_tree_summary_failure_does_not_mutate_history(): +def test_new_turn_after_navigation_creates_sibling_branch(tmp_path): + plugin = _plugin_module() history = _history() - agent = MagicMock() - agent.get_message_history.return_value = history + agent, context = _command_context(tmp_path, history) + with context, patch.object(plugin, "_emit"): + plugin._handle_custom_command("/tree 4", "tree") - with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._summarize_through", - side_effect=RuntimeError("no biscuits"), - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._emit_error" - ) as error, - ): - result = _plugin_module()._handle_custom_command("/tree summarize 2", "tree") + alternate = [*history[:4], Message(role="user", content="replacement")] + agent, context = _command_context(tmp_path, alternate) + with context, patch.object(plugin, "_emit"): + plugin._handle_custom_command("/tree nope", "tree") - assert result is True - agent.set_message_history.assert_not_called() - assert "failed to summarize" in str(error.call_args) + stored = TreeStore(tmp_path / "session_trees/current.pkl").load() + parent = stored.path_ids(stored.active_leaf_id)[3] + assert len(stored.children(parent)) == 2 -def test_tree_user_selection_returns_prompt_and_restores_parent(): +def test_tree_non_user_selection_restores_through_selected_node(tmp_path): + plugin = _plugin_module() history = _history() - agent = MagicMock() - agent.get_message_history.return_value = history - - with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), - ): - result = _plugin_module()._handle_custom_command("/tree 4", "tree") - - assert str(result) == "second task" - agent.set_message_history.assert_called_once_with(history[:4]) + agent, context = _command_context(tmp_path, history) + with context, patch.object(plugin, "_emit"): + result = plugin._handle_custom_command("/tree 2", "tree") + assert result is True + agent.set_message_history.assert_called_once_with(history[:3]) -def test_tree_warns_for_empty_or_system_only_history(): - agent = MagicMock() - agent.get_message_history.return_value = [Message(role="system", content="system")] +def test_summary_receives_only_abandoned_tail_and_attaches_at_target(tmp_path): + plugin = _plugin_module() + history = _history() + summary = Message(role="user", content="branch summary") + agent, context = _command_context(tmp_path, history) with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._emit_warning" - ) as warning, + context, + patch.object(plugin, "_emit"), + patch.object(plugin, "_summary_messages", return_value=[summary]) as summarize, ): - result = _plugin_module()._handle_custom_command("/tree", "tree") + plugin._handle_custom_command("/tree summary 2", "tree") - assert result is True - agent.set_message_history.assert_not_called() - assert "nothing to restore" in str(warning.call_args) + assert summarize.call_args.args[1] == history[3:] + agent.set_message_history.assert_called_once_with([*history[:3], summary]) -def test_tree_warns_for_invalid_selection(): +def test_summary_failure_does_not_mutate_history(tmp_path): + plugin = _plugin_module() history = _history() - agent = MagicMock() - agent.get_message_history.return_value = history - + agent, context = _command_context(tmp_path, history) with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, + context, + patch.object(plugin, "_emit"), + patch.object( + plugin, "_summary_messages", side_effect=RuntimeError("no biscuits") ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._emit_warning" - ) as warning, ): - result = _plugin_module()._handle_custom_command("/tree nope", "tree") + result = plugin._handle_custom_command("/tree summary 2", "tree") assert result is True agent.set_message_history.assert_not_called() - assert "no history point matches" in str(warning.call_args) -def test_fork_returns_selected_prompt_and_trims_history_before_it(): +def test_fork_without_argument_uses_tree_selector(tmp_path): + plugin = _plugin_module() history = _history() - agent = MagicMock() - agent.get_message_history.return_value = history + agent, context = _command_context(tmp_path, history) + + def select_last_user(tree, **_kwargs): + selected = [node for node in selectable_nodes(tree) if node.is_userish][-1] + return SimpleNamespace(node=selected, cancelled=False) with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), + context, + patch.object(plugin, "_emit"), + patch.object(plugin, "_run_tree_menu", side_effect=select_last_user), ): - result = _plugin_module()._handle_custom_command("/fork 2", "fork") + result = plugin._handle_custom_command("/fork", "fork") assert str(result) == "second task" agent.set_message_history.assert_called_once_with(history[:4]) -def test_fork_warns_when_no_user_messages_exist(): - agent = MagicMock() - agent.get_message_history.return_value = [ - Message(role="system", content="system"), - Message(role="assistant", content="hello"), - ] - - with ( - patch.dict( - sys.modules, - {"code_puppy.agents.agent_manager": _agent_manager_module(agent)}, - ), - patch( - "code_puppy.plugins.session_tree.register_callbacks._emit_warning" - ) as warning, - ): - result = _plugin_module()._handle_custom_command("/fork", "fork") - - assert result is True - agent.set_message_history.assert_not_called() - assert "no previous user messages" in str(warning.call_args) - - -def test_clone_saves_active_branch_to_new_autosave_session(tmp_path): +def test_clone_does_not_rotate_current_session_id(tmp_path): + plugin = _plugin_module() history = _history() - agent = MagicMock() - agent.get_message_history.return_value = history + agent, context = _command_context(tmp_path, history) agent.estimate_tokens_for_message.return_value = 1 + with context, patch.object(plugin, "_emit"): + assert plugin._handle_custom_command("/clone", "clone") is True - config = SimpleNamespace( - AUTOSAVE_DIR=str(tmp_path), rotate_autosave_id=lambda: "abc123" - ) - with ( - patch.dict( - sys.modules, - { - "code_puppy.agents.agent_manager": _agent_manager_module(agent), - "code_puppy.config": config, - }, - ), - patch("code_puppy.plugins.session_tree.register_callbacks._emit_success"), - ): - result = _plugin_module()._handle_custom_command("/clone", "clone") + assert len(list(tmp_path.glob("auto_session_*.pkl"))) == 1 - assert result is True - assert (tmp_path / "autosave_abc123.pkl").exists() - assert (tmp_path / "autosave_abc123_meta.json").exists() + +def test_unknown_command_is_ignored(): + assert _plugin_module()._handle_custom_command("/nope", "nope") is None