Add local history management and improve the command (#74)

Now we can save/view and query (basic) the history locally. The history
module allows for modularity to switch between LocalHistory and a future
implementation of a RemoteHistory that will be stored in the servers.

We also improved the `history` command to show in the terminal the
conversation history by having the following modes:
- All history
- First history
- Last history

More modes will be implemented in the future, but for now, this is enough.

command_line_assistant/commands/history.py

@@ -4,26 +4,129 @@
 from dasbus.error import DBusError
 
 from command_line_assistant.dbus.constants import HISTORY_IDENTIFIER
+from command_line_assistant.dbus.structures import HistoryEntry
+from command_line_assistant.rendering.decorators.colors import ColorDecorator
+from command_line_assistant.rendering.decorators.text import (
+    EmojiDecorator,
+    TextWrapDecorator,
+)
+from command_line_assistant.rendering.renders.spinner import SpinnerRenderer
+from command_line_assistant.rendering.renders.text import TextRenderer
+from command_line_assistant.rendering.stream import StdoutStream
 from command_line_assistant.utils.cli import BaseCLICommand, SubParsersAction
 
 logger = logging.getLogger(__name__)
 
 
+def _initialize_spinner_renderer() -> SpinnerRenderer:
+    spinner = SpinnerRenderer(message="Loading history", stream=StdoutStream(end=""))
+    spinner.update(TextWrapDecorator())
+
+    return spinner
+
+
+def _initialize_qa_renderer(is_assistant: bool = False) -> TextRenderer:
+    text = TextRenderer(stream=StdoutStream())
+    foreground = "lightblue" if is_assistant else "lightgreen"
+    text.update(ColorDecorator(foreground=foreground))
+    text.update(EmojiDecorator("🤖"))
+    return text
+
+
+def _initialize_text_renderer() -> TextRenderer:
+    text = TextRenderer(stream=StdoutStream())
+    return text
+
+
 class HistoryCommand(BaseCLICommand):
-    def __init__(self, clear: bool) -> None:
+    def __init__(self, clear: bool, first: bool, last: bool) -> None:
         self._clear = clear
+        self._first = first
+        self._last = last
+
+        self._proxy = HISTORY_IDENTIFIER.get_proxy()
+        self._user_renderer = _initialize_qa_renderer()
+        self._assistant_renderer = _initialize_qa_renderer(is_assistant=True)
+        self._text_renderer = _initialize_text_renderer()
+        self._spinner_renderer = _initialize_spinner_renderer()
         super().__init__()
 
     def run(self) -> None:
-        proxy = HISTORY_IDENTIFIER.get_proxy()
-
         if self._clear:
-            try:
-                logger.info("Cleaning the history.")
-                proxy.ClearHistory()
-            except DBusError as e:
-                logger.info("Failed to clean the history: %s", e)
-                raise e
+            return self._clear_history()
+
+        if self._first:
+            return self._retrieve_first_conversation()
+
+        if self._last:
+            return self._retrieve_last_conversation()
+
+        return self._retrieve_all_conversations()
+
+    def _retrieve_all_conversations(self):
+        """Retrieve and display all conversations from history."""
+        try:
+            logger.info("Getting all conversations from history.")
+            response = self._proxy.GetHistory()
+            history = HistoryEntry.from_structure(response)
+
+            if history.entries:
+                for entry in history.entries:
+                    self._user_renderer.render(f"Query: {entry.query}")
+                    self._assistant_renderer.render(f"Answer: {entry.response}")
+                    self._text_renderer.render(f"Time: {entry.timestamp}")
+                    self._text_renderer.render(
+                        "-" * 50
+                    )  # Separator between conversations
+            else:
+                print("No history found.")
+        except DBusError as e:
+            logger.info("Failed to get history: %s", e)
+            raise e
+
+    def _retrieve_first_conversation(self):
+        try:
+            logger.info("Getting first conversation from history.")
+            response = self._proxy.GetFirstConversation()
+            history = HistoryEntry.from_structure(response)
+            if history.entries:
+                # Display the conversation
+                entry = history.entries[0]
+                self._user_renderer.render(f"Query: {entry.query}")
+                self._assistant_renderer.render(f"Answer: {entry.response}")
+                self._text_renderer.render(f"Time: {entry.timestamp}")
+            else:
+                print("No history found.")
+        except DBusError as e:
+            logger.info("Failed to get first conversation: %s", e)
+            raise e
+
+    def _retrieve_last_conversation(self):
+        try:
+            logger.info("Getting last conversation from history.")
+            response = self._proxy.GetLastConversation()
+
+            # Handle and display the response
+            history = HistoryEntry.from_structure(response)
+            if history.entries:
+                # Display the conversation
+                entry = history.entries[0]
+                self._user_renderer.render(f"Query: {entry.query}")
+                self._assistant_renderer.render(f"Answer: {entry.response}")
+                self._text_renderer.render(f"Time: {entry.timestamp}")
+            else:
+                print("No history found.")
+        except DBusError as e:
+            logger.info("Failed to get last conversation: %s", e)
+            raise e
+
+    def _clear_history(self) -> None:
+        try:
+            logger.info("Cleaning the history.")
+            self._proxy.ClearHistory()
+        except DBusError as e:
+            logger.info("Failed to clean the history: %s", e)
+            raise e
 
 
 def register_subcommand(parser: SubParsersAction):
@@ -38,10 +141,22 @@ def register_subcommand(parser: SubParsersAction):
         help="Manage conversation history",
     )
     history_parser.add_argument(
-        "--clear", action="store_true", help="Clear the history."
+        "--clear",
+        action="store_true",
+        help="Clear the history.",
+    )
+    history_parser.add_argument(
+        "--first",
+        action="store_true",
+        help="Get the first conversation from history.",
+    )
+    history_parser.add_argument(
+        "--last",
+        action="store_true",
+        help="Get the last conversation from history.",
     )
     history_parser.set_defaults(func=_command_factory)
 
 
 def _command_factory(args: Namespace) -> HistoryCommand:
