Fix HistoryAccessor.get_tail bug (#13666)

Current implementation of get_tail in HistoryAccessor assumes context
present only in subclass , so it's moved there and the old
implementation is restored.

IPython/core/history.py

@@ -202,7 +202,6 b' class HistoryAccessor(HistoryAccessorBase):'
         config : :class:`~traitlets.config.loader.Config`
             Config object. hist_file can also be set through this.
         """
-        # We need a pointer back to the shell for various tasks.
         super(HistoryAccessor, self).__init__(**traits)
         # defer setting hist_file from kwarg until after init,
         # otherwise the default kwarg value would clobber any value
@@ -344,11 +343,6 b' class HistoryAccessor(HistoryAccessorBase):'
     def get_tail(self, n=10, raw=True, output=False, include_latest=False):
         """Get the last n lines from the history database.
 
-        Most recent entry last.
-
-        Completion will be reordered so that that the last ones are when
-        possible from current session.
-
         Parameters
         ----------
         n : int
@@ -367,31 +361,11 b' class HistoryAccessor(HistoryAccessorBase):'
         self.writeout_cache()
         if not include_latest:
             n += 1
-        # cursor/line/entry
-        this_cur = list(
-            self._run_sql(
-                "WHERE session == ? ORDER BY line DESC LIMIT ?  ",
-                (self.session_number, n),
-                raw=raw,
-                output=output,
-            )
-        )
-        other_cur = list(
-            self._run_sql(
-                "WHERE session != ? ORDER BY session DESC, line DESC LIMIT ?",
-                (self.session_number, n),
-                raw=raw,
-                output=output,
-            )
-        )
-
-        everything = this_cur + other_cur
-
-        everything = everything[:n]
-
+        cur = self._run_sql("ORDER BY session DESC, line DESC LIMIT ?",
+                                (n,), raw=raw, output=output)
         if not include_latest:
-            return list(everything)[:0:-1]
-        return list(everything)[::-1]
+            return reversed(list(cur)[1:])
+        return reversed(list(cur))
 
     @catch_corrupt_db
     def search(self, pattern="*", raw=True, search_raw=True,
@@ -560,7 +534,6 b' class HistoryManager(HistoryAccessor):'
     def __init__(self, shell=None, config=None, **traits):
         """Create a new history manager associated with a shell instance.
         """
-        # We need a pointer back to the shell for various tasks.
         super(HistoryManager, self).__init__(shell=shell, config=config,
             **traits)
         self.save_flag = threading.Event()
@@ -656,6 +629,59 b' class HistoryManager(HistoryAccessor):'
 
         return super(HistoryManager, self).get_session_info(session=session)
 
+    @catch_corrupt_db
+    def get_tail(self, n=10, raw=True, output=False, include_latest=False):
+        """Get the last n lines from the history database.
+
+        Most recent entry last.
+
+        Completion will be reordered so that that the last ones are when
+        possible from current session.
+
+        Parameters
+        ----------
+        n : int
+            The number of lines to get
+        raw, output : bool
+            See :meth:`get_range`
+        include_latest : bool
+            If False (default), n+1 lines are fetched, and the latest one
+            is discarded. This is intended to be used where the function
+            is called by a user command, which it should not return.
+
+        Returns
+        -------
+        Tuples as :meth:`get_range`
+        """
+        self.writeout_cache()
+        if not include_latest:
+            n += 1
+        # cursor/line/entry
+        this_cur = list(
+            self._run_sql(
+                "WHERE session == ? ORDER BY line DESC LIMIT ?  ",
+                (self.session_number, n),
+                raw=raw,
+                output=output,
+            )
+        )
+        other_cur = list(
+            self._run_sql(
+                "WHERE session != ? ORDER BY session DESC, line DESC LIMIT ?",
+                (self.session_number, n),
+                raw=raw,
+                output=output,
+            )
+        )
+
+        everything = this_cur + other_cur
+
+        everything = everything[:n]
+
+        if not include_latest:
+            return list(everything)[:0:-1]
+        return list(everything)[::-1]
+
     def _get_range_session(self, start=1, stop=None, raw=True, output=False):
         """Get input and output history from the current session. Called by
         get_range, and takes similar parameters."""

IPython/core/tests/test_history.py

@@ -17,7 +17,7 b' from tempfile import TemporaryDirectory'
 # our own packages
 from traitlets.config.loader import Config
 
-from IPython.core.history import HistoryManager, extract_hist_ranges
+from IPython.core.history import HistoryAccessor, HistoryManager, extract_hist_ranges
 
 
 def test_proper_default_encoding():
@@ -227,3 +227,68 b' def test_histmanager_disabled():'
 
     # hist_file should not be created
     assert hist_file.exists() is False
+
+def test_get_tail_session_awareness():
+    """Test .get_tail() is:
+                - session specific in HistoryManager
+                - session agnostic in HistoryAccessor
+            same for .get_last_session_id()
+    """
+    ip = get_ipython()
+    with TemporaryDirectory() as tmpdir:
+        tmp_path = Path(tmpdir)
+        hist_file = tmp_path / "history.sqlite"
+        get_source = lambda x: x[2]
+
+        # hm1 creates a new session and adds history entries,
+        # ha catches up
+        hm1 = HistoryManager(shell=ip, hist_file=hist_file)
+        hm1_last_sid = hm1.get_last_session_id
+        ha = HistoryAccessor(hist_file=hist_file)
+        ha_last_sid = ha.get_last_session_id
+
+        hist1 = ["a=1", "b=1", "c=1"]
+        for i, h in enumerate(hist1 + [""], start=1):
+            hm1.store_inputs(i, h)
+        assert list(map(get_source, hm1.get_tail())) == hist1
+        assert list(map(get_source, ha.get_tail())) == hist1
+        sid1 = hm1_last_sid()
+        assert sid1 is not None
+        assert ha_last_sid() == sid1
+
+        # hm2 creates a new session and adds entries,
+        # ha catches up
+        hm2 = HistoryManager(shell=ip, hist_file=hist_file)
+        hm2_last_sid = hm2.get_last_session_id
+
+        hist2 = ["a=2", "b=2", "c=2"]
+        for i, h in enumerate(hist2 + [""], start=1):
+            hm2.store_inputs(i, h)
+        tail = hm2.get_tail(n=3)
+        assert list(map(get_source, tail)) == hist2
+        tail = ha.get_tail(n=3)
+        assert list(map(get_source, tail)) == hist2
+        sid2 = hm2_last_sid()
+        assert sid2 is not None
+        assert ha_last_sid() == sid2
+        assert sid2 != sid1
+
+        # but hm1 still maintains its point of reference
+        # and adding more entries to it doesn't change others
+        # immediate perspective
+        assert hm1_last_sid() == sid1
+        tail = hm1.get_tail(n=3)
+        assert list(map(get_source, tail)) == hist1
+
+        hist3 = ["a=3", "b=3", "c=3"]
+        for i, h in enumerate(hist3 + [""], start=5):
+            hm1.store_inputs(i, h)
+        tail = hm1.get_tail(n=7)
+        assert list(map(get_source, tail)) == hist1 + [""] + hist3
+        tail = hm2.get_tail(n=3)
+        assert list(map(get_source, tail)) == hist2
+        tail = ha.get_tail(n=3)
+        assert list(map(get_source, tail)) == hist2
+        assert hm1_last_sid() == sid1
+        assert hm2_last_sid() == sid2
+        assert ha_last_sid() == sid2