From fa4e2244cd65a061e50669f123edf95f7903ddda Mon Sep 17 00:00:00 2001 From: Alishahryar1 Date: Wed, 28 Jan 2026 22:07:05 -0800 Subject: [PATCH] fixed messages length requirement and replies not queueing to same session --- providers/claude_cli.py | 4 ++- providers/cli_session_manager.py | 20 ++++++++++---- server.py | 45 ++++++++++++++++++++++++++------ 3 files changed, 55 insertions(+), 14 deletions(-) diff --git a/providers/claude_cli.py b/providers/claude_cli.py index 3263ca07..e09e0ae6 100644 --- a/providers/claude_cli.py +++ b/providers/claude_cli.py @@ -158,7 +158,8 @@ class CLISession: env["PYTHONIOENCODING"] = "utf-8" # Build command based on whether resuming or starting new - if session_id: + # Important: only use REAL session IDs for --resume, not our internal 'pending_' IDs + if session_id and not session_id.startswith("pending_"): # Resume existing session cmd = [ "claude", @@ -237,6 +238,7 @@ class CLISession: ) return_code = await self.process.wait() + logger.info(f"Claude CLI process exited with code {return_code}") yield {"type": "exit", "code": return_code} finally: self._is_busy = False diff --git a/providers/cli_session_manager.py b/providers/cli_session_manager.py index dbd1ce58..bf6786af 100644 --- a/providers/cli_session_manager.py +++ b/providers/cli_session_manager.py @@ -73,10 +73,17 @@ class CLISessionManager: For new sessions, session_id is a temporary ID until CLI assigns real one. """ async with self._lock: - # Case 1: Resume existing session - if session_id and session_id in self._sessions: - logger.debug(f"Reusing existing session: {session_id}") - return self._sessions[session_id], session_id, False + # Case 1: Resume existing session (active or pending) + if session_id: + # Resolve temp_id to real_id if needed + lookup_id = self._temp_to_real.get(session_id, session_id) + + if lookup_id in self._sessions: + logger.debug(f"Reusing existing session: {lookup_id}") + return self._sessions[lookup_id], lookup_id, False + if lookup_id in self._pending_sessions: + logger.debug(f"Reusing pending session: {lookup_id}") + return self._pending_sessions[lookup_id], lookup_id, False # Case 2: Check if we're at capacity total_sessions = len(self._sessions) + len(self._pending_sessions) @@ -94,7 +101,10 @@ class CLISessionManager: ) # Case 3: Create new session - temp_id = f"pending_{uuid.uuid4().hex[:8]}" + # If session_id was provided (but not found), use it as temp_id + # Otherwise generate a new one + temp_id = session_id if session_id else f"pending_{uuid.uuid4().hex[:8]}" + new_session = CLISession( workspace_path=self.workspace, api_url=self.api_url, diff --git a/server.py b/server.py index 1e0f697c..d9dea196 100644 --- a/server.py +++ b/server.py @@ -364,6 +364,21 @@ def register_bot_handlers(client: "TelegramClient"): temp_session_id = None # Track temp ID for new sessions cli_session = None # The CLISession instance for this task + def safe_markdown_truncate(text, limit=3800): + """Truncate text carefully to avoid breaking markdown entities or blocks.""" + if len(text) <= limit: + return text + + # Show the end of the content as it's usually the most relevant + truncated = "..." + text[-(limit-5):] + + # Simple check for unclosed code blocks + # This is a heuristic but covers common CLI output issues + if truncated.count("```") % 2 != 0: + truncated += "\n```" + + return truncated + def build_unified_message(status=None): lines = [] if status: @@ -384,9 +399,7 @@ def register_bot_handlers(client: "TelegramClient"): lines.append(f"⚠️ {content}") result = "\n".join(lines) - if len(result) > 4000: - result = "..." + result[-3997:] - return result + return safe_markdown_truncate(result) async def update_bot_ui(status=None, force=False): nonlocal last_ui_update @@ -399,7 +412,7 @@ def register_bot_handlers(client: "TelegramClient"): await status_msg.edit(display, parse_mode="markdown") last_ui_update = now except Exception as e: - logger.debug(f"UI update failed: {e}") + logger.error(f"BOT: UI update failed: {e}") try: # Get or create CLI session from the manager @@ -429,7 +442,7 @@ def register_bot_handlers(client: "TelegramClient"): return # Process CLI events - async for event_data in cli_session.start_task(prompt, session_id=session_id_to_resume): + async for event_data in cli_session.start_task(prompt, session_id=captured_session_id): if not isinstance(event_data, dict): continue @@ -630,9 +643,25 @@ def register_bot_handlers(client: "TelegramClient"): processor=process_claude_task, ) else: - # NEW session - process directly in a new task (parallel!) - # Each new message gets its own CLI instance immediately - asyncio.create_task(process_claude_task(None, queued_msg)) + # NEW session - create a temporary ID based on the trigger message + temp_session_id = f"pending_{event.id}" + logger.info(f"BOT: Starting NEW session {temp_session_id}") + + # Pre-register in session store so replies to this NEW message or its status + # can be identified and enqueued immediately. + session_store.save_session( + session_id=temp_session_id, + chat_id=event.chat_id, + initial_msg_id=event.id + ) + session_store.update_last_message(temp_session_id, status_msg.id) + + # Process via queue to ensure we track busy state even for new sessions + await message_queue.enqueue( + session_id=temp_session_id, + message=queued_msg, + processor=process_claude_task, + ) FAST_PREFIX_DETECTION = os.getenv("FAST_PREFIX_DETECTION", "true").lower() == "true"