-    return HistoryCommand(args.clear)
+    return HistoryCommand(args.clear, args.first, args.last)

command_line_assistant/config/schemas.py

@@ -38,9 +38,8 @@ class HistorySchema:
 
     enabled: bool = True
     file: Union[str, Path] = Path(  # type: ignore
-        "~/.local/share/command-line-assistant/command-line-assistant_history.json"
+        "/var/lib/command-line-assistant/history.json"
     )
-    max_size: int = 100
 
     def __post_init__(self):
         self.file: Path = Path(self.file).expanduser()

command_line_assistant/dbus/interfaces.py

@@ -9,7 +9,8 @@
     HistoryEntry,
     Message,
 )
-from command_line_assistant.history import handle_history_read, handle_history_write
+from command_line_assistant.history.manager import HistoryManager
+from command_line_assistant.history.plugins.local import LocalHistory
 
 
 @dbus_interface(QUERY_IDENTIFIER.interface_name)
@@ -19,11 +20,13 @@ class QueryInterface(InterfaceTemplate):
     @property
     def RetrieveAnswer(self) -> Structure:
         """This method is mainly called by the client to retrieve it's answer."""
-        llm_response = submit(
-            self.implementation.query.message, self.implementation.config
-        )
+        query = self.implementation.query.message
+        llm_response = submit(query, self.implementation.config)
         message = Message()
         message.message = llm_response
+        manager = HistoryManager(self.implementation.config, LocalHistory)
+        current_history = manager.read()
+        manager.write(current_history, query, llm_response)
         return Message.to_structure(message)
 
     @emits_properties_changed
@@ -34,11 +37,47 @@ def ProcessQuery(self, query: Structure) -> None:
 
 @dbus_interface(HISTORY_IDENTIFIER.interface_name)
 class HistoryInterface(InterfaceTemplate):
-    @property
     def GetHistory(self) -> Structure:
-        history = HistoryEntry()
-        history.entries = handle_history_read(self.implementation.config)
-        return history.to_structure(history)
+        """Get all conversations from history."""
+        manager = HistoryManager(self.implementation.config, LocalHistory)
+        history = manager.read()
+
+        history_entry = HistoryEntry()
+        if history.history:
+            [history_entry.set_from_dict(entry.to_dict()) for entry in history.history]
+        else:
+            history_entry.entries = []
+
+        return HistoryEntry.to_structure(history_entry)
+
+    # Add new methods with parameters
+    def GetFirstConversation(self) -> Structure:
+        """Get first conversation from history."""
+        manager = HistoryManager(self.implementation.config, LocalHistory)
+        history = manager.read()
+        history_entry = HistoryEntry()
+        if history.history:
+            last_entry = history.history[0]
+            history_entry.set_from_dict(last_entry.to_dict())
+        else:
+            history_entry.entries = []
+
+        return HistoryEntry.to_structure(history_entry)
+
+    def GetLastConversation(self) -> Structure:
+        """Get last conversation from history."""
+        manager = HistoryManager(self.implementation.config, LocalHistory)
+        history = manager.read()
+        history_entry = HistoryEntry()
+
+        if history.history:
+            last_entry = history.history[-1]
+            history_entry.set_from_dict(last_entry.to_dict())
+        else:
+            history_entry.entries = []
+
+        return HistoryEntry.to_structure(history_entry)
 
     def ClearHistory(self) -> None:
-        handle_history_write(self.implementation.config.history.file, [], "")
+        manager = HistoryManager(self.implementation.config, LocalHistory)
+        manager.clear()

command_line_assistant/dbus/structures.py

@@ -1,5 +1,5 @@
 from dasbus.structure import DBusData
-from dasbus.typing import Str
+from dasbus.typing import List, Str
 
 
 class Message(DBusData):
@@ -18,15 +18,58 @@ def message(self, value: Str) -> None:
         self._message = value
 
 
+class HistoryItem(DBusData):
+    """Represents a single history item with query and response"""
+
+    def __init__(self) -> None:
+        self._query: Str = ""
+        self._response: Str = ""
+        self._timestamp: Str = ""
+        super().__init__()
+
+    @property
+    def query(self) -> Str:
+        return self._query
+
+    @query.setter
+    def query(self, value: Str) -> None:
+        self._query = value
+
+    @property
+    def response(self) -> Str:
+        return self._response
+
+    @response.setter
+    def response(self, value: Str) -> None:
+        self._response = value
+
+    @property
+    def timestamp(self) -> Str:
+        return self._timestamp
+
+    @timestamp.setter
+    def timestamp(self, value: Str) -> None:
+        self._timestamp = value
+
+
 class HistoryEntry(DBusData):
     def __init__(self) -> None:
-        self._entries: list[str] = []
+        self._entries: List[HistoryItem] = []
         super().__init__()
 
     @property
-    def entries(self) -> list[str]:
+    def entries(self) -> List[HistoryItem]:
         return self._entries
 
     @entries.setter
-    def entries(self, value: list[str]) -> None:
+    def entries(self, value: List[HistoryItem]) -> None:
+        # This handles setting from DBus structure
         self._entries = value
+
+    def set_from_dict(self, entry: dict) -> None:
+        """Separate method to handle conversion from history dictionary"""
+        item = HistoryItem()
+        item.query = entry["interaction"]["query"]["text"] or ""
+        item.response = entry["interaction"]["response"]["text"] or ""
+        item.timestamp = entry["timestamp"] or ""
+        self._entries.append(item)