Preserve history session boundaries across list mutations

CHANGELOG.md

@@ -34,6 +34,9 @@
       it directly would then have retained current-session entries hidden from unbounded ranges and
       searches. Interactive `history` use was unaffected, since `cmd2` only truncates when
       persisting history at exit.
+    - Fixed deleting, popping, removing, inserting, or slice-assigning `History` entries before the
+      current session leaving `session_start_index` stale. In-place filtering recognizes retained
+      persisted entries by identity, so equal commands from the current session are not hidden.
     - Fixed `set_window_title()` leaving title changes buffered until the next prompt render. The
       title is now flushed before the method returns when no prompt is running. While a prompt or
       `select()` menu is running, it is written on that prompt's event loop so it doesn't interleave

cmd2/history.py

@@ -1,6 +1,7 @@
 """History management classes."""
 
 import json
+import operator
 import re
 from collections.abc import (
     Callable,
@@ -9,6 +10,9 @@
 from dataclasses import dataclass
 from typing import (
     Any,
+    Self,
+    SupportsIndex,
+    cast,
     overload,
 )
 
@@ -167,6 +171,79 @@ def __init__(self, seq: Iterable[HistoryItem] = ()) -> None:
         super().__init__(seq)
         self.session_start_index = 0
 
+    def __delitem__(self, key: SupportsIndex | slice, /) -> None:
+        """Delete entries and move the session boundary past any removed persisted entries."""
+        if isinstance(key, slice):
+            removed_before = sum(index < self.session_start_index for index in range(*key.indices(len(self))))
+        else:
+            key = operator.index(key)
+            position = key if key >= 0 else len(self) + key
+            removed_before = int(position < self.session_start_index)
+        super().__delitem__(key)
+        self.session_start_index -= removed_before
+
+    @overload
+    def __setitem__(self, key: SupportsIndex, value: HistoryItem, /) -> None: ...  # pragma: no cover
+
+    @overload
+    def __setitem__(self, key: slice, value: Iterable[HistoryItem], /) -> None: ...  # pragma: no cover
+
+    def __setitem__(self, key: SupportsIndex | slice, value: HistoryItem | Iterable[HistoryItem], /) -> None:
+        """Replace entries, retaining persisted item identities when a slice spans the session boundary."""
+        if not isinstance(key, slice):
+            super().__setitem__(key, cast(HistoryItem, value))
+            return
+        start, stop, step = key.indices(len(self))
+        replacement = list(cast(Iterable[HistoryItem], value))
+        persisted_ids = (
+            {id(entry) for entry in self[start : self.session_start_index]}
+            if step == 1 and start < self.session_start_index < stop
+            else None
+        )
+        super().__setitem__(key, replacement)
+        if persisted_ids is not None:
+            boundary = start
+            for entry in replacement:
+                if id(entry) not in persisted_ids:
+                    break
+                boundary += 1
+            self.session_start_index = boundary
+        elif step == 1 and start < self.session_start_index:
+            stop = max(start, stop)
+            self.session_start_index += len(replacement) - (stop - start)
+
+    def insert(self, index: SupportsIndex, value: HistoryItem, /) -> None:
+        """Insert an entry, shifting the session boundary if the insertion precedes it."""
+        index = operator.index(index)
+        position = min(len(self), max(0, index if index >= 0 else len(self) + index))
+        super().insert(index, value)
+        if position < self.session_start_index:
+            self.session_start_index += 1
+
+    def pop(self, index: SupportsIndex = -1, /) -> HistoryItem:
+        """Remove and return an entry, retaining the session boundary."""
+        index = operator.index(index)
+        position = index if index >= 0 else len(self) + index
+        result = super().pop(index)
+        if position < self.session_start_index:
+            self.session_start_index -= 1
+        return result
+
+    def remove(self, value: HistoryItem, /) -> None:
+        """Remove the first matching entry, retaining the session boundary."""
+        position = super().index(value)
+        super().pop(position)
+        if position < self.session_start_index:
+            self.session_start_index -= 1
+
+    def __imul__(self, count: SupportsIndex, /) -> Self:
+        """Repeat entries in place, resetting the boundary if repetition clears the list."""
+        count = operator.index(count)
+        super().__imul__(count)
+        if count <= 0:
+            self.session_start_index = 0
+        return self
+
     def start_session(self) -> None:
         """Start a new session, thereby setting the next index as the first index in the new session."""
         self.session_start_index = len(self)
@@ -331,8 +408,6 @@ def truncate(self, max_length: int) -> None:
         elif len(self) > max_length:
             last_element = len(self) - max_length
             del self[0:last_element]
-            # keep the session boundary pointing at the same retained items
-            self.session_start_index = max(0, self.session_start_index - last_element)
 
     def _build_result_dictionary(
         self, start: int, end: int, filter_func: Callable[[HistoryItem], bool] | None = None

tests/test_history_mutations.py

@@ -0,0 +1,251 @@
+"""Session selection survives structural changes to the public history list."""
+
+import pytest
+
+from cmd2.history import History, HistoryItem
+from cmd2.parsing import Statement
+
+
+def item(raw):
+    return HistoryItem(Statement("", raw=raw))
+
+
+@pytest.fixture
+def history():
+    hist = History(item(f"old-{i}") for i in range(4))
+    hist.start_session()
+    hist.extend(item(f"new-{i}") for i in range(2))
+    return hist
+
+
+def assert_session(hist, boundary, expected):
+    assert hist.session_start_index == boundary
+    assert [entry.raw for entry in hist.span(":").values()] == expected
+    assert [entry.raw for entry in hist.str_search("-").values()] == expected
+    assert [entry.raw for entry in hist.regex_search(".*").values()] == expected
+    assert list(hist.span(":")) == list(range(boundary + 1, len(hist) + 1))
+
+
+@pytest.mark.parametrize(
+    ("index", "boundary", "expected"),
+    [
+        (0, 3, ["new-0", "new-1"]),
+        (-6, 3, ["new-0", "new-1"]),
+        (-1, 4, ["new-0"]),
+        (slice(None, 2), 2, ["new-0", "new-1"]),
+        (slice(2, 5), 2, ["new-1"]),
+        (slice(None, None, 2), 2, ["new-1"]),
+        (slice(None, None, -2), 2, ["new-0"]),
+        (slice(5, 1, -2), 3, ["new-0"]),
+        (slice(None), 0, []),
+        (slice(1, 1), 4, ["new-0", "new-1"]),
+    ],
+)
+def test_delete_tracks_session(history, index, boundary, expected):
+    del history[index]
+    assert_session(history, boundary, expected)
+
+
+@pytest.mark.parametrize(
+    ("index", "removed", "boundary", "expected"),
+    [
+        (0, "old-0", 3, ["new-0", "new-1"]),
+        (-3, "old-3", 3, ["new-0", "new-1"]),
+        (-1, "new-1", 4, ["new-0"]),
+    ],
+)
+def test_pop_tracks_session(history, index, removed, boundary, expected):
+    assert history.pop(index).raw == removed
+    assert_session(history, boundary, expected)
+
+
+@pytest.mark.parametrize(("raw", "boundary", "expected"), [("old-1", 3, ["new-0", "new-1"]), ("new-1", 4, ["new-0"])])
+def test_remove_tracks_session(history, raw, boundary, expected):
+    assert history.remove(item(raw)) is None
+    assert_session(history, boundary, expected)
+
+
+@pytest.mark.parametrize(
+    ("index", "boundary", "expected"),
+    [
+        (0, 5, ["new-0", "new-1"]),
+        (3, 5, ["new-0", "new-1"]),
+        (-3, 5, ["new-0", "new-1"]),
+        (-99, 5, ["new-0", "new-1"]),
+        (4, 4, ["inserted-0", "new-0", "new-1"]),
+        (99, 4, ["new-0", "new-1", "inserted-0"]),
+        (-1, 4, ["new-0", "inserted-0", "new-1"]),
+    ],
+)
+def test_insert_tracks_session(history, index, boundary, expected):
+    history.insert(index, item("inserted-0"))
+    assert_session(history, boundary, expected)
+
+
+@pytest.mark.parametrize(
+    ("index", "count", "boundary", "expected"),
+    [
+        (slice(0, 1), 2, 5, ["new-0", "new-1"]),
+        (slice(1, 3), 0, 2, ["new-0", "new-1"]),
+        (slice(2, 5), 2, 2, ["replacement-0", "replacement-1", "new-1"]),
+        (slice(2, 5), 4, 2, ["replacement-0", "replacement-1", "replacement-2", "replacement-3", "new-1"]),
+        (slice(4, 5), 2, 4, ["replacement-0", "replacement-1", "new-1"]),
+        (slice(None, 4), 1, 1, ["new-0", "new-1"]),
+        (slice(None), 2, 0, ["replacement-0", "replacement-1"]),
+        (slice(2, 2), 2, 6, ["new-0", "new-1"]),
+        (slice(4, 2), 2, 4, ["replacement-0", "replacement-1", "new-0", "new-1"]),
+        (slice(3, 1), 2, 6, ["new-0", "new-1"]),
+        (slice(None, None, 2), 3, 4, ["replacement-2", "new-1"]),
+        (slice(None, None, -2), 3, 4, ["new-0", "replacement-0"]),
+    ],
+)
+def test_slice_assignment_tracks_session(history, index, count, boundary, expected):
+    history[index] = (item(f"replacement-{i}") for i in range(count))
+    assert_session(history, boundary, expected)
+
+
+def test_self_assignment_preserves_session(history):
+    history[:] = history
+    assert_session(history, 4, ["new-0", "new-1"])
+    history[1:3] = history
+    assert_session(history, 8, ["new-0", "new-1"])
+
+
+@pytest.mark.parametrize("index", [0, 3, 4, -1])
+def test_index_assignment_preserves_session(history, index):
+    history[index] = item("replacement-0")
+    assert_session(history, 4, [entry.raw for entry in history[4:]])
+
+
+@pytest.mark.parametrize("count", [0, -1, 1, 2])
+def test_repetition_preserves_session_boundary(history, count):
+    history *= count
+    boundary = 0 if count <= 0 else 4
+    assert_session(history, boundary, [entry.raw for entry in history[boundary:]])
+
+
+@pytest.mark.parametrize("operation", ["delete", "pop", "remove", "slice_step", "slice_length", "slice_iterator"])
+def test_failed_mutation_preserves_history_and_boundary(history, operation):
+    original = list(history)
+
+    def broken_items():
+        yield item("replacement-0")
+        raise RuntimeError("incomplete replacement")
+
+    error = IndexError if operation in ("delete", "pop") else RuntimeError if operation == "slice_iterator" else ValueError
+
+    def mutate():
+        if operation == "delete":
+            del history[-99]
+        elif operation == "pop":
+            history.pop(-99)
+        elif operation == "remove":
+            history.remove(item("missing-0"))
+        elif operation == "slice_step":
+            del history[::0]
+        elif operation == "slice_length":
+            history[::2] = [item("replacement-0")]
+        else:
+            history[:2] = broken_items()
+
+    with pytest.raises(error):
+        mutate()
+    assert history == original
+    assert_session(history, 4, ["new-0", "new-1"])
+
+
+def test_mutations_compose_with_truncate_and_append(history):
+    history.insert(0, item("inserted-0"))
+    history.pop(1)
+    history.truncate(4)
+    history.append(Statement("", raw="new-2"))
+    assert_session(history, 2, ["new-0", "new-1", "new-2"])
+    history.clear()
+    history.extend([item("new-3")])
+    assert_session(history, 0, ["new-3"])
+
+
+def test_remove_retains_native_list_dispatch(history):
+    class CustomHistory(History):
+        def index(self, *_args):
+            raise AssertionError("remove must not dispatch to an overridden index")
+
+        def pop(self, *_args):
+            raise AssertionError("remove must not dispatch to an overridden pop")
+
+    custom = CustomHistory(history)
+    custom.session_start_index = history.session_start_index
+    custom.remove(item("old-1"))
+    assert_session(custom, 3, ["new-0", "new-1"])
+
+
+@pytest.mark.parametrize("as_iterator", [False, True])
+@pytest.mark.parametrize(
+    ("start", "stop", "removed"),
+    [
+        (0, 6, {1}),
+        (0, 6, {0, 2}),
+        (0, 6, {4}),
+        (0, 6, {5}),
+        (0, 6, {0, 4}),
+        (1, 6, {1, 3}),
+        (2, 6, {2, 4}),
+        (3, 5, {3}),
+        (0, 5, {1, 4}),
+    ],
+)
+def test_filtering_straddling_slice_matches_deletion(history, start, stop, removed, as_iterator):
+    original = list(history)
+    expected = History(original)
+    expected.session_start_index = history.session_start_index
+    for index in sorted(removed, reverse=True):
+        del expected[index]
+    replacement = [entry for index, entry in enumerate(original) if start <= index < stop and index not in removed]
+    history[start:stop] = iter(replacement) if as_iterator else replacement
+    assert all(actual is wanted for actual, wanted in zip(history, expected, strict=True))
+    assert_session(history, expected.session_start_index, [entry.raw for entry in expected.span(":").values()])
+
+
+def test_filtering_distinguishes_equal_commands_from_different_sessions(history):
+    current = item(history[0].raw)
+    assert current == history[0]
+    assert current is not history[0]
+    history[4] = current
+    history[:] = history[1:]
+    assert history[3] is current
+    assert_session(history, 3, ["old-0", "new-1"])
+
+
+@pytest.mark.parametrize(
+    ("indices", "boundary", "current"),
+    [
+        ([3, 1, 4, 5], 2, ["new-0", "new-1"]),
+        ([4, 1, 5], 0, ["new-0", "old-1", "new-1"]),
+        ([1, 3], 2, []),
+        ([0, 0, 4], 2, ["new-0"]),
+        ([4, 0, 5, 1], 0, ["new-0", "old-0", "new-1", "old-1"]),
+        ([0, 2, 1, 3, 4, 5], 4, ["new-0", "new-1"]),
+    ],
+)
+def test_reordered_straddling_slice_retains_only_leading_persisted_entries(history, indices, boundary, current):
+    history[:] = [history[index] for index in indices]
+    assert_session(history, boundary, current)
+
+
+def test_new_equal_entry_in_straddling_slice_starts_current_session(history):
+    new_entry = item(history[0].raw)
+    history[:] = [new_entry, history[1], history[4]]
+    assert_session(history, 0, ["old-0", "old-1", "new-0"])
+
+
+def test_failed_straddling_replacement_preserves_history(history):
+    original = list(history)
+
+    def broken_items():
+        yield original[0]
+        raise RuntimeError("incomplete replacement")
+
+    with pytest.raises(RuntimeError, match="incomplete replacement"):
+        history[:] = broken_items()
+    assert all(actual is expected for actual, expected in zip(history, original, strict=True))
+    assert_session(history, 4, ["new-0", "new-1"])