Skip to content
Back to top

# Source code for xonsh.history.json

    """Implements JSON version of xonsh history backend."""

    import collections
    import collections.abc as cabc
    import os
    import re
    import sys
    import tempfile
    import threading
    import time

    from xonsh.built_ins import XSH

    try:
        import ujson as json

        JSONDecodeError = json.JSONDecodeError  # type: ignore
    except ImportError:
        import json  # type: ignore

        JSONDecodeError = json.decoder.JSONDecodeError  # type: ignore

    import xonsh.lib.lazyjson as xlj
    import xonsh.tools as xt
    import xonsh.xoreutils.uptime as uptime
    from xonsh.history.base import History

    def _xhj_gc_commands_to_rmfiles(hsize, files):
        """Return number of units and list of history files to remove to get under the limit,

        Parameters:
        -----------
        hsize (int):  units of history, # of commands in this case.
        files ((mod_ts, num_commands, path)[], fsize): history files, sorted oldest first.

        Returns:
        --------
        hsize_removed (int):  units of history to be removed
        rm_files ((mod_ts, num_commands, path, fsize)[]): list of files to remove.
        """
        n = 0
        ncmds = 0
        for _, fcmds, _, _ in reversed(files):
            # `files` comes in with empty files included (now), don't need special handling to gc them here.

            if ncmds + fcmds > hsize:
                break
            ncmds += fcmds
            n += 1

        cmds_removed = 0
        files_removed = files[:-n] if n > 0 else files
        for _, fcmds, _, _ in files_removed:
            cmds_removed += fcmds

        return cmds_removed, files_removed

    def _xhj_gc_files_to_rmfiles(hsize, files):
        """Return the number and list of history files to remove to get under the file limit."""
        rmfiles = files[:-hsize] if len(files) > hsize else []
        return len(rmfiles), rmfiles

    def _xhj_gc_seconds_to_rmfiles(hsize, files):
        """Return excess duration and list of history files to remove to get under the age limit."""
        now = time.time()
        n = 0

        for ts, _, _, _ in files:
            if (now - ts) < hsize:
                break
            n += 1

        rmfiles = files[:n]
        size_over = now - hsize - rmfiles[0][0] if n > 0 else 0
        return size_over, rmfiles

    def _xhj_gc_bytes_to_rmfiles(hsize, files):
        """Return the history files to remove to get under the byte limit."""
        n = 0
        nbytes = 0
        for _, _, _, fsize in reversed(files):
            if nbytes + fsize > hsize:
                break
            nbytes += fsize
            n += 1
        bytes_removed = 0
        files_removed = files[:-n] if n > 0 else files
        for _, _, _, fsize in files_removed:
            bytes_removed += fsize

        return bytes_removed, files_removed

    def _xhj_get_data_dir():
        dir = xt.expanduser_abs_path(
            os.path.join(XSH.env.get("XONSH_DATA_DIR"), "history_json")
        )
        if not os.path.exists(dir):
            os.makedirs(dir)
        return dir

    def _xhj_get_data_dir_files(data_dir, include_mtime=False):
        """Iterate over all the history files in a data dir,
        optionally including the `mtime` for each file.
        """
        # list of (file, mtime) pairs
        data_dir = xt.expanduser_abs_path(data_dir)
        try:
            entries = os.listdir(data_dir)
        except OSError:
            if XSH.env.get("XONSH_DEBUG"):
                xt.print_exception(
                    f"Could not collect xonsh history json files from {data_dir}"
                )
            return
        for file in entries:
            if not (file.startswith("xonsh-") and file.endswith(".json")):
                continue
            fullpath = os.path.join(data_dir, file)
            if include_mtime:
                try:
                    mtime = os.path.getmtime(fullpath)
                except OSError as err:
                    if XSH.env.get("XONSH_DEBUG"):
                        print(
                            f"xonsh history: skipping unreadable file {fullpath!r}: {err}",
                            file=sys.stderr,
                        )
                    continue
            else:
                mtime = None
            yield fullpath, mtime

    def _xhj_get_history_files(sort=True, newest_first=False, modified_since=None):
        """Find and return the history files. Optionally sort files by
        modify time, or include only those modified after a certain time.
        """
        data_dirs = [
            _xhj_get_data_dir(),
            XSH.env.get("XONSH_DATA_DIR"),  # backwards compatibility, remove in the future
        ]

        files = []
        for data_dir in data_dirs:
            include_mtime = sort or (modified_since is not None)
            for file, mtime in _xhj_get_data_dir_files(data_dir, include_mtime):
                if modified_since is None or mtime > modified_since:
                    files.append((file, mtime))
        if sort:
            files.sort(key=lambda x: x[1], reverse=newest_first)
        # drop the mtimes
        files = [f[0] for f in files]

        custom_history_file = XSH.env.get("XONSH_HISTORY_FILE", None)
        if custom_history_file:
            custom_history_file = xt.expanduser_abs_path(custom_history_file)
            if custom_history_file not in files:
                files.insert(0, custom_history_file)
        return files

    def _xhj_pull_items(pull_times, src_sessionid=None):
        """List all history items after a given start time.
        Optionally restrict to just items from a single session.
        """
        last_full_pull_time = pull_times[None]

        if src_sessionid:
            filename = os.path.join(_xhj_get_data_dir(), f"xonsh-{src_sessionid}.json")
            src_paths = [filename]
        else:
            src_paths = _xhj_get_history_files(
                sort=True, modified_since=last_full_pull_time
            )

        # src_paths may include the current session's file, so skip it to avoid duplicates
        hist = XSH.history
        current_session_path = getattr(hist, "filename", None) or ""
        items = []
        for path in src_paths:
            if path == current_session_path:
                continue
            try:
                lj = xlj.LazyJSON(path)
            except (JSONDecodeError, ValueError):
                continue

            sessionid = os.path.split(path)[-1][6:-5]
            cmds = lj["cmds"]
            if len(cmds) == 0:
                continue
            # the cutoff point is likely to be very near the end of the session, so iterate backward
            for i in range(len(cmds) - 1, -1, -1):
                item = cmds[i].load()
                if item["ts"][1] > pull_times.get(sessionid, last_full_pull_time):
                    items.append(item)
                else:
                    break

        items.sort(key=lambda i: i["ts"][1])
        return items

    class JsonHistoryGC(threading.Thread):
        """Shell history garbage collection."""

        def __init__(self, wait_for_shell=True, size=None, force=False, *args, **kwargs):
            """Thread responsible for garbage collecting old history.

            May wait for shell (and for xonshrc to have been loaded) to start work.
            """
            super().__init__(*args, **kwargs)
            self.daemon = True
            self.size = size
            self.wait_for_shell = wait_for_shell
            self.force_gc = force
            self.gc_units_to_rmfiles = {
                "commands": _xhj_gc_commands_to_rmfiles,
                "files": _xhj_gc_files_to_rmfiles,
                "s": _xhj_gc_seconds_to_rmfiles,
                "b": _xhj_gc_bytes_to_rmfiles,
            }
            self.start()

        def run(self):
            while self.wait_for_shell:
                time.sleep(0.01)
            env = XSH.env  # pylint: disable=no-member
            xonsh_debug = env.get("XONSH_DEBUG", 0)
            if self.size is None:
                hsize, units = env.get("XONSH_HISTORY_SIZE")
            else:
                hsize, units = xt.to_history_tuple(self.size)
            files = self.files(only_unlocked=True)
            rmfiles_fn = self.gc_units_to_rmfiles.get(units)
            if rmfiles_fn is None:
                raise ValueError(f"Units type {units!r} not understood")

            size_over, rm_files = rmfiles_fn(hsize, files)
            hist = getattr(XSH, "history", None)
            if hist is not None:  # remember last gc pass history size
                hist.hist_size = size_over + hsize
                hist.hist_units = units

            if self.force_gc or size_over < hsize:
                i = 0
                for _, _, f, _ in rm_files:
                    try:
                        os.remove(f)
                        if xonsh_debug:
                            print(
                                f"... Deleted {i:7d} of {len(rm_files):7d} history files.\r",
                                end="",
                            )
                        pass
                    except OSError:
                        pass
                    i += 1
            else:
                print(
                    f"Warning: History garbage collection would discard more history ({size_over} {units}) than it would keep ({hsize}).\n"
                    "Not removing any history for now. Either increase your limit ($XONSH_HISTORY_SIZE), or run `history gc --force`."
                )

        def files(self, only_unlocked=False):
            """Find and return the history files. Optionally locked files may be
            excluded.

            This is sorted by the last closed time. Returns a list of
            (file_size, timestamp, number of cmds, file name) tuples.
            """
            env = XSH.env
            if env is None:
                return []

            xonsh_debug = env.get("XONSH_DEBUG", 0)
            boot = uptime.boottime()
            fs = _xhj_get_history_files(sort=False)
            files = []
            time_start = time.time()
            for f in fs:
                try:
                    cur_file_size = os.path.getsize(f)
                    if cur_file_size == 0:
                        # collect empty files (for gc)
                        files.append((os.path.getmtime(f), 0, f, cur_file_size))
                        continue
                    lj = xlj.LazyJSON(f, reopen=False)
                    if lj.get("locked", False) and lj["ts"][0] < boot:
                        # computer was rebooted between when this history was created
                        # and now and so this history should be unlocked.
                        hist = lj.load()
                        lj.close()
                        hist["locked"] = False
                        with open(f, "w", newline="\n", encoding="utf-8") as fp:
                            xlj.ljdump(hist, fp, sort_keys=True)
                        lj = xlj.LazyJSON(f, reopen=False)
                    if only_unlocked and lj.get("locked", False):
                        continue
                    # info: file size, closing timestamp, number of commands, filename
                    ts = lj.get("ts", (0.0, None))
                    files.append(
                        (ts[1] or ts[0], len(lj.sizes["cmds"]) - 1, f, cur_file_size)
                    )
                    lj.close()
                    if xonsh_debug:
                        time_lag = time.time() - time_start
                        print(
                            f"[history.{json.__name__}] Enumerated {len(files):,d} history files for {time_lag:0.4f}s.\r",
                            end="",
                            file=sys.stderr,
                        )
                except (OSError, ValueError):
                    continue
            files.sort()  # this sorts by elements of the tuple,
            # the first of which just happens to be file mod time.
            # so sort by oldest first.
            return files

    class JsonHistoryFlusher(threading.Thread):
        """Flush shell history to disk periodically."""

        def __init__(
            self, filename, buffer, queue, cond, at_exit=False, skip=None, *args, **kwargs
        ):
            """Thread for flushing history."""
            super().__init__(*args, **kwargs)
            self.filename = filename
            self.buffer = buffer
            self.queue = queue
            queue.append(self)
            self.cond = cond
            self.at_exit = at_exit
            self.skip = skip
            if at_exit:
                with self.cond:
                    self.cond.wait_for(self.i_am_at_the_front)
                    self.dump()
                    self.queue.popleft()
                    self.cond.notify_all()
            else:
                self.start()

        def run(self):
            with self.cond:
                self.cond.wait_for(self.i_am_at_the_front)
                self.dump()
                self.queue.popleft()
                self.cond.notify_all()

        def i_am_at_the_front(self):
            """Tests if the flusher is at the front of the queue."""
            return self is self.queue[0]

        def dump(self):
            """Write the cached history to external storage."""
            opts = XSH.env.get("HISTCONTROL", "")
            last_inp = None
            cmds = []
            for cmd in self.buffer:
                if "ignoredups" in opts and cmd["inp"] == last_inp:
                    # Skipping dup cmd
                    if self.skip is not None:
                        self.skip(1)
                    continue
                if "ignoreerr" in opts and cmd["rtn"] != 0:
                    # Skipping failed cmd
                    if self.skip is not None:
                        self.skip(1)
                    continue

                cmds.append(cmd)
                last_inp = cmd["inp"]
            try:
                with open(self.filename, newline="\n", encoding="utf-8") as f:
                    hist = xlj.LazyJSON(f).load()
            except (JSONDecodeError, ValueError, OSError):
                # File is corrupted or unreadable - start with empty history
                hist = {"cmds": [], "sessionid": "", "ts": [time.time(), 0], "locked": True}
            load_hist_len = len(hist["cmds"])
            hist["cmds"].extend(cmds)
            if self.at_exit:
                # todo: check why this is here.
                if "ts" in hist:
                    hist["ts"][1] = time.time()  # apply end time
                hist["locked"] = False
            if not XSH.env.get("XONSH_STORE_STDOUT", False):
                [cmd.pop("out") for cmd in hist["cmds"][load_hist_len:] if "out" in cmd]
            # Atomic write: write to temp file, then os.replace()
            dirname = os.path.dirname(self.filename)
            fd, tmpname = tempfile.mkstemp(dir=dirname, suffix=".json.tmp")
            try:
                with os.fdopen(fd, "w", newline="\n", encoding="utf-8") as f:
                    xlj.ljdump(hist, f, sort_keys=True)
            except Exception as err:
                print(f"history: failed to write {tmpname!r}: {err}", file=sys.stderr)
                return
            try:
                os.replace(tmpname, self.filename)
            except Exception as err:
                print(
                    f"history: failed to replace {tmpname!r} -> {self.filename!r}: {err}",
                    file=sys.stderr,
                )

    class JsonCommandField(cabc.Sequence):
        """A field in the 'cmds' portion of history."""

        def __init__(self, field, hist, default=None):
            """Represents a field in the 'cmds' portion of history.

            Will query the buffer for the relevant data, if possible. Otherwise it
            will lazily acquire data from the file.

            Parameters
            ----------
            field : str
                The name of the field to query.
            hist : History object
                The history object to query.
            default : optional
                The default value to return if key is not present.
            """
            self.field = field
            self.hist = hist
            self.default = default

        def __len__(self):
            return len(self.hist)

        def __getitem__(self, key):
            if not self.hist.remember_history:
                return ""

            size = len(self)
            if isinstance(key, slice):
                return [self[i] for i in range(*key.indices(size))]
            elif not isinstance(key, int):
                raise IndexError("JsonCommandField may only be indexed by int or slice.")
            elif size == 0:
                raise IndexError("JsonCommandField is empty.")
            # now we know we have an int
            key = size + key if key < 0 else key  # ensure key is non-negative
            bufsize = len(self.hist.buffer)
            if size - bufsize <= key:  # key is in buffer
                return self.hist.buffer[key + bufsize - size].get(self.field, self.default)
            # now we know we have to go into the file
            queue = self.hist._queue
            queue.append(self)
            with self.hist._cond:
                self.hist._cond.wait_for(self.i_am_at_the_front)
                with open(self.hist.filename, newline="\n", encoding="utf-8") as f:
                    lj = xlj.LazyJSON(f, reopen=False)
                    rtn = lj["cmds"][key].get(self.field, self.default)
                    if isinstance(rtn, xlj.LJNode):
                        rtn = rtn.load()
                queue.popleft()
            return rtn

        def i_am_at_the_front(self):
            """Tests if the command field is at the front of the queue."""
            return self is self.hist._queue[0]

    class JsonHistory(History):
        """Xonsh history backend implemented with JSON files.

        JsonHistory implements an extra action: ``diff``
        """

        def __init__(
            self,
            filename=None,
            sessionid=None,
            buffersize=100,
            gc=True,
            save_cwd=None,
            **meta,
        ):
            """Represents a xonsh session's history as an in-memory buffer that is
            periodically flushed to disk.

            Parameters
            ----------
            filename : str, optional
                Location of history file, defaults to
                ``$XONSH_DATA_DIR/history_json/xonsh-{sessionid}.json``.
            sessionid : int, uuid, str, optional
                Current session identifier, will generate a new sessionid if not
                set.
            buffersize : int, optional
                Maximum buffersize in memory.
            meta : optional
                Top-level metadata to store along with the history. The kwargs
                'cmds' and 'sessionid' are not allowed and will be overwritten.
            gc : bool, optional
                Run garbage collector flag.
            """
            super().__init__(sessionid=sessionid, **meta)
            if filename is None:
                # pylint: disable=no-member
                data_dir = _xhj_get_data_dir()
                self.filename = os.path.join(data_dir, f"xonsh-{self.sessionid}.json")
            else:
                self.filename = filename

            if self.filename and not os.path.exists(os.path.expanduser(self.filename)):
                meta["cmds"] = []
                meta["sessionid"] = str(self.sessionid)
                with open(self.filename, "w", newline="\n", encoding="utf-8") as f:
                    xlj.ljdump(meta, f, sort_keys=True)

                try:
                    sudo_uid = os.environ.get("SUDO_UID")
                    sudo_gid = os.environ.get("SUDO_GID")
                    if None not in (sudo_uid, sudo_gid):
                        os.chown(self.filename, int(sudo_uid), int(sudo_gid))
                    os.chmod(self.filename, 0o600)
                except Exception:  # pylint: disable=broad-except
                    pass

            self.buffer = []
            self.buffersize = buffersize
            self._queue = collections.deque()
            self._cond = threading.Condition()
            self._len = 0
            self._skipped = 0
            self.last_cmd_out = None
            self.last_cmd_rtn = None
            self.gc = JsonHistoryGC() if gc else None
            # pull times are tracked per-source-session; None means all sesssions
            self.last_pull_times = {None: time.time()}
            # command fields that are known
            self.tss = JsonCommandField("ts", self)
            self.inps = JsonCommandField("inp", self)
            self.outs = JsonCommandField("out", self)
            self.rtns = JsonCommandField("rtn", self)
            self.cwds = JsonCommandField("cwd", self)
            self.save_cwd = (
                save_cwd
                if save_cwd is not None
                else XSH.env.get("XONSH_HISTORY_SAVE_CWD", True)
            )

        def __len__(self):
            return self._len - self._skipped

        def append(self, cmd):
            """Appends command to history. Will periodically flush the history to file.

            Parameters
            ----------
            cmd : dict
                This dict contains information about the command that is to be
                added to the history list. It should contain the keys ``inp``,
                ``rtn`` and ``ts``. These key names mirror the same names defined
                as instance variables in the ``HistoryEntry`` class.
                Additionally, an optional key ``spc`` may be present which will
                affect commands from being stored if ignorespace is in $HISTCONTROL.

            Returns
            -------
            hf : JsonHistoryFlusher or None
                The thread that was spawned to flush history
            """
            if (not self.remember_history) or self.is_ignored(cmd):
                return

            opts = XSH.env.get("HISTCONTROL", "")
            skipped_by_ignore_space = "ignorespace" in opts and cmd.get("spc")
            if skipped_by_ignore_space:
                return None

            self.buffer.append(cmd)
            self._len += 1  # must come before flushing

            if not self.save_cwd and "cwd" in cmd:
                del cmd["cwd"]

            try:
                del cmd["spc"]
            except KeyError:
                pass

            if len(self.buffer) >= self.buffersize:
                hf = self.flush()
            else:
                hf = None
            return hf

        def flush(self, at_exit=False):
            """Flushes the current command buffer to disk.

            Parameters
            ----------
            at_exit : bool, optional
                Whether the JsonHistoryFlusher should act as a thread in the
                background, or execute immediately and block.

            Returns
            -------
            hf : JsonHistoryFlusher or None
                The thread that was spawned to flush history
            """
            # Implicitly covers case of self.remember_history being False.
            if len(self.buffer) == 0:
                return

            def skip(num):
                self._skipped += num

            hf = JsonHistoryFlusher(
                self.filename,
                tuple(self.buffer),
                self._queue,
                self._cond,
                at_exit=at_exit,
                skip=skip,
            )
            self.buffer = []
            return hf

        def items(self, newest_first=False):
            """Display history items of current session."""
            if newest_first:
                items = zip(reversed(self.inps), reversed(self.tss), strict=False)
            else:
                items = zip(self.inps, self.tss, strict=False)
            for item, tss in items:
                yield {"inp": item.rstrip(), "ts": tss[0]}

        def all_items(self, newest_first=False, **kwargs):
            """
            Returns all history as found in XONSH_DATA_DIR.

            yield format: {'inp': cmd, 'rtn': 0, ...}
            """
            while self.gc and self.gc.is_alive():
                time.sleep(0.011)  # gc sleeps for 0.01 secs, sleep a beat longer
            for f in _xhj_get_history_files(newest_first=newest_first):
                if f == self.filename:
                    continue
                try:
                    json_file = xlj.LazyJSON(f, reopen=False)
                except (ValueError, OSError) as err:
                    if isinstance(err, OSError):
                        if str(f) in str(err):
                            msg = f"history: skip: {err}"
                        else:
                            msg = f"history: skip {f}: {err}"
                        xt.print_above_prompt(msg)
                    continue
                try:
                    commands = json_file.load()["cmds"]
                except (JSONDecodeError, ValueError):
                    # file is corrupted somehow
                    if XSH.env.get("XONSH_DEBUG", 0) > 0:
                        msg = "xonsh history file {0!r} is not valid JSON"
                        print(msg.format(f), file=sys.stderr)
                    continue
                if newest_first:
                    commands = reversed(commands)
                for c in commands:
                    yield {"inp": c["inp"].rstrip(), "ts": c["ts"][0]}
            # all items should also include session items
            yield from self.items()

        def info(self):
            data = collections.OrderedDict()
            data["backend"] = "json"
            data["sessionid"] = str(self.sessionid)
            data["filename"] = self.filename
            data["length"] = len(self)
            data["buffersize"] = self.buffersize
            data["bufferlength"] = len(self.buffer)
            envs = XSH.env
            data["gc options"] = envs.get("XONSH_HISTORY_SIZE")
            data["gc_last_size"] = f"{(self.hist_size, self.hist_units)}"
            return data

        def pull(self, show_commands=False, src_sessionid=None):
            if not hasattr(XSH.shell.shell, "prompter"):
                print(f"Shell type {XSH.shell.shell} is not supported.")
                return 0

            cnt = 0
            prev = None
            for item in _xhj_pull_items(self.last_pull_times, src_sessionid):
                line = item["inp"].rstrip()
                if show_commands:
                    print(line)
                if line != prev:
                    XSH.shell.shell.prompter.history.append_string(line)
                    cnt += 1
                prev = line

            # we can dump the session-specific pull times if this is a full pull
            if src_sessionid is None:
                self.last_pull_times = {}
            self.last_pull_times[src_sessionid] = time.time()

            return cnt

        def run_gc(self, size=None, blocking=True, force=False, **_):
            self.gc = JsonHistoryGC(wait_for_shell=False, size=size, force=force)
            if blocking:
                while self.gc.is_alive():  # while waiting for gc.
                    time.sleep(0.1)  # don't monopolize the thread (or Python GIL?)

        def clear(self):
            """Clears the current session's history from both memory and disk."""

            # Wipe history from memory. Keep sessionid and other metadata.
            self.buffer = []
            self.tss = JsonCommandField("ts", self)
            self.inps = JsonCommandField("inp", self)
            self.outs = JsonCommandField("out", self)
            self.rtns = JsonCommandField("rtn", self)
            self.cwds = JsonCommandField("cwd", self)
            self._len = 0
            self._skipped = 0

            # Write empty history directly — flush() would skip empty buffer.
            if self.filename:
                meta = {"cmds": [], "sessionid": str(self.sessionid)}
                with open(self.filename, "w", newline="\n", encoding="utf-8") as f:
                    xlj.ljdump(meta, f, sort_keys=True)

        def delete(self, pattern):
            """Deletes all entries in history which matches a pattern."""
            pattern = re.compile(pattern)

            deleted = 0
            # First, delete any matching commands in the in-memory buffer.
            orig_len = len(self.buffer)
            self.buffer[:] = [cmd for cmd in self.buffer if not pattern.match(cmd["inp"])]
            deleted += orig_len - len(self.buffer)

            # Then, delete any matching commands on disk.
            while self.gc and self.gc.is_alive():
                time.sleep(0.011)  # gc sleeps for 0.01 secs, sleep a beat longer
            for f in _xhj_get_history_files():
                try:
                    json_file = xlj.LazyJSON(f, reopen=False)
                except ValueError:
                    # Invalid json file
                    continue
                try:
                    file_content = json_file.load()
                    commands = file_content["cmds"]
                    new_commands = [c for c in commands if not pattern.match(c["inp"])]
                    deleted += len(commands) - len(new_commands)
                    file_content["cmds"] = new_commands
                    dirname = os.path.dirname(f)
                    fd, tmpname = tempfile.mkstemp(dir=dirname, suffix=".json.tmp")
                    try:
                        with os.fdopen(fd, "w", newline="\n", encoding="utf-8") as fp:
                            xlj.ljdump(file_content, fp)
                        os.replace(tmpname, f)
                    except Exception as err:
                        try:
                            os.unlink(tmpname)
                        except OSError:
                            pass
                        print(
                            f"history delete: failed to update {f!r}: {err}",
                            file=sys.stderr,
                        )
                        continue
                except (JSONDecodeError, ValueError):
                    # file is corrupted somehow
                    if XSH.env.get("XONSH_DEBUG", 0) > 0:
                        msg = "xonsh history file {0!r} is not valid JSON"
                        print(msg.format(f), file=sys.stderr)
                    continue

            return deleted

        def erasedups(self):
            """Remove duplicate commands across all history files, keeping the latest."""
            while self.gc and self.gc.is_alive():
                time.sleep(0.011)

            # Collect all commands from all files with their source file.
            # Each entry: (inp, tsb, file_path, index_in_file)
            entries = []
            for f in _xhj_get_history_files():
                try:
                    json_file = xlj.LazyJSON(f, reopen=False)
                except ValueError:
                    continue
                try:
                    file_content = json_file.load()
                    for idx, cmd in enumerate(file_content.get("cmds", [])):
                        inp = cmd.get("inp", "").rstrip()
                        ts = cmd.get("ts", [0, 0])
                        tsb = ts[0] if ts else 0
                        entries.append((inp, tsb, f, idx))
                except (JSONDecodeError, ValueError):
                    continue

            # Find which entries to keep: for each inp, keep the one with max tsb.
            keep = {}  # inp -> (tsb, file_path, idx)
            for inp, tsb, fpath, idx in entries:
                if inp not in keep or tsb > keep[inp][0]:
                    keep[inp] = (tsb, fpath, idx)

            # Build set of (file_path, idx) to keep.
            keep_set = {(v[1], v[2]) for v in keep.values()}

            total_before = len(entries)
            total_after = len(keep_set)
            if total_before == total_after:
                return 0, total_before

            # Rewrite each file, removing duplicate entries.
            for f in _xhj_get_history_files():
                try:
                    json_file = xlj.LazyJSON(f, reopen=False)
                except ValueError:
                    continue
                try:
                    file_content = json_file.load()
                    del json_file  # release file handle before replacing (Windows)
                    cmds = file_content.get("cmds", [])
                    new_cmds = [cmd for idx, cmd in enumerate(cmds) if (f, idx) in keep_set]
                    if len(new_cmds) == len(cmds):
                        continue  # no changes needed
                    file_content["cmds"] = new_cmds
                    dirname = os.path.dirname(f)
                    fd, tmpname = tempfile.mkstemp(dir=dirname, suffix=".json.tmp")
                    try:
                        with os.fdopen(fd, "w", newline="\n", encoding="utf-8") as fp:
                            xlj.ljdump(file_content, fp)
                        os.replace(tmpname, f)
                    except Exception:
                        try:
                            os.unlink(tmpname)
                        except OSError:
                            pass
                except (JSONDecodeError, ValueError):
                    continue

            return total_before - total_after, total_before