From 8d3b48c4c210b7d3642280d83c3d4c671956fa1c Mon Sep 17 00:00:00 2001 From: Dita Aji Pratama Date: Fri, 16 Oct 2026 14:13:34 +0700 Subject: [PATCH] Fixing inference cancellation --- interfaces/tui/actions.py | 10 +- interfaces/tui/agent.py | 53 +++- interfaces/tui/app.py | 5 +- interfaces/tui/render.py | 8 +- lib/agent_loop.py | 88 ++++++- services/llm_client.py | 415 ++++++++++++++++++++++++++------ services/session_manager_neo.py | 20 ++ services/telegram_client.py | 14 ++ services/xmpp_client.py | 21 ++ 9 files changed, 539 insertions(+), 95 deletions(-) diff --git a/interfaces/tui/actions.py b/interfaces/tui/actions.py index a77922a..09ac363 100644 --- a/interfaces/tui/actions.py +++ b/interfaces/tui/actions.py @@ -39,9 +39,13 @@ def _exit_app(app, stdscr): def _cancel_inference(app, stdscr): - app.llm.cancel_requested = True - app.agent_done.set() - log(app, "system", " Inference cancelled by user") + # Hanya menandai + membangunkan stream; penghentian loop dan pembersihan + # log dilakukan oleh agent thread sendiri (CursesObserver.on_cancelled), + # supaya tidak ada balapan tulis terhadap app.log dari dua thread. + # `agent_done` sengaja tidak di-set: main loop harus menunggu thread agen + # benar-benar berhenti, bukan berpura-pura selesai lalu mulai prompt baru + # di atas messages yang sedang ditulisnya. + app.llm.cancel() def _send_prompt(app, stdscr): diff --git a/interfaces/tui/agent.py b/interfaces/tui/agent.py index f1dfb1d..a7f99a0 100644 --- a/interfaces/tui/agent.py +++ b/interfaces/tui/agent.py @@ -27,6 +27,14 @@ def _persist_msg(app, message): app.current_session.doc_id, message["role"], message["content"], **extra ) + +def _truncate_msg(app, count): + """Buang `count` pesan terakhir dari DB session (dipanggil saat turn + dibatalkan), supaya riwayat tersimpan identik dengan app.messages.""" + if not app.current_session or count <= 0: + return + app.session_mgr.remove_last_messages(app.current_session.doc_id, count) + WELCOME_ART = """ ,--. ,--.,------.,--. ,--.,------. ,------. ,--.,--. ,--. | '--' || .---'| ,'.| || .-. \\ | .--. '| || .' / @@ -224,6 +232,23 @@ class CursesObserver(agent_loop.AgentObserver): def on_max_iterations(self, max_iterations=None): log(self.app, "error", "Max iterations reached without final answer.") + def on_cancelled(self, result): + """Turn dihentikan user. Berjalan di thread agen — satu-satunya writer + app.log, jadi tidak ada balapan dengan main loop.""" + self._pop_thinking() + # Sisa turn dibuang dari konteks (assistant parsial dengan tool_calls + # tanpa hasil = payload tidak valid), tapi teks yang sudah terbaca tetap + # ditampilkan supaya user tidak kehilangan apa yang sudah dihasilkan. + partial = result.partial or "" + self._drop_placeholder() + if partial.strip(): + entry = log(self.app, "ai", partial, model_info=self.pair, + usage=result.usage) + entry["cancelled"] = True + log(self.app, "system", " Inference cancelled by user") + log(self.app, "sep", "") + self._follow() + # ----------------------------------------------------------------- utils --- def _follow(self): if self.app.scroll_follow: @@ -251,16 +276,23 @@ def _run_agent(app): app.messages, on_append=lambda msg: _persist_msg(app, msg), keep_model_info=True, + on_truncate=lambda n: _truncate_msg(app, n), ) - agent_loop.run_agent_loop( - store, - app.llm, - app.TOOLS, - app.TOOL_HANDLERS, - app.agent_max_iterations, - observer=CursesObserver(app), - tool_reminder=None, - ) + try: + agent_loop.run_agent_loop( + store, + app.llm, + app.TOOLS, + app.TOOL_HANDLERS, + app.agent_max_iterations, + observer=CursesObserver(app), + tool_reminder=None, + ) + finally: + # Selalu tandai selesai, meskipun engine melempar. Tanpa ini UI bisa + # terkunci di PROCESSING selamanya kalau ada exception tak terduga + # (mis. saat pembatalan). + app.agent_done.set() def submit(app, stdscr): @@ -289,6 +321,9 @@ def submit(app, stdscr): _add_msg(app, "user", query) + # Turn baru: bersihkan state cancel sebelumnya supaya cancel yang tertunda + # tidak membatalkan prompt yang baru dikirim. + app.llm.begin_turn() app.agent_done.clear() app.agent_thread = threading.Thread( target=_run_agent, diff --git a/interfaces/tui/app.py b/interfaces/tui/app.py index eaab6d5..540f91f 100644 --- a/interfaces/tui/app.py +++ b/interfaces/tui/app.py @@ -289,8 +289,9 @@ class HendrikTUI: key = keycodes.read_key(stdscr, timeout_ms) except KeyboardInterrupt: if self.processing: - self.llm.cancel_requested = True - self.agent_done.set() + # Sama seperti Esc: cukup minta pembatalan, biarkan thread + # agen yang membersihkan log lalu menandai dirinya selesai. + self.llm.cancel() else: break key = -1 diff --git a/interfaces/tui/render.py b/interfaces/tui/render.py index e9ffbc6..bdc3b8d 100644 --- a/interfaces/tui/render.py +++ b/interfaces/tui/render.py @@ -186,7 +186,13 @@ def draw_chat(app, stdscr): if info_line: _add_row([(C_AI, info_line, False)]) - label = f" {app.character_name} ({item['time']}) " + # Teks parsial dari turn yang dibatalkan: diberi penanda supaya + # user tahu ini bukan jawaban lengkap (dan tidak ikut dikirim + # sebagai konteks pada prompt berikutnya). + if item.get("cancelled"): + label = f" {app.character_name} ({item['time']}) \u2717 cancelled " + else: + label = f" {app.character_name} ({item['time']}) " _add_row([(C_AI, label)]) _wrap_render(text, indent=1, color=C_INPUT, bold=False) # Baris token usage setelah teks pesan respons diff --git a/lib/agent_loop.py b/lib/agent_loop.py index 0ebb275..95203a5 100644 --- a/lib/agent_loop.py +++ b/lib/agent_loop.py @@ -53,34 +53,58 @@ class SessionStore: `messages` state aktif yang dikirim ke LLM (list dict). `on_append` dipanggil setelah tiap pesan masuk — untuk persistensi (TUI/GUI/API via NeoSessionManager). Boleh None. + `on_truncate` dipanggil dengan jumlah pesan yang harus dihapus dari ekor + saat turn dibatalkan. Boleh None. `keep_model_info` False (default): kunci `model_info` dibuang sebelum pesan masuk list, jadi payload ke provider identik dengan sebelum refactor. Interface yang menampilkan info model menyalakannya. """ - def __init__(self, messages, on_append=None, keep_model_info=False): + def __init__(self, messages, on_append=None, keep_model_info=False, + on_truncate=None): self.messages = messages self._on_append = on_append self.keep_model_info = keep_model_info + self._on_truncate = on_truncate + # Berapa pesan yang sudah ditambahkan loop ini pada turn berjalan. + self.turn_appends = 0 def append(self, message): if not self.keep_model_info: message.pop('model_info', None) message.pop('usage', None) # metadata display, bukan payload LLM self.messages.append(message) + self.turn_appends += 1 if self._on_append: self._on_append(message) + def discard_turn(self): + """Buang semua pesan yang ditulis loop ini pada turn yang dibatalkan. + + Penting: assistant message dengan `tool_calls` yang hasilnya tidak ikut + lengkap membuat konteks jadi tidak valid bagi banyak provider (400 karena + tool_call_id menggantung). Yang tersimpan di DB pun ikut dipangkas lewat + `on_truncate`, supaya session yang dibuka lagi tidak membawa sisa turn.""" + n = self.turn_appends + if n > 0: + del self.messages[len(self.messages) - n:] + self.turn_appends = 0 + if n and self._on_truncate: + self._on_truncate(n) + class AgentResult: # Hasil satu kali jalan loop. - def __init__(self, content=None, session_ended=False, model_info=None, warning=None, is_multimodal=False, usage=None): + def __init__(self, content=None, session_ended=False, model_info=None, warning=None, + is_multimodal=False, usage=None, cancelled=False, partial=""): self.content = content # str | None self.session_ended = session_ended # tool end_session pernah dipanggil self.model_info = model_info or {} # model yang benar-benar menjawab self.warning = warning # mis. fallback chain model self.is_multimodal = is_multimodal self.usage = usage # token turn ini: prompt/completion/total (dict | None) + self.cancelled = cancelled # turn dihentikan user sebelum selesai + self.partial = partial # teks yang sempat dihasilkan sebelum berhenti class AgentObserver: @@ -129,6 +153,10 @@ class AgentObserver: def on_max_iterations(self, max_iterations=None): pass + def on_cancelled(self, result): + """Turn dihentikan user sebelum ada jawaban final.""" + pass + class ConsoleObserver(AgentObserver): """Observer default proses headless (service bot): log gorilla ke stdout. @@ -163,9 +191,20 @@ class ConsoleObserver(AgentObserver): loggorilla.prcss("agent_loop", f"Max iterations ({max_iterations}) reached") + def on_cancelled(self, result): + loggorilla.prcss("agent_loop", "Turn cancelled by user") + # ------------------------------------------------------------------- loop --- +def _cancelled(llm_client) -> bool: + """Apakah user meminta turn ini dibatalkan? + + Hanya bertanya ke LLM client (single source of truth state cancel) — engine + tidak punya salinan flag sendiri, jadi tidak ada yang bisa tersinkron rusak.""" + return bool(getattr(llm_client, "cancel_requested", False)) + + def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, observer=None, tool_reminder=None, disable_reasoning=None): """Jalankan loop LLM <-> tool sampai ada jawaban final atau batas iterasi. @@ -176,6 +215,16 @@ def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, Return AgentResult. Engine tidak mengklasifikasikan konten sebagai error; itu keputusan interface (lihat CursesObserver di interfaces/tui/agent.py). + + Pembatalan (cancel) dicek pada setiap boundary: sebelum panggilan LLM, + tepat setelah respons tiba, dan sebelum tiap tool dijalankan. Saat user + membatalkan, loop berhenti seketika dan sisa turn dibuang dari konteks lewat + `store.discard_turn()`. Interface yang ingin menawarkan "lanjutkan" dapat + menaruh kembali teks parsial (`result.partial`) sebagai pesan user. + + State cancel TIDAK direset di sini: flag hanya boleh dibersihkan oleh + interface tepat saat pesan user dikirim. Kalau engine yang membersihkannya, + cancel yang ditekan sebelum thread ini sempat jalan akan hilang begitu saja. """ if observer is None: observer = ConsoleObserver() @@ -206,9 +255,31 @@ def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, if isinstance(pt, (int, float)): turn_usage["context_tokens"] = int(pt) + def _stop(partial=""): + """Akhiri turn karena dibatalkan: buang sisa turn dari konteks, beri + tahu observer, kembalikan hasil tanpa jawaban final.""" + result = AgentResult( + content=None, + session_ended=session_ended, + model_info=model_info, + warning=warning, + usage=dict(turn_usage) if any_usage else None, + cancelled=True, + partial=partial, + ) + store.discard_turn() + observer.on_cancelled(result) + observer.on_run_end() + return result + observer.on_run_start() for step in range(max_iterations): + # Cancel bisa datang dari mana saja: dari antrean sebelum thread ini + # sempat jalan, atau dari user saat tool masih berjalan. + if _cancelled(llm_client): + return _stop() + observer.on_step_start(step) response = llm_client.chat( @@ -225,6 +296,12 @@ def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, warning = getattr(response, "warning", None) usage = getattr(response, "usage", None) _accumulate(usage) + + # Stream dipotong user: jangan tulis sisa turn ke konteks. + if getattr(response, "cancelled", False) or _cancelled(llm_client): + partial = response.content if isinstance(response.content, str) else "" + return _stop(partial) + observer.on_response(model_info, warning, usage) if response.tool_calls: @@ -246,6 +323,11 @@ def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, [tc['function']['name'] for tc in response.tool_calls]) for tc in response.tool_calls: + # Sebelum eksekusi tool: batalkan tanpa menjalankan efek samping + # (tool adalah bagian paling berbahaya untuk disetop tengah — + # lebih baik tidak dijalankan sama sekali daripada separuh jalan). + if _cancelled(llm_client): + return _stop(response.content or "") observer.on_tool_call(tc['function']['name'], tc['function']['arguments']) if tc['function']['name'] == 'end_session': session_ended = True @@ -265,6 +347,8 @@ def run_agent_loop(store, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, 'tool_call_id': tc['id'], 'content': str(result), }) + if _cancelled(llm_client): + return _stop(response.content or "") observer.on_step_end(step) continue diff --git a/services/llm_client.py b/services/llm_client.py index 073a971..b795071 100644 --- a/services/llm_client.py +++ b/services/llm_client.py @@ -1,9 +1,161 @@ import json import socket +import threading from lib import gadget, moop import urllib.request import urllib.error + +class StreamCancelled(Exception): + """User meminta pembatalan saat stream berjalan. + + `partial` menyimpan teks yang sudah ter-stream supaya tidak hilang. + Sengaja bukan turunan _LLMFail: rantai fallback/auto-switch tidak boleh + memperlakukan cancel sebagai kegagalan provider lalu mencoba model + berikutnya (jumlah percobaan tidak terbatas).""" + + def __init__(self, partial=""): + super().__init__("Stream cancelled by user") + self.partial = partial + + +class _CancelState: + """Penghubung thread UI (yang membatalkan) dan thread agen (pembaca SSE). + + Masalah yang dipecahkan: pembacaan stream memblokir di dalam recv() selama + provider diam — model masih 'thinking' sebelum token pertama, header belum + dikirim, atau koneksi menggantung. Flag cancel saja tidak cukup karena ia + hanya diperiksa ketika satu baris penuh tiba, jadi Cancel/Stop baru + bereaksi setelah provider mengirim sesuatu (bisa puluhan detik, atau tidak + pernah). + + Solusi: objek socket koneksi aktif dicatat begitu TCP/TLS terbentuk; saat + cancel diminta, shutdown(SHUT_RDWR) pada objek itu membangunkan recv() + seketika di thread mana pun. Yang dipegang adalah objek socket-nya (bukan + fd mentah), sehingga balapan antara cancel dan penutupan koneksi tidak + pernah mengenai fd yang sudah dipakai ulang oleh koneksi lain. + + Beberapa stream bisa berjalan bersamaan pada satu klien (Telegram/XMPP + memproses tiap pesan di thread sendiri dan berbagi satu LLMClient), jadi + semua socket aktif dilacak dan dibangunkan sekaligus — persis seperti flag + global lama yang dilihat oleh setiap loop.""" + + def __init__(self): + self._lock = threading.Lock() + self._socks = [] + self.requested = False + + def set_socket(self, sock): + """Daftarkan socket stream aktif. Jika cancel sudah diminta sebelum + stream sempat terbentuk, bangunkan sekarang juga.""" + wake = None + with self._lock: + if self.requested: + wake = [sock] + else: + # Bersihkan entri yang socket-nya sudah ditutup (mis. sisa + # stream yang selesai tanpa sempat release) supaya daftar tidak + # tumbuh tanpa batas. + self._socks = [s for s in self._socks if not self._dead(s)] + if sock not in self._socks: + self._socks.append(sock) + if wake: + self._shutdown_all(wake) + + def release(self, sock): + """Panggil ketika pembacaan selesai: setelah ini cancel tidak pernah + menyentuh socket ini lagi. Hanya melepas satu socket — stream thread + lain pada klien yang sama tidak boleh ikut terputus pendaftarannya.""" + if sock is None: + return + with self._lock: + if sock in self._socks: + self._socks.remove(sock) + + def request(self): + with self._lock: + self.requested = True + socks = self._socks + self._socks = [] + self._shutdown_all(socks) + + def reset(self): + # Daftar socket TIDAK dikosongkan: pada klien yang dipakai bersama + # (Telegram/XMPP) bisa ada stream thread lain yang masih berjalan dan + # harus tetap bisa dibatalkan. Socket mati tersapu sendiri oleh + # set_socket/release. + with self._lock: + self.requested = False + + def _shutdown_all(self, socks): + for s in socks: + self._shutdown(s) + + @staticmethod + def _dead(sock): + """True jika socket sudah ditutup (fileno() tidak bisa dipakai lagi).""" + try: + sock.fileno() + except (OSError, ValueError): + return True + return False + + @staticmethod + def _shutdown(sock): + """Putus koneksi sekarang juga. Aman dipanggil dua kali atau pada + socket yang sudah ditutup: efek yang dibutuhkan hanya membangunkan + recv() yang terblokir — sekaligus menghentikan generation di sisi + provider, jadi token tidak ditagih untuk output yang dibuang.""" + try: + sock.shutdown(socket.SHUT_RDWR) + except (OSError, ValueError): + pass + except Exception: + pass + + +class _CapturingHandler: + """Campuran handler urllib: mendaftarkan socket koneksi ke _CancelState + tepat setelah connect(), sehingga cancel bisa membangunkan pembacaan + header maupun body yang terblokir.""" + + def __init__(self, state, base_class): + super().__init__() + self._state = state + self._base = base_class + + def do_open(self, http_class, req, **http_conn_args): + state = self._state + captured = [] + + class _Conn(http_class): + def connect(self): + http_class.connect(self) + sock = getattr(self, "sock", None) + if sock is not None: + captured.append(sock) + state.set_socket(sock) + + try: + return self._base.do_open(self, _Conn, req, **http_conn_args) + except BaseException: + # do_open menutup socket-nya sendiri saat gagal: lepas pendaftaran + # supaya thread cancel tidak menyentuhnya lagi. + for sock in captured: + state.release(sock) + raise + + +class _HTTPHandler(_CapturingHandler, urllib.request.HTTPHandler): + def __init__(self, state): + _CapturingHandler.__init__(self, state, urllib.request.HTTPHandler) + + +class _HTTPSHandler(_CapturingHandler, urllib.request.HTTPSHandler): + def __init__(self, state): + _CapturingHandler.__init__(self, state, urllib.request.HTTPSHandler) + + class _LLMFail(Exception): """ - 'auth' : 401/403 → rotasi key (coba key berikutnya di provider yang sama) @@ -21,7 +173,11 @@ class LLMClient: self.model = model self.api_key = api_key self.timeout = timeout - self.cancel_requested = False + + # State cancel + opener yang menangkap socket koneksi (_CancelState). + self._cancel = _CancelState() + self._opener = urllib.request.build_opener( + _HTTPHandler(self._cancel), _HTTPSHandler(self._cancel)) # Chaining kandidat set model per tipe ('llm', 'imagevision', ...) self._chains: dict[str, list[dict]] = {} @@ -29,12 +185,51 @@ class LLMClient: self._chain_key: str | None = None self._chain_index = 0 + # -------------------------------------------------------- cancellation --- + + @property + def cancel_requested(self) -> bool: + return self._cancel.requested + + @cancel_requested.setter + def cancel_requested(self, value): + # Kompatibel dengan pemanggil lama (`llm.cancel_requested = True/False`). + if value: + self.cancel() + else: + self.begin_turn() + + def cancel(self): + """Minta inference berhenti. Aman dipanggil dari thread UI mana pun, + kapan pun — termasuk saat stream belum terbentuk atau tool sedang + berjalan. Membangunkan pembacaan socket yang terblokir seketika.""" + self._cancel.request() + + def begin_turn(self): + """Bersihkan state cancel pada awal turn baru (dipanggil interface saat + user mengirim pesan). Cancel turn sebelumnya tidak boleh merembes ke + turn berikutnya.""" + self._cancel.reset() + + def _cancelled_message(self, partial=""): + """Respons untuk turn yang dibatalkan user. + + Teks parsial dikembalikan apa adanya di `content` dan ditandai lewat + atribut `cancelled` — penanda visual ("dibatalkan") adalah urusan + interface, bukan bagian dari isi pesan. Rantai model tidak dianggap + gagal: cancel adalah perintah user, bukan masalah provider.""" + result = self.Message({'content': partial or "", 'tool_calls': None}) + result.cancelled = True + return result + class Message: def __init__(self, msg, model_info=None): raw_content = msg.get('content', '') # Ambil konten mentah self.content = gadget.strip_thinking(raw_content) if isinstance(raw_content, str) else raw_content # Auto-strip dari content self.tool_calls = msg.get('tool_calls', None) # Ambil tool calls self.warning = None + # True jika pesan ini adalah hasil turn yang dihentikan user. + self.cancelled = False # Pemakaian token dari provider (prompt/completion/total) jika ada. self.usage = None # Model yang benar-benar dipakai menghasilkan pesan ini. Bisa berbeda @@ -129,15 +324,24 @@ class LLMClient: def chat(self, messages, tools=None, on_stream_chunk=None, disable_reasoning=False, tool_reminder=None): + if self._cancel.requested: + # Cancel diminta sebelum permintaan sempat berangkat (mis. saat + # tool masih berjalan di langkah sebelumnya). Jangan mulai request + # baru — kembalikan turn yang ditandai sebagai dibatalkan. + return self._cancelled_message("") + if not self._chains: # kalau bukan chain - result = self._chat_once( - self.base_url, self.model, self.api_key, - messages, - tools=tools, - on_stream_chunk=on_stream_chunk, - disable_reasoning=disable_reasoning, - tool_reminder=tool_reminder, - ) + try: + result = self._chat_once( + self.base_url, self.model, self.api_key, + messages, + tools=tools, + on_stream_chunk=on_stream_chunk, + disable_reasoning=disable_reasoning, + tool_reminder=tool_reminder, + ) + except StreamCancelled as e: + return self._cancelled_message(e.partial) result.model_info = self._model_info() return result @@ -204,6 +408,13 @@ class LLMClient: # permintaan) supaya pemanggil bisa menampilkan data yang jujur. result.model_info = self._model_info(cand) return result + except StreamCancelled as e: + # Cancel user: hentikan rantai total — bukan kegagalan provider, + # jadi jangan lompat ke kandidat model berikutnya. + self._chain_index = idx + result = self._cancelled_message(e.partial) + result.model_info = self._model_info(cand) + return result except _LLMFail as e: last = e if e.kind == "auth" and idx + 1 < len(chain) and self._same_target(chain[idx + 1], cand): @@ -285,87 +496,103 @@ class LLMClient: stream_started = False usage = None # prompt/completion/total tokens dari chunk terakhir + my_sock = None try: - self.cancel_requested = False - with urllib.request.urlopen(req, timeout=self.timeout) as resp: - # Streaming: baca line by line - for line in resp: - if self.cancel_requested: - # Stream cancellation - full_content += "\n\n[Stream cancelled by user]" - break + with self._opener.open(req, timeout=self.timeout) as resp: + # Cadangan pendaftaran socket (jika handler capture tidak + # sempat mengisi, mis. perubahan internal urllib): cukup untuk + # membangunkan pembacaan body. + my_sock = self._register_socket(resp) + try: + # Streaming: baca line by line + for line in resp: + if self._cancel.requested: + raise StreamCancelled(full_content) - line = line.decode('utf-8').strip() - if not line or not line.startswith('data: '): - continue + line = line.decode('utf-8').strip() + if not line or not line.startswith('data: '): + continue - data_str = line[6:] # Hapus "data: " prefix - if data_str == '[DONE]': - break + data_str = line[6:] # Hapus "data: " prefix + if data_str == '[DONE]': + break - try: - chunk = json.loads(data_str) - except json.JSONDecodeError: - continue + try: + chunk = json.loads(data_str) + except json.JSONDecodeError: + continue - # Pemakaian token datang di chunk penutup (choices kosong - # pada banyak provider) — ambil selama masih None / terbaru. - if chunk.get('usage'): - usage = chunk['usage'] + # Pemakaian token datang di chunk penutup (choices kosong + # pada banyak provider) — ambil selama masih None / terbaru. + if chunk.get('usage'): + usage = chunk['usage'] - # Parse delta dari chunk - choices = chunk.get('choices') or [] - delta = choices[0].get('delta', {}) if choices else {} - finish_reason = choices[0].get('finish_reason', None) if choices else None + # Parse delta dari chunk + choices = chunk.get('choices') or [] + delta = choices[0].get('delta', {}) if choices else {} + finish_reason = choices[0].get('finish_reason', None) if choices else None - # Stream reasoning content jika ada - if 'reasoning_content' in delta: - reasoning_content += delta['reasoning_content'] + # Stream reasoning content jika ada + if 'reasoning_content' in delta: + reasoning_content += delta['reasoning_content'] - # Stream tool_calls jika ada - if 'tool_calls' in delta: - tool_calls = delta['tool_calls'] - for tc in tool_calls: - idx = tc.get('index', 0) - # Pastikan list cukup panjang - while len(full_tool_calls) <= idx: - full_tool_calls.append({ - "id": "", - "type": "function", - "function": {"name": "", "arguments": ""} - }) + # Stream tool_calls jika ada + if 'tool_calls' in delta: + tool_calls = delta['tool_calls'] + for tc in tool_calls: + idx = tc.get('index', 0) + # Pastikan list cukup panjang + while len(full_tool_calls) <= idx: + full_tool_calls.append({ + "id": "", + "type": "function", + "function": {"name": "", "arguments": ""} + }) - # Update ID - if 'id' in tc and tc['id']: - full_tool_calls[idx]['id'] = tc['id'] + # Update ID + if 'id' in tc and tc['id']: + full_tool_calls[idx]['id'] = tc['id'] - # Update function name - if 'function' in tc and 'name' in tc['function']: - full_tool_calls[idx]['function']['name'] += tc['function']['name'] + # Update function name + if 'function' in tc and 'name' in tc['function']: + full_tool_calls[idx]['function']['name'] += tc['function']['name'] - # Update arguments - if 'function' in tc and 'arguments' in tc['function']: - full_tool_calls[idx]['function']['arguments'] += tc['function']['arguments'] + # Update arguments + if 'function' in tc and 'arguments' in tc['function']: + full_tool_calls[idx]['function']['arguments'] += tc['function']['arguments'] - # Stream content (text response atau multimodal blocks) - if 'content' in delta: - chunk = delta['content'] - if isinstance(chunk, str): - full_content += chunk or "" - if chunk: - stream_started = True - if on_stream_chunk and chunk: - on_stream_chunk(chunk) - elif isinstance(chunk, list): - full_multimodal_content.extend(chunk) - for block in chunk: - if isinstance(block, dict) and block.get('type') == 'text': - t = block.get('text', '') - if t: - full_content += t - stream_started = True - if on_stream_chunk: - on_stream_chunk(t) + # Stream content (text response atau multimodal blocks) + if 'content' in delta: + chunk = delta['content'] + if isinstance(chunk, str): + full_content += chunk or "" + if chunk: + stream_started = True + if on_stream_chunk and chunk: + on_stream_chunk(chunk) + elif isinstance(chunk, list): + full_multimodal_content.extend(chunk) + for block in chunk: + if isinstance(block, dict) and block.get('type') == 'text': + t = block.get('text', '') + if t: + full_content += t + stream_started = True + if on_stream_chunk: + on_stream_chunk(t) + finally: + # Lepas pendaftaran begitu pembacaan selesai: socket akan + # ditutup oleh with-block, dan thread cancel tidak boleh + # menyentuhnya lagi. Hanya socket milik panggilan ini — + # stream thread lain pada klien yang sama tetap terdaftar. + self._cancel.release(my_sock) + my_sock = None + + # shutdown() dari thread cancel bisa membuat pembacaan berakhir rapi + # seperti EOF — pastikan itu tidak disalahartikan sebagai jawaban + # lengkap. + if self._cancel.requested: + raise StreamCancelled(full_content) # Build final response if full_multimodal_content: @@ -402,7 +629,11 @@ class LLMClient: message['tool_calls'] = valid_tool_calls response = {'choices': [{'message': message}]} + except StreamCancelled: + raise except urllib.error.HTTPError as e: + if self._cancel.requested: + raise StreamCancelled(full_content) body_text = "" try: body_text = e.read().decode('utf-8', errors='replace') @@ -436,14 +667,20 @@ class LLMClient: raise _LLMFail("auth", f"HTTP {e.code} {e.reason}{detail}") raise _LLMFail("unreachable", f"HTTP {e.code} {e.reason}{detail}") except urllib.error.URLError as e: + if self._cancel.requested: + raise StreamCancelled(full_content) if stream_started: return self.Message({'content': f"Error: {e.reason}", 'tool_calls': None}) raise _LLMFail("unreachable", f"Connection error: {e.reason}") except (TimeoutError, socket.timeout) as e: + if self._cancel.requested: + raise StreamCancelled(full_content) if stream_started: return self.Message({'content': f"Error: Timeout", 'tool_calls': None}) raise _LLMFail("unreachable", f"Timeout") except Exception as e: + if self._cancel.requested: + raise StreamCancelled(full_content) if stream_started: return self.Message({'content': f"Error: {str(e)}", 'tool_calls': None}) raise _LLMFail("unreachable", f"Error: {str(e)}") @@ -483,3 +720,25 @@ class LLMClient: result = self.Message(message) result.usage = usage return result + + @staticmethod + def _socket_of(resp): + """Objek socket di balik HTTPResponse; None jika struktur urllib + berbeda dari yang diharapkan (maka fallback wake tidak tersedia).""" + try: + fp = resp.fp + raw = getattr(fp, "raw", None) + sock = getattr(raw, "_sock", None) + return sock if isinstance(sock, socket.socket) else None + except (AttributeError, TypeError, ValueError): + return None + + def _register_socket(self, resp): + """Daftarkan socket respons ke state cancel; kembalikan socket-nya + supaya pemanggil bisa melepasnya sendiri saat selesai (tidak menyentuh + stream thread lain). Jika cancel sudah diminta, set_socket langsung + memutusnya sehingga pembacaan pertama tidak sempat terblokir.""" + sock = self._socket_of(resp) + if sock is not None: + self._cancel.set_socket(sock) + return sock diff --git a/services/session_manager_neo.py b/services/session_manager_neo.py index 1d5c8d4..83cd459 100644 --- a/services/session_manager_neo.py +++ b/services/session_manager_neo.py @@ -136,6 +136,26 @@ class NeoSessionManager: def update_model_info(self, doc_id: int, model_info: dict) -> bool: return self._table.update({"model_info": model_info}, doc_ids=[doc_id]) + def remove_last_messages(self, doc_id: int, count: int) -> bool: + """Hapus `count` pesan terakhir dari session. + + Dipakai saat turn dibatalkan: engine sudah membuang sisa turn dari + konteks aktif, dan DB harus ikut dipangkas supaya session yang dibuka + lagi tidak menampilkan — atau mengirim — pesan assistant dengan + tool_calls yang hasilnya tidak ada.""" + if count <= 0: + return True + doc = self._table.get(doc_id=doc_id) + if doc is None: + return False + messages = doc["messages"] + del messages[len(messages) - min(count, len(messages)):] + now = datetime.now(timezone.utc).isoformat() + return self._table.update( + {"messages": messages, "updated_at": now}, + doc_ids=[doc_id], + ) + @staticmethod def _doc_to_session(doc) -> NeoSession: return NeoSession( diff --git a/services/telegram_client.py b/services/telegram_client.py index f325a89..11eaf2e 100644 --- a/services/telegram_client.py +++ b/services/telegram_client.py @@ -259,6 +259,11 @@ class TelegramClient: tool_reminder = None + # Turn baru: bersihkan state cancel. LLMClient dipakai bersama antar + # interface, jadi cancel yang tertinggal (mis. dari TUI) tidak boleh + # membuat turn ini langsung berhenti. + self._llm.begin_turn() + outcome = run_agent_loop( SessionStore(session.messages), self._llm, self._TOOLS, self._TOOL_HANDLERS, self._max_iterations, @@ -267,6 +272,15 @@ class TelegramClient: ) final_content, should_close = outcome.content, outcome.session_ended + if outcome.cancelled: + # Turn dihentikan (cancel adalah flag global pada LLMClient yang + # dipakai bersama antar interface). Sisa turn sudah dibuang engine + # dari konteks; jangan kirim pesan "max iterations" yang menyesatkan. + loggorilla.prcss('telegram', f'Turn cancelled for {chat_id}') + session.start_timer(86400 if chat_type == 'private' else 1800, + self._timeout_session, chat_id) + return + def _pop_ghost(sentinel: str): # Hapus pesan assistant sentinel (NO-REPLY / END-SESSION) yang # tidak dikirim agar konteks tidak tercemar balasan hantu. diff --git a/services/xmpp_client.py b/services/xmpp_client.py index f5b4b15..3f6cc78 100644 --- a/services/xmpp_client.py +++ b/services/xmpp_client.py @@ -624,6 +624,10 @@ class XMPPClient(ClientXMPP): tool_reminder = None + # Turn baru: bersihkan state cancel warisan antar interface/thread + # yang berbagi satu LLMClient. + self._llm.begin_turn() + outcome = run_agent_loop( SessionStore(session.messages), self._llm, self._TOOLS, self._TOOL_HANDLERS, self._max_iterations, @@ -632,6 +636,13 @@ class XMPPClient(ClientXMPP): ) final_content, should_close = outcome.content, outcome.session_ended + if outcome.cancelled: + # Turn dihentikan (flag cancel global pada LLMClient bersama). + # Sisa turn sudah dibuang engine; jangan kirim "max iterations". + print(f'[{_ts()}] DM: turn cancelled for {jid}', flush=True) + session.start_timer(86400, self._timeout_session, jid, 'chat') + return + if final_content is not None: if is_roleplay: # For DM: No selective response here. @@ -712,6 +723,9 @@ class XMPPClient(ClientXMPP): tool_reminder = None + # Turn baru: bersihkan state cancel warisan antar interface/thread. + self._llm.begin_turn() + outcome = run_agent_loop( SessionStore(session.messages), self._llm, self._TOOLS, self._TOOL_HANDLERS, self._max_iterations, @@ -720,6 +734,13 @@ class XMPPClient(ClientXMPP): ) final_content, should_close = outcome.content, outcome.session_ended + if outcome.cancelled: + # Turn dihentikan: sisa turn sudah dibuang engine dari konteks. + # Diam total (tanpa "max iterations") lalu lanjutkan timer sesi. + print(f'[{_ts()}] MUC: turn cancelled for {room}', flush=True) + session.start_timer(1800, self._timeout_session, room, 'groupchat') + return + def _pop_ghost(sentinel: str): # Hapus pesan assistant sentinel (NO-REPLY / END-SESSION) yang # tidak dikirim agar konteks tidak tercemar balasan hantu.