hendrik/interfaces/tui/app.py

import curses
import json
import threading
from datetime import datetime
import config
from .render import init_colors, draw
from .input import handle_key
from .agent import log, WELCOME_ART
from services.session_manager_neo import NeoSessionManager, NeoSession
from lib import ragroleplay
from lib import personality as personality_mod


class HendrikTUI:
    def __init__(self, llm_client, tools_definition, TOOLS, TOOL_HANDLERS,
                 build_system_prompt, agent_max_iterations):
        self.llm = llm_client
        self.tools_def = tools_definition
        self.TOOLS = TOOLS
        self.TOOL_HANDLERS = TOOL_HANDLERS
        self.build_system_prompt = build_system_prompt
        self.agent_max_iterations = agent_max_iterations
        self.character_name = config.AGENT_CHARACTER or "AI Agent"

        self.messages = None
        self.log = []
        self.input_buffer = [""]
        self.input_line = 0
        self.input_col = 0
        self.scroll = 0
        self.processing = False
        self.running = True
        self.h, self.w = 0, 0

        self.agent_thread: threading.Thread | None = None
        self.agent_done = threading.Event()

        self.session_mgr = NeoSessionManager()
        self.current_session: NeoSession | None = None

    def switch_model(self, item: dict):
        self.llm.base_url = item["base_url"]
        self.llm.model = item["model"]
        self.llm.api_key = item["api_key"]
        if self.current_session:
            self.session_mgr.update_model_info(
                self.current_session.doc_id, self._model_info()
            )

    def _model_info(self) -> dict:
        return {
            "provider": config.resolve_provider(self.llm.base_url, self.llm.model),
            "base_url": self.llm.base_url,
            "model": self.llm.model,
        }

    def _inject_user_context(self) -> str | None:
        user_id = config.TUI_USER_ID
        if not user_id:
            return None

        try:
            results = ragroleplay.user_load(
                config.ragroleplay_db_path,
                user_id=user_id,
                character=personality_mod.PERSONALITY.name
            )
            if results:
                u = results[0]
                return (
                    f'[User Context]\n'
                    f'ID: {u["id"]}\n'
                    f'Nama: {u["fullname"]} ({u["nickname"]})\n'
                    f'Family: {u.get("familyname", "-") or "-"}\n'
                    f'Character: {u.get("character", "-") or "-"}\n'
                    f'Alias: {u.get("alias", "-") or "-"}\n'
                    f'Salutation: {u.get("salutation", "-") or "-"}\n'
                    f'Persona: {u.get("persona", "-") or "-"}\n'
                    f'Telegram: {u.get("telegram_id", "-") or "-"} / @{u.get("telegram_username", "-") or "-"}\n'
                    f'XMPP: {u.get("xmpp_username", "-") or "-"}\n'
                    f'[/User Context]\n'
                    f'[PENTING: Kamu SUDAH mengenal user ini. WAJIB panggil memories_latest(character="{personality_mod.PERSONALITY.name}", user_id="{u["id"]}", limit=10) untuk mengambil riwayat percakapan terbaru. GUNAKAN data kondisi emotional/physical dari memori terbaru untuk melanjutkan state character secara natural sebelum merespon.]'
                )
            else:
                return (
                    f'[User Context: Pengguna baru — belum ada di database]\n'
                    f'[Platform: TUI]\n'
                    f'[PENTING: Kamu BELUM mengenal user ini. WAJIB tanya nama sebagai pembuka. '
                    f'Setelah nama diketahui, simpan via users_store. '
                    f'Lanjutkan percakapan natural dan proaktif melengkapi data user lainnya di pesan berikutnya.]'
                )
        except Exception:
            return None

    def new_session(self):
        self.current_session = None

        system_prompt = self.build_system_prompt(
            tools_definition=self.tools_def,
            character=config.AGENT_CHARACTER or None,
            skills=config.AGENT_SKILLS.split(",") if config.AGENT_SKILLS else None,
        )
        messages = [{"role": "system", "content": system_prompt}]

        context = self._inject_user_context()
        if context:
            messages.append({"role": "system", "content": context})

        self.messages = messages
        # ---------------------------------

        self.log.clear()
        self.scroll = 0
        self.input_buffer = [""]
        self.input_line = 0
        self.input_col = 0
        log(self, "welcome", WELCOME_ART)

    def switch_session(self, doc_id: int):
        session = self.session_mgr.get(doc_id)
        if not session:
            return
        self.current_session = session
        self.messages = list(session.messages)
        self.log.clear()
        self.scroll = 0
        log(self, "welcome", WELCOME_ART)
        for i, msg in enumerate(session.messages):
            if msg["role"] == "user":
                if i > 0:
                    log(self, "sep", "")
                log(self, "user", msg["content"])
                mi = msg.get("model_info")
                if mi:
                    self.log[-1]["model_info"] = (mi.get("provider"), mi.get("model"))
            elif msg["role"] == "assistant":
                # Log tool calls jika ada
                if msg.get("tool_calls"):
                    for tc in msg["tool_calls"]:
                        log(self, "tool_call", json.dumps({
                            "name": tc["function"]["name"],
                            "arguments": tc["function"]["arguments"],
                        }))
                # Log AI content jika ada dan ini adalah response final
                next_role = session.messages[i + 1]["role"] if i + 1 < len(session.messages) else None
                if next_role in ("user", None):
                    if msg.get("content"):
                        log(self, "ai", msg["content"])
            elif msg["role"] == "tool":
                # Tool result - tidak perlu di-log secara eksplisit
                pass
        if session.messages and session.messages[-1]["role"] == "assistant":
            log(self, "sep", "")

    def run(self):
        try:
            curses.wrapper(self._main)
        except KeyboardInterrupt:
            pass

    def _main(self, stdscr):
        curses.use_default_colors()
        init_colors()
        stdscr.keypad(True)
        curses.raw()  # Ctrl+C sebagai key code 3, bukan SIGINT → KeyboardInterrupt
        stdscr.refresh()

        self.messages = [{"role": "system", "content": self.build_system_prompt(
            tools_definition=self.tools_def,
            character=config.AGENT_CHARACTER or None,
            skills=config.AGENT_SKILLS.split(",") if config.AGENT_SKILLS else None,
        )}]
        context = self._inject_user_context()
        if context:
            self.messages.append({"role": "system", "content": context})
        log(self, "welcome", WELCOME_ART)

        while self.running:
            self.h, self.w = stdscr.getmaxyx()
            if self.h < 14 or self.w < 40:
                stdscr.erase()
                stdscr.addstr(0, 0, "Terminal too small (min 40x14)")
                stdscr.refresh()
                stdscr.getch()
                continue

            draw(self, stdscr)
            curses.curs_set(2)

            if self.processing:
                stdscr.timeout(100)
            else:
                stdscr.timeout(-1)

            try:
                key = stdscr.getch()
            except KeyboardInterrupt:
                if self.processing:
                    self.llm.cancel_requested = True
                    self.agent_done.set()
                else:
                    break
                key = -1

            handle_key(self, stdscr, key)

            if self.agent_done.is_set():
                self.agent_thread.join()
                self.agent_done.clear()
                self.processing = False
                self.agent_thread = None