Conversation History

conversation

Conversation history — in-memory storage with optional disk persistence.

Messages are kept in a plain Python list so they are always available for context without any I/O. The list is also written to a JSON file on disk (in the user's data directory) so prior conversations survive restarts.

Design notes

• No database dependency; JSON is self-contained and human-readable.
• Only text content is persisted. Image attachments are not stored on disk (they can be large and are usually transient). The has_image flag lets the UI indicate that images were part of a turn.
• max_messages caps the in-memory list so RAM stays bounded during very long sessions. Older messages are trimmed from the front (oldest first), preserving the most recent context.

Encryption

Pass encrypt=True (the default when the cryptography package is installed) to store the history file as a Fernet-encrypted blob. The symmetric key is kept in a separate history.key file in the same directory, protected with 0o600 permissions.

::

history = ConversationHistory(encrypt=True)   # key auto-created
history.add("user", "Hello!")
# ~/.local/share/linux-ai-npu-assistant/history.enc  ← ciphertext
# ~/.local/share/linux-ai-npu-assistant/history.key  ← AES key (owner only)

Message dataclass

Message(role, content, timestamp=(lambda: isoformat())(), has_image=False)

A single turn in the conversation.

ConversationHistory

ConversationHistory(max_messages=_DEFAULT_MAX_MESSAGES, persist_path=_DEFAULT_HISTORY_FILE, system_prompt='', encrypt=False, encryption_key=None)

Thread-safe, persistent conversation history.

Parameters

max_messages: Maximum number of messages kept in memory. When the list exceeds this limit the oldest messages are removed first. persist_path: JSON file path for persistence. Pass None to disable disk persistence (history lives only for the current session). When encrypt is True the path is rewritten with a .enc extension automatically. system_prompt: An optional system message prepended to every API call to establish the assistant's persona / instructions. encrypt: When True (and the cryptography package is installed), the history file is encrypted with Fernet symmetric encryption. A key file is stored alongside the history file with 0o600 permissions. Defaults to True when cryptography is available, False otherwise (graceful degradation). encryption_key: Optional pre-existing Fernet key bytes. When omitted the key is loaded from (or created in) a history.key file next to the history file.

Source code in src/conversation.py

def __init__(
    self,
    max_messages: int = _DEFAULT_MAX_MESSAGES,
    persist_path: Path | str | None = _DEFAULT_HISTORY_FILE,
    system_prompt: str = "",
    encrypt: bool = False,            # opt-in; set True or call set_password()
    encryption_key: bytes | None = None,  # pre-supplied key (tests / custom setup)
) -> None:
    self._max = max_messages
    self._system_prompt = system_prompt
    self._messages: list[Message] = []
    self._lock = threading.Lock()

    # ── Encryption setup ────────────────────────────────────────────────
    # Encryption is opt-in.  Pass encrypt=True or call set_password() to
    # enable it.  The cryptography package must be installed.
    self._encrypt = bool(encrypt) and _fernet_available()

    if self._encrypt:
        # Derive paths
        base = Path(persist_path) if persist_path else None
        if base is not None:
            # Store ciphertext in .enc sidecar next to the plain file.
            self._path: Path | None = base.with_suffix(".enc")
            key_path = base.parent / "history.key"
        else:
            self._path = None
            key_path = _DEFAULT_KEY_FILE

        if encryption_key is not None:
            self._key: bytes | None = encryption_key
        elif self._path is not None:
            try:
                self._key = load_or_create_key(key_path)
            except OSError as exc:
                logger.warning(
                    "Could not load/create history encryption key (%s); "
                    "disabling encryption for this session.",
                    exc,
                )
                self._encrypt = False
                self._key = None
                self._path = Path(persist_path) if persist_path else None
        else:
            self._key = None
    else:
        self._path = Path(persist_path) if persist_path else None
        self._key = None

    self._load()

is_encrypted property

is_encrypted

Return True when the history is stored encrypted on disk.

add

add(role, content, *, has_image=False)

Append a message and persist immediately.

Parameters

role: "user" or "assistant". content: Text content of the message. has_image: Set to True when the turn included an image (screenshot or uploaded file). The image itself is not stored here.

Returns

Message The newly added message object.

Source code in src/conversation.py

def add(
    self,
    role: str,
    content: str,
    *,
    has_image: bool = False,
) -> Message:
    """Append a message and persist immediately.

    Parameters
    ----------
    role:
        ``"user"`` or ``"assistant"``.
    content:
        Text content of the message.
    has_image:
        Set to ``True`` when the turn included an image (screenshot or
        uploaded file).  The image itself is not stored here.

    Returns
    -------
    Message
        The newly added message object.
    """
    msg = Message(role=role, content=content, has_image=has_image)
    with self._lock:
        self._messages.append(msg)
        # Trim oldest messages if over the cap
        if len(self._messages) > self._max:
            self._messages = self._messages[-self._max :]
    self._save()
    return msg

clear

clear()

Remove all messages from memory and erase the on-disk file.

Source code in src/conversation.py

def clear(self) -> None:
    """Remove all messages from memory and erase the on-disk file."""
    with self._lock:
        self._messages.clear()
    self._save()
    logger.info("Conversation history cleared.")

all_messages

all_messages()

Return a snapshot of all messages (oldest first).

Source code in src/conversation.py

def all_messages(self) -> list[Message]:
    """Return a snapshot of all messages (oldest first)."""
    with self._lock:
        return list(self._messages)

recent

recent(n)

Return the n most recent messages.

Source code in src/conversation.py

def recent(self, n: int) -> list[Message]:
    """Return the *n* most recent messages."""
    with self._lock:
        return list(self._messages[-n:])

to_openai_messages

to_openai_messages(*, include_system=True, max_context=None)

Return the message list in OpenAI /chat/completions format.

Parameters

include_system: Prepend the system prompt if one is configured. max_context: Only include the most recent max_context messages (besides the system message). Use this to avoid hitting context-length limits.

Source code in src/conversation.py

def to_openai_messages(
    self,
    *,
    include_system: bool = True,
    max_context: int | None = None,
) -> list[dict]:
    """Return the message list in OpenAI ``/chat/completions`` format.

    Parameters
    ----------
    include_system:
        Prepend the system prompt if one is configured.
    max_context:
        Only include the most recent *max_context* messages (besides the
        system message).  Use this to avoid hitting context-length limits.
    """
    messages: list[dict] = []
    if include_system and self._system_prompt:
        messages.append({"role": "system", "content": self._system_prompt})

    history = self.all_messages()
    if max_context is not None:
        history = history[-max_context:]

    for msg in history:
        messages.append({"role": msg.role, "content": msg.content})
    return messages

to_ollama_messages

to_ollama_messages(*, max_context=None)

Return the message list in Ollama /api/chat format.

Ollama's chat endpoint mirrors the OpenAI format, so this is a thin wrapper around :meth:to_openai_messages.

Source code in src/conversation.py

def to_ollama_messages(
    self,
    *,
    max_context: int | None = None,
) -> list[dict]:
    """Return the message list in Ollama ``/api/chat`` format.

    Ollama's chat endpoint mirrors the OpenAI format, so this is a thin
    wrapper around :meth:`to_openai_messages`.
    """
    return self.to_openai_messages(
        include_system=True, max_context=max_context
    )

set_password

set_password(password)

Derive a new encryption key from password and re-save.

Uses PBKDF2-HMAC-SHA256 with a fresh random salt so the on-disk key file is updated. The history is immediately re-written with the new key.

Parameters

password: User-chosen password. An empty string disables password-based encryption and falls back to the auto-generated random key.

Raises

RuntimeError If the cryptography package is not installed.

Source code in src/conversation.py

def set_password(self, password: str) -> None:
    """Derive a new encryption key from *password* and re-save.

    Uses PBKDF2-HMAC-SHA256 with a fresh random salt so the on-disk key
    file is updated.  The history is immediately re-written with the new
    key.

    Parameters
    ----------
    password:
        User-chosen password.  An empty string disables password-based
        encryption and falls back to the auto-generated random key.

    Raises
    ------
    RuntimeError
        If the ``cryptography`` package is not installed.
    """
    if not _fernet_available():
        raise RuntimeError(
            "Install the 'cryptography' package to enable encryption: "
            "pip install cryptography"
        )
    if not password:
        # Reset to random key
        if self._path is not None:
            key_path = self._path.parent / "history.key"
            self._key = load_or_create_key(key_path)
        self._encrypt = True
    else:
        self._key = _derive_key_from_password(
            password,
            self._path.parent if self._path else _DEFAULT_HISTORY_DIR,
        )
        self._encrypt = True
    self._save()

change_password

change_password(old_password, new_password)

Change the encryption password.

Verifies that old_password correctly decrypts the current file before switching to new_password.

Raises

ValueError If old_password is wrong (decryption fails).

Source code in src/conversation.py

def change_password(self, old_password: str, new_password: str) -> None:
    """Change the encryption password.

    Verifies that *old_password* correctly decrypts the current file
    before switching to *new_password*.

    Raises
    ------
    ValueError
        If *old_password* is wrong (decryption fails).
    """
    if self._path is None or not self._path.exists():
        self.set_password(new_password)
        return
    # Verify old password by trying to decrypt
    old_key = _derive_key_from_password(
        old_password,
        self._path.parent,
        create_salt=False,
    )
    if old_key is None:
        raise ValueError("Old password is incorrect (no salt file found).")
    try:
        ciphertext = self._path.read_text(encoding="ascii")
        decrypt_data(ciphertext, old_key)
    except Exception as exc:
        raise ValueError(f"Old password is incorrect: {exc}") from exc
    self.set_password(new_password)

export_plaintext

export_plaintext(export_path)

Write the conversation history as unencrypted JSON to export_path.

The exported file is not protected by encryption or restricted permissions — callers are responsible for handling it securely.

Parameters

export_path: Destination file path (will be created or overwritten).

Source code in src/conversation.py

def export_plaintext(self, export_path: Path | str) -> None:
    """Write the conversation history as unencrypted JSON to *export_path*.

    The exported file is **not** protected by encryption or restricted
    permissions — callers are responsible for handling it securely.

    Parameters
    ----------
    export_path:
        Destination file path (will be created or overwritten).
    """
    with self._lock:
        data = [m.to_dict() for m in self._messages]
    text = json.dumps(data, indent=2, ensure_ascii=False)
    out = Path(export_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    from src.security import secure_write
    secure_write(out, text)
    logger.info(
        "Exported %d messages (plaintext) → %s", len(data), out
    )

import_history

import_history(import_path, password=None, *, merge=False)

Import conversation history from a file.

Supports both plain-JSON exports and Fernet-encrypted .enc files.

Parameters

import_path: Path to the file to import. May be a plain-JSON file produced by :meth:export_plaintext or an encrypted .enc produced when encrypt is enabled. password: Password for encrypted files. Pass None for plain-JSON. If the file looks encrypted and no password is supplied a :class:ValueError is raised. merge: When True messages from the import are merged with the existing history (deduplication by timestamp). When False (default) the current history is replaced by the imported one.

Returns

int Number of messages successfully imported.

Raises

FileNotFoundError If import_path does not exist. ValueError If the password is wrong, the file is not valid history JSON, or the file appears encrypted but no password was given.

Source code in src/conversation.py

def import_history(
    self,
    import_path: Path | str,
    password: str | None = None,
    *,
    merge: bool = False,
) -> int:
    """Import conversation history from a file.

    Supports both plain-JSON exports and Fernet-encrypted ``.enc`` files.

    Parameters
    ----------
    import_path:
        Path to the file to import.  May be a plain-JSON file produced by
        :meth:`export_plaintext` or an encrypted ``.enc`` produced when
        *encrypt* is enabled.
    password:
        Password for encrypted files.  Pass ``None`` for plain-JSON.
        If the file looks encrypted and no password is supplied a
        :class:`ValueError` is raised.
    merge:
        When ``True`` messages from the import are merged with the
        existing history (deduplication by timestamp).  When ``False``
        (default) the current history is **replaced** by the imported one.

    Returns
    -------
    int
        Number of messages successfully imported.

    Raises
    ------
    FileNotFoundError
        If *import_path* does not exist.
    ValueError
        If the password is wrong, the file is not valid history JSON, or
        the file appears encrypted but no password was given.
    """
    path = Path(import_path)
    if not path.exists():
        raise FileNotFoundError(f"Import file not found: {path}")

    raw_text = path.read_text(encoding="utf-8")
    stripped = raw_text.strip()

    # Detect encrypted files: Fernet tokens start with 'gAAAAA' (base64)
    # and are never valid JSON.
    looks_encrypted = not (stripped.startswith("[") or stripped.startswith("{"))

    if looks_encrypted:
        if password is None:
            raise ValueError(
                "This file appears to be encrypted. "
                "Please provide the password used when it was saved."
            )
        # Try salt file next to the import file first, then the default dir.
        key = _derive_key_from_password(
            password, path.parent, create_salt=False
        )
        if key is None:
            key = _derive_key_from_password(
                password, _DEFAULT_HISTORY_DIR, create_salt=False
            )
        if key is None:
            # No salt on disk — try the current session key (same password
            # re-derived with a fresh salt is not useful, but the user may
            # be importing a file whose salt was next to the *original*
            # history file on another machine).
            raise ValueError(
                "Cannot derive decryption key: no salt file (history.salt) "
                "was found alongside the import file.  Copy 'history.salt' "
                "from the original machine into the same directory as the "
                "exported file and try again."
            )
        try:
            raw_text = decrypt_data(raw_text, key)
        except Exception as exc:
            raise ValueError(
                f"Decryption failed — wrong password or corrupted file. ({exc})"
            ) from exc
    elif password is not None and not looks_encrypted:
        # Plain JSON provided with a password → ignore the password
        # (backward-compat: the file might have been exported unencrypted).
        logger.debug(
            "import_history: file appears to be plain JSON; "
            "the provided password will be ignored."
        )

    try:
        data = json.loads(raw_text)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid history file format: {exc}") from exc

    imported: list[Message] = []
    for d in data:
        try:
            imported.append(Message.from_dict(d))
        except (KeyError, TypeError) as exc:
            logger.warning(
                "Skipped malformed message during import: %s", exc
            )

    with self._lock:
        if merge:
            existing_ts = {m.timestamp for m in self._messages}
            new_msgs = [m for m in imported if m.timestamp not in existing_ts]
            self._messages.extend(new_msgs)
            self._messages.sort(key=lambda m: m.timestamp)
        else:
            self._messages = list(imported)
        if len(self._messages) > self._max:
            self._messages = self._messages[-self._max :]

    self._save()
    count = len(imported)
    logger.info(
        "Imported %d messages from %s (merge=%s)", count, path, merge
    )
    return count

generate_encryption_key

generate_encryption_key()

Generate a new Fernet key (32 bytes, URL-safe base64-encoded).

Returns

bytes A 44-byte URL-safe base64 string suitable for Fernet(key).

Source code in src/conversation.py

def generate_encryption_key() -> bytes:
    """Generate a new Fernet key (32 bytes, URL-safe base64-encoded).

    Returns
    -------
    bytes
        A 44-byte URL-safe base64 string suitable for ``Fernet(key)``.
    """
    from cryptography.fernet import Fernet
    return Fernet.generate_key()

load_or_create_key

load_or_create_key(key_path)

Load an existing Fernet key from key_path, or create one.

The key file is written with 0o600 permissions (owner read/write only). If the file already exists and has correct permissions its contents are returned unchanged.

Parameters

key_path: Path to the history.key file.

Returns

bytes The Fernet key bytes.

Source code in src/conversation.py

def load_or_create_key(key_path: Path) -> bytes:
    """Load an existing Fernet key from *key_path*, or create one.

    The key file is written with ``0o600`` permissions (owner read/write
    only).  If the file already exists and has correct permissions its
    contents are returned unchanged.

    Parameters
    ----------
    key_path:
        Path to the ``history.key`` file.

    Returns
    -------
    bytes
        The Fernet key bytes.
    """
    if key_path.exists():
        check_path_permissions(key_path, label="history key file")
        key = key_path.read_bytes().strip()
        if key:
            return key
        # Fall through to regenerate if file is empty / corrupt.

    key = generate_encryption_key()
    key_path.parent.mkdir(parents=True, exist_ok=True)
    # Write atomically with 0o600 permissions.
    tmp = key_path.with_suffix(".key.tmp")
    try:
        fd = os.open(tmp, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        with os.fdopen(fd, "wb") as fh:
            fh.write(key)
        tmp.chmod(0o600)
        tmp.replace(key_path)
    except OSError:
        if tmp.exists():
            tmp.unlink(missing_ok=True)
        raise

    logger.info("Generated new history encryption key: %s", key_path)
    return key

encrypt_data

encrypt_data(plaintext, key)

Encrypt plaintext with Fernet and return a base64 ciphertext string.

The returned string is ASCII-safe and can be written to a text file.

Source code in src/conversation.py

def encrypt_data(plaintext: str, key: bytes) -> str:
    """Encrypt *plaintext* with Fernet and return a base64 ciphertext string.

    The returned string is ASCII-safe and can be written to a text file.
    """
    from cryptography.fernet import Fernet
    f = Fernet(key)
    return f.encrypt(plaintext.encode("utf-8")).decode("ascii")

decrypt_data

decrypt_data(ciphertext, key)

Decrypt Fernet ciphertext and return the original plaintext.

Raises

cryptography.fernet.InvalidToken If the key is wrong or the data was tampered with.

Source code in src/conversation.py

def decrypt_data(ciphertext: str, key: bytes) -> str:
    """Decrypt Fernet *ciphertext* and return the original plaintext.

    Raises
    ------
    cryptography.fernet.InvalidToken
        If the key is wrong or the data was tampered with.
    """
    from cryptography.fernet import Fernet
    f = Fernet(key)
    return f.decrypt(ciphertext.encode("ascii")).decode("utf-8")