History Module API Reference¶

def add(
    self,
    command: str,
    *,
    params: Sequence[str] | None = None,
    success: bool | None = True,
    return_code: int | None = 0,
    duration_ms: float | None = None,
) -> None:
    """Appends a new command invocation to the history.

    This operation is cross-process safe. It acquires a lock, reloads the
    latest history from disk, appends the new entry, and writes the
    updated history back atomically. Errors are logged but suppressed to
    allow the originating command to complete its execution.

    Args:
        command (str): The command name (ASCII characters are enforced).
        params (Sequence[str] | None): A list of parameters and flags.
        success (bool | None): Whether the command succeeded.
        return_code (int | None): The exit code of the command.
        duration_ms (float | None): The command's duration in milliseconds.
    """
    fp = self._get_history_path()
    entry = {
        "command": _ascii_clean(command),
        "params": list(params or []),
        "timestamp": _now(),
        "success": bool(success),
        "return_code": return_code if return_code is not None else 0,
        "duration_ms": float(duration_ms) if duration_ms is not None else None,
    }
    with _interprocess_lock(fp):
        self._reload()
        if self._load_error:
            msg = f"[error] Could not load command history: {self._load_error}"
            self._obs.log("error", msg, extra={"path": str(fp)})
            print(msg, file=sys.stderr)
            self._events = []
        self._events.append(entry)
        try:
            _atomic_write_json(fp, self._events)
            self._load_error = None
        except PermissionError as exc:
            msg = f"[error] Could not record command history: {exc}"
            self._obs.log("error", msg, extra={"path": str(fp)})
            print(msg, file=sys.stderr)
            self._load_error = msg
            return
        except OSError as exc:
            if exc.errno in _ENOSPC_ERRORS:
                msg = f"[error] Could not record command history: {exc}"
                self._obs.log("error", msg, extra={"path": str(fp)})
                print(msg, file=sys.stderr)
                self._load_error = msg
                return
            msg = f"[error] Could not record command history: {exc}"
            self._obs.log("error", msg, extra={"path": str(fp)})
            print(msg, file=sys.stderr)
            self._load_error = msg
            return
    with suppress(Exception):
        self._tel.event("history_event_added", {"command": entry["command"]})

def clear(self) -> None:
    """Erases all persisted history.

    This operation is cross-process safe and atomic.

    Raises:
        PermissionError: If the history file or directory is not writable.
        OSError: For other filesystem-related failures.
    """
    fp = self._get_history_path()
    try:
        with _interprocess_lock(fp):
            self._events = []
            _atomic_write_json(fp, self._events)
            self._load_error = None
            self._tel.event("history_cleared", {})
    except Exception as exc:
        msg = f"History clear failed: {exc}"
        self._obs.log("error", msg, extra={"path": str(fp)})
        self._load_error = msg
        raise
    finally:
        self._reload()

def export(self, path: Path) -> None:
    """Exports the current history to a file as a JSON array.

    This operation is a read-only snapshot and does not lock the source file.

    Args:
        path (Path): The destination file path.

    Raises:
        RuntimeError: On I/O failures.
    """
    self._reload()
    try:
        path = path.expanduser()
        path.parent.mkdir(parents=True, exist_ok=True)
        text = json.dumps(self._events, ensure_ascii=False, indent=2) + "\n"
        path.write_text(text, encoding="utf-8")
    except Exception as exc:
        raise RuntimeError(f"Failed exporting history: {exc}") from exc

def flush(self) -> None:
    """Persists all in-memory history data to disk."""
    self._dump()

def import_(self, path: Path) -> None:
    """Imports history entries from a file, merging with current history.

    This operation is cross-process safe and atomic.

    Args:
        path (Path): The source file path containing a JSON array of entries.

    Raises:
        RuntimeError: On I/O or parsing failures.
    """
    fp = self._get_history_path()
    try:
        with _interprocess_lock(fp):
            self._reload()
            if self._load_error:
                raise RuntimeError(self._load_error)
            path = path.expanduser()
            if not path.exists():
                raise RuntimeError(f"Import file not found: {path}")
            raw = path.read_text(encoding="utf-8")
            data = json.loads(raw)
            if not isinstance(data, list):
                raise RuntimeError(
                    f"Invalid import format (not JSON array): {path}"
                )
            imported: list[dict[str, Any]] = []
            for item in data:
                if not isinstance(item, dict):
                    continue
                e = dict(item)
                e["command"] = _ascii_clean(str(e.get("command", "")))
                if "timestamp" not in e:
                    e["timestamp"] = _now()
                imported.append(e)
            self._events.extend(imported)
            if len(self._events) > _MAX_IN_MEMORY:
                self._events = self._events[-_MAX_IN_MEMORY:]
            _atomic_write_json(fp, self._events)
            self._load_error = None
            with suppress(Exception):
                self._tel.event("history_imported", {"count": len(imported)})

    except Exception as exc:
        msg = f"History import failed: {exc}"
        self._obs.log(
            "error", msg, extra={"import_path": str(path), "history_path": str(fp)}
        )
        raise RuntimeError(msg) from exc

def list(
    self,
    *,
    limit: int | None = 20,
    group_by: str | None = None,
    filter_cmd: str | None = None,
    sort: str | None = None,
) -> list[dict[str, Any]]:
    """Returns a view of the command history, with optional transformations.

    This is a read-only operation and does not acquire a cross-process lock,
    meaning it may not reflect writes from concurrent processes.

    Args:
        limit (int | None): The maximum number of entries to return. A value
            of 0 returns an empty list.
        group_by (str | None): If provided, returns a grouped summary.
        filter_cmd (str | None): If provided, returns only entries whose
            command contains this case-sensitive substring.
        sort (str | None): If 'timestamp', sorts entries by timestamp.

    Returns:
        list[dict[str, Any]]: A list of history entries or grouped summaries.

    Raises:
        RuntimeError: If the history file is corrupt.
    """
    self._reload()
    fp = self._get_history_path()
    try:
        writable = os.access(fp.parent, os.W_OK)
    except Exception:
        writable = True
    if not writable:
        msg = f"Permission denied for history directory: {fp.parent}"
        self._obs.log("error", msg, extra={"path": str(fp)})
        print(msg, file=sys.stderr)
    if self._load_error:
        raise RuntimeError(self._load_error)
    if limit == 0:
        return []
    entries: list[dict[str, Any]] = list(self._events)
    if filter_cmd:
        needle = str(filter_cmd)
        entries = [e for e in entries if needle in (e.get("command") or "")]
    if sort == "timestamp":
        entries.sort(key=lambda e: e.get("timestamp", 0))
    if group_by:
        grouped: dict[Any, MutableSequence[dict[str, Any]]] = {}
        for e in entries:
            grouped.setdefault(e.get(group_by, "unknown"), []).append(e)
        summary = [
            {
                "group": k,
                "count": len(v),
                "last_run": max((x.get("timestamp", 0) for x in v), default=0),
            }
            for k, v in grouped.items()
        ]
        return summary[:limit] if (limit and limit > 0) else summary
    if limit and limit > 0:
        entries = entries[-limit:]
    return entries