From 7b01a0c05fd6052bfdb20643502dd677d69e1508 Mon Sep 17 00:00:00 2001 From: Fede Luzzi Date: Mon, 18 May 2026 14:11:28 -0300 Subject: [PATCH] update pdoc --- docs/connpy/cli/terminal_ui.html | 266 ++++++++++++++++++++------- docs/connpy/grpc_layer/server.html | 91 +++++---- docs/connpy/grpc_layer/stubs.html | 185 +++++++------------ docs/connpy/index.html | 222 +++++++++++++++------- docs/connpy/services/ai_service.html | 166 ++++++++++++----- docs/connpy/services/index.html | 166 ++++++++++++----- docs/connpy/utils.html | 36 +++- 7 files changed, 729 insertions(+), 403 deletions(-) diff --git a/docs/connpy/cli/terminal_ui.html b/docs/connpy/cli/terminal_ui.html index cebfcc5..aff4dcb 100644 --- a/docs/connpy/cli/terminal_ui.html +++ b/docs/connpy/cli/terminal_ui.html @@ -90,9 +90,10 @@ el.replaceWith(d); async def run_session(self, raw_bytes: bytes, - cmd_byte_positions: List[tuple], node_info: dict, - on_ai_call: Callable): + on_ai_call: Callable, + cmd_byte_positions: List[tuple] = None, + blocks: List[tuple] = None): """ Runs the interactive Copilot session. on_ai_call: async function(active_buffer, question) -> result_dict @@ -102,9 +103,11 @@ el.replaceWith(d); try: # Prepare UI state buffer = log_cleaner(raw_bytes.decode(errors='replace')) - blocks = self.ai_service.build_context_blocks(raw_bytes, cmd_byte_positions, node_info) - last_line = buffer.split('\n')[-1].strip() if buffer.strip() else "(prompt)" - blocks.append((len(raw_bytes), last_line[:80])) + + # Use pre-calculated blocks if provided (remote mode), otherwise calculate locally (local mode) + if blocks is None: + last_line = buffer.split('\n')[-1].strip() if buffer.strip() else "(prompt)" + blocks = self.ai_service.build_context_blocks(raw_bytes, cmd_byte_positions, node_info, last_line=last_line) state = { 'context_cmd': 1, @@ -121,7 +124,7 @@ el.replaceWith(d); self.console.print("") # Salto de línea real self.console.print(Rule(title="[bold cyan] AI TERMINAL COPILOT [/bold cyan]", style="cyan")) self.console.print(Panel( - "[dim]Type your question. Enter to send, Escape/Ctrl+C to cancel.\n" + "[dim]Type your question. Enter to send, Escape/Ctrl+C to cancel. Type / for commands.\n" "Tab to change context mode. Ctrl+\u2191/\u2193 to adjust context. \u2191\u2193 for question history.[/dim]", border_style="cyan" )) @@ -161,9 +164,8 @@ el.replaceWith(d); if state['context_mode'] == self.mode_lines: return '\n'.join(buffer.split('\n')[-state['context_lines']:]) idx = max(0, state['total_cmds'] - state['context_cmd']) - start, preview = blocks[idx] - if state['context_mode'] == self.mode_single and idx + 1 < state['total_cmds']: - end = blocks[idx + 1][0] + start, end, preview = blocks[idx] + if state['context_mode'] == self.mode_single: active_raw = raw_bytes[start:end] else: active_raw = raw_bytes[start:] @@ -205,7 +207,40 @@ el.replaceWith(d); base_str = f'\u25b6 Ctrl+\u2191/\u2193 adjusts by 50 lines [Tab: {m_label}]' else: idx = max(0, state['total_cmds'] - state['context_cmd']) - desc = blocks[idx][1] + import re + + def clean_preview(text): + # Limpia saltos de línea y el prompt inicial (todo hasta #, > o $) para que quede solo el comando + original = text.strip().replace('\r', '').replace('\n', ' ') + cleaned = re.sub(r'^.*?[#>\$]\s*', '', original) + # Si limpiar el prompt nos deja con un string vacío (ej: era solo "iol#"), devolvemos el original + return cleaned if cleaned else original + + if state['context_mode'] == self.mode_range: + range_blocks = blocks[idx:] + # Si hay más de un bloque, el último es siempre el prompt vacío/actual. Lo omitimos visualmente. + if len(range_blocks) > 1: + range_blocks = range_blocks[:-1] + + # Limpiar y truncar comandos muy largos para que no rompan la UI + previews = [] + for b in range_blocks: + p = clean_preview(b[2]) + if p: + # Truncar comandos individuales largos + if len(p) > 25: p = p[:22] + "..." + previews.append(p) + + if not previews: + desc = clean_preview(blocks[idx][2]) + elif len(previews) <= 3: + desc = " + ".join(previews) + else: + desc = f"{previews[0]} + {previews[1]} + {previews[2]} ... (+{len(previews)-3})" + else: + # Modo SINGLE original + desc = clean_preview(blocks[idx][2]) + base_str = f'\u25b6 {desc} [Tab: {m_label}]' # Wrap base_str in a style to maintain consistency and avoid glitches @@ -332,39 +367,67 @@ el.replaceWith(d); # Use persona from overrides (one-shot) or from session state active_persona = merged_node_info.get('persona', self.session_state.get('persona', 'engineer')) persona_color = self._get_theme_color(active_persona, fallback="cyan") + persona_title = "Network Architect" if active_persona == "architect" else "Network Engineer" active_buffer = get_active_buffer() - live_text = "Thinking..." - panel = Panel(live_text, title=f"[bold {persona_color}]Copilot Guide[/bold {persona_color}]", border_style=persona_color) + live_text = "" + first_chunk = True + + import sys + from rich.rule import Rule + from rich.status import Status + from connpy.printer import IncrementalMarkdownParser + + md_parser = IncrementalMarkdownParser(console=self.console) + + status_spinner = Status( + f"[bold {persona_color}]{persona_title}:[/bold {persona_color}] [dim]Thinking...[/dim]", + console=self.console, + spinner="dots" + ) + status_spinner.start() def on_chunk(text): - nonlocal live_text - if live_text == "Thinking...": live_text = "" + nonlocal live_text, first_chunk + if first_chunk: + status_spinner.stop() + # Print header rule before first chunk arrives + self.console.print(Rule( + f"[bold {persona_color}]{persona_title}[/bold {persona_color}]", + style=persona_color + )) + first_chunk = False live_text += text + md_parser.feed(text) - with Live(panel, console=self.console, refresh_per_second=10) as live: - def update_live(t): - live.update(Panel(Markdown(t), title=f"[bold {persona_color}]Copilot Guide[/bold {persona_color}]", border_style=persona_color)) - - wrapped_chunk = lambda t: (on_chunk(t), update_live(live_text)) - - # Check for interruption during AI call - ai_task = asyncio.create_task(on_ai_call(active_buffer, clean_question, wrapped_chunk, merged_node_info)) - - try: - while not ai_task.done(): - await asyncio.sleep(0.05) - result = await ai_task - except asyncio.CancelledError: - return "cancel", None, None + # Check for interruption during AI call + ai_task = asyncio.create_task(on_ai_call(active_buffer, clean_question, on_chunk, merged_node_info)) + + try: + while not ai_task.done(): + await asyncio.sleep(0.05) + result = await ai_task + except asyncio.CancelledError: + status_spinner.stop() + return "cancel", None, None + + # Ensure spinner is stopped if no chunks arrived + if first_chunk: + status_spinner.stop() + # Close the streamed output with a Rule + if not first_chunk: + md_parser.flush() + self.console.print(Rule(style=persona_color)) + if not result or result.get("error"): - if result and result.get("error"): self.console.print(f"[red]Error: {result['error']}[/red]") + if first_chunk and result and result.get("error"): + self.console.print(f"[red]Error: {result['error']}[/red]") return "cancel", None, None - # 4. Handle result - if live_text == "Thinking..." and result.get("guide"): - self.console.print(Panel(Markdown(result["guide"]), title=f"[bold {persona_color}]Copilot Guide[/bold {persona_color}]", border_style=persona_color)) + # If no chunks were streamed but we have a guide, print it as a panel + if first_chunk and result and result.get("guide"): + self.console.print(Panel(Markdown(result["guide"]), title=f"[bold {persona_color}]{persona_title}[/bold {persona_color}]", border_style=persona_color)) commands = result.get("commands", []) if not commands: @@ -466,14 +529,13 @@ el.replaceWith(d); return "cancel", None, None finally: - state['cancelled'] = True - self.console.print("[dim]Returning to session...[/dim]") + state['cancelled'] = True

Methods

-async def run_session(self,
raw_bytes: bytes,
cmd_byte_positions: List[tuple],
node_info: dict,
on_ai_call: Callable) +async def run_session(self,
raw_bytes: bytes,
node_info: dict,
on_ai_call: Callable,
cmd_byte_positions: List[tuple] = None,
blocks: List[tuple] = None)
@@ -482,9 +544,10 @@ el.replaceWith(d); 
async def run_session(self, 
                       raw_bytes: bytes, 
-                      cmd_byte_positions: List[tuple], 
                       node_info: dict,
-                      on_ai_call: Callable):
+                      on_ai_call: Callable,
+                      cmd_byte_positions: List[tuple] = None, 
+                      blocks: List[tuple] = None):
     """
     Runs the interactive Copilot session.
     on_ai_call: async function(active_buffer, question) -> result_dict
@@ -494,9 +557,11 @@ el.replaceWith(d);
     try:
         # Prepare UI state
         buffer = log_cleaner(raw_bytes.decode(errors='replace'))
-        blocks = self.ai_service.build_context_blocks(raw_bytes, cmd_byte_positions, node_info)
-        last_line = buffer.split('\n')[-1].strip() if buffer.strip() else "(prompt)"
-        blocks.append((len(raw_bytes), last_line[:80]))
+        
+        # Use pre-calculated blocks if provided (remote mode), otherwise calculate locally (local mode)
+        if blocks is None:
+            last_line = buffer.split('\n')[-1].strip() if buffer.strip() else "(prompt)"
+            blocks = self.ai_service.build_context_blocks(raw_bytes, cmd_byte_positions, node_info, last_line=last_line)
         
         state = {
             'context_cmd': 1,
@@ -513,7 +578,7 @@ el.replaceWith(d);
         self.console.print("") # Salto de línea real
         self.console.print(Rule(title="[bold cyan] AI TERMINAL COPILOT [/bold cyan]", style="cyan"))
         self.console.print(Panel(
-            "[dim]Type your question. Enter to send, Escape/Ctrl+C to cancel.\n"
+            "[dim]Type your question. Enter to send, Escape/Ctrl+C to cancel. Type / for commands.\n"
             "Tab to change context mode. Ctrl+\u2191/\u2193 to adjust context. \u2191\u2193 for question history.[/dim]",
             border_style="cyan"
         ))
@@ -553,9 +618,8 @@ el.replaceWith(d);
             if state['context_mode'] == self.mode_lines:
                 return '\n'.join(buffer.split('\n')[-state['context_lines']:])
             idx = max(0, state['total_cmds'] - state['context_cmd'])
-            start, preview = blocks[idx]
-            if state['context_mode'] == self.mode_single and idx + 1 < state['total_cmds']:
-                end = blocks[idx + 1][0]
+            start, end, preview = blocks[idx]
+            if state['context_mode'] == self.mode_single:
                 active_raw = raw_bytes[start:end]
             else:
                 active_raw = raw_bytes[start:]
@@ -597,7 +661,40 @@ el.replaceWith(d);
                 base_str = f'\u25b6 Ctrl+\u2191/\u2193 adjusts by 50 lines  [Tab: {m_label}]'
             else:
                 idx = max(0, state['total_cmds'] - state['context_cmd'])
-                desc = blocks[idx][1]
+                import re
+                
+                def clean_preview(text):
+                    # Limpia saltos de línea y el prompt inicial (todo hasta #, > o $) para que quede solo el comando
+                    original = text.strip().replace('\r', '').replace('\n', ' ')
+                    cleaned = re.sub(r'^.*?[#>\$]\s*', '', original)
+                    # Si limpiar el prompt nos deja con un string vacío (ej: era solo "iol#"), devolvemos el original
+                    return cleaned if cleaned else original
+
+                if state['context_mode'] == self.mode_range:
+                    range_blocks = blocks[idx:]
+                    # Si hay más de un bloque, el último es siempre el prompt vacío/actual. Lo omitimos visualmente.
+                    if len(range_blocks) > 1:
+                        range_blocks = range_blocks[:-1]
+                        
+                    # Limpiar y truncar comandos muy largos para que no rompan la UI
+                    previews = []
+                    for b in range_blocks:
+                        p = clean_preview(b[2])
+                        if p:
+                            # Truncar comandos individuales largos
+                            if len(p) > 25: p = p[:22] + "..."
+                            previews.append(p)
+                    
+                    if not previews:
+                        desc = clean_preview(blocks[idx][2])
+                    elif len(previews) <= 3:
+                        desc = " + ".join(previews)
+                    else:
+                        desc = f"{previews[0]} + {previews[1]} + {previews[2]} ... (+{len(previews)-3})"
+                else:
+                    # Modo SINGLE original
+                    desc = clean_preview(blocks[idx][2])
+                    
                 base_str = f'\u25b6 {desc}  [Tab: {m_label}]'
             
             # Wrap base_str in a style to maintain consistency and avoid glitches
@@ -724,39 +821,67 @@ el.replaceWith(d);
             # Use persona from overrides (one-shot) or from session state
             active_persona = merged_node_info.get('persona', self.session_state.get('persona', 'engineer'))
             persona_color = self._get_theme_color(active_persona, fallback="cyan")
+            persona_title = "Network Architect" if active_persona == "architect" else "Network Engineer"
             
             active_buffer = get_active_buffer()
-            live_text = "Thinking..."
-            panel = Panel(live_text, title=f"[bold {persona_color}]Copilot Guide[/bold {persona_color}]", border_style=persona_color)
+            live_text = ""
+            first_chunk = True
+            
+            import sys
+            from rich.rule import Rule
+            from rich.status import Status
+            from connpy.printer import IncrementalMarkdownParser
+            
+            md_parser = IncrementalMarkdownParser(console=self.console)
+            
+            status_spinner = Status(
+                f"[bold {persona_color}]{persona_title}:[/bold {persona_color}] [dim]Thinking...[/dim]",
+                console=self.console,
+                spinner="dots"
+            )
+            status_spinner.start()
             
             def on_chunk(text):
-                nonlocal live_text
-                if live_text == "Thinking...": live_text = ""
+                nonlocal live_text, first_chunk
+                if first_chunk:
+                    status_spinner.stop()
+                    # Print header rule before first chunk arrives
+                    self.console.print(Rule(
+                        f"[bold {persona_color}]{persona_title}[/bold {persona_color}]",
+                        style=persona_color
+                    ))
+                    first_chunk = False
                 live_text += text
+                md_parser.feed(text)
             
-            with Live(panel, console=self.console, refresh_per_second=10) as live:
-                def update_live(t):
-                    live.update(Panel(Markdown(t), title=f"[bold {persona_color}]Copilot Guide[/bold {persona_color}]", border_style=persona_color))
-
-                wrapped_chunk = lambda t: (on_chunk(t), update_live(live_text))
-                
-                # Check for interruption during AI call
-                ai_task = asyncio.create_task(on_ai_call(active_buffer, clean_question, wrapped_chunk, merged_node_info))
-                
-                try:
-                    while not ai_task.done():
-                        await asyncio.sleep(0.05)
-                    result = await ai_task
-                except asyncio.CancelledError:
-                    return "cancel", None, None
+            # Check for interruption during AI call
+            ai_task = asyncio.create_task(on_ai_call(active_buffer, clean_question, on_chunk, merged_node_info))
+            
+            try:
+                while not ai_task.done():
+                    await asyncio.sleep(0.05)
+                result = await ai_task
+            except asyncio.CancelledError:
+                status_spinner.stop()
+                return "cancel", None, None
+            
+            # Ensure spinner is stopped if no chunks arrived
+            if first_chunk:
+                status_spinner.stop()
 
+            # Close the streamed output with a Rule
+            if not first_chunk:
+                md_parser.flush()
+                self.console.print(Rule(style=persona_color))
+            
             if not result or result.get("error"):
-                if result and result.get("error"): self.console.print(f"[red]Error: {result['error']}[/red]")
+                if first_chunk and result and result.get("error"):
+                    self.console.print(f"[red]Error: {result['error']}[/red]")
                 return "cancel", None, None
 
-            # 4. Handle result
-            if live_text == "Thinking..." and result.get("guide"):
-                self.console.print(Panel(Markdown(result["guide"]), title=f"[bold {persona_color}]Copilot Guide[/bold {persona_color}]", border_style=persona_color))
+            # If no chunks were streamed but we have a guide, print it as a panel
+            if first_chunk and result and result.get("guide"):
+                self.console.print(Panel(Markdown(result["guide"]), title=f"[bold {persona_color}]{persona_title}[/bold {persona_color}]", border_style=persona_color))
 
             commands = result.get("commands", [])
             if not commands:
@@ -858,8 +983,7 @@ el.replaceWith(d);
         return "cancel", None, None
 
     finally:
-        state['cancelled'] = True
-        self.console.print("[dim]Returning to session...[/dim]")
+ state['cancelled'] = True

Runs the interactive Copilot session. on_ai_call: async function(active_buffer, question) -> result_dict

diff --git a/docs/connpy/grpc_layer/server.html b/docs/connpy/grpc_layer/server.html index 2a4472a..3b3bf41 100644 --- a/docs/connpy/grpc_layer/server.html +++ b/docs/connpy/grpc_layer/server.html @@ -807,15 +807,34 @@ interceptor chooses to service this RPC, or None otherwise.

import json import asyncio import os + from ..services.ai_service import AIService + + service = AIService(self.service.config) if node_info is None: node_info = {} + + # Calculate real command blocks from history using the central service + raw_bytes = n.mylog.getvalue() if hasattr(n, 'mylog') else buffer + if not isinstance(raw_bytes, bytes): + raw_bytes = str(raw_bytes).encode() + + from connpy.utils import log_cleaner + last_line = log_cleaner(raw_bytes.decode(errors='replace')).split('\n')[-1].strip() + blocks = service.build_context_blocks(raw_bytes, n.cmd_byte_positions, node_info, last_line=last_line) + node_info["context_blocks"] = blocks node_info_json = json.dumps(node_info) # Convert buffer to string if it's bytes for the preview preview_str = buffer[-200:].decode(errors='replace') if isinstance(buffer, bytes) else str(buffer)[-200:] + # Generate a unique session ID for this copilot interaction to prevent race conditions + import uuid + copilot_session_id = str(uuid.uuid4()) + node_info["session_id"] = copilot_session_id + node_info_json = json.dumps(node_info) + # 1. Send prompt to client response_queue.put(connpy_pb2.InteractResponse( copilot_prompt=True, @@ -824,6 +843,13 @@ interceptor chooses to service this RPC, or None otherwise.

)) while True: + # 0. Drain the queue of any stale messages before starting a new interaction + while not remote_stream.copilot_queue.empty(): + try: + remote_stream.copilot_queue.get_nowait() + except: + break + # 2. Await the question from client via the copilot_queue import threading def preload_ai_deps(): @@ -836,8 +862,17 @@ interceptor chooses to service this RPC, or None otherwise.

try: req_data = await asyncio.wait_for(remote_stream.copilot_queue.get(), timeout=120) if not req_data: return - if "question" not in req_data or not req_data["question"] or req_data["question"] == "CANCEL" or req_data.get("action") == "cancel": - os.write(child_fd, b'\x15\r') + + # Validate session ID if provided by client (skip validation if not provided for CLI compatibility) + req_session_id = req_data.get("session_id") + if req_session_id and req_session_id != copilot_session_id: + continue # Ignore stale request from a previous session + + if "question" not in req_data or not req_data["question"] or req_data["question"] == "CANCEL" or req_data.get("action") in ("cancel", "web_cancel"): + if req_data.get("action") == "web_cancel": + os.write(child_fd, b'\x05') + else: + os.write(child_fd, b'\x15\r') return question = req_data["question"] @@ -864,9 +899,6 @@ interceptor chooses to service this RPC, or None otherwise.

return # 3. Call AI Service with streaming - from ..services.ai_service import AIService - service = AIService(self.service.config) - def chunk_callback(chunk_text): if chunk_text: response_queue.put(connpy_pb2.InteractResponse( @@ -887,8 +919,11 @@ interceptor chooses to service this RPC, or None otherwise.

if wait_action_task in done: req_data = wait_action_task.result() ai_task.cancel() - if req_data.get("action") == "cancel" or req_data.get("question") == "CANCEL": - os.write(child_fd, b'\x15\r') + if req_data.get("action") in ("cancel", "web_cancel") or req_data.get("question") == "CANCEL": + if req_data.get("action") == "web_cancel": + os.write(child_fd, b'\x05') + else: + os.write(child_fd, b'\x15\r') return continue # Loop back instead of returning to keep session alive else: @@ -912,45 +947,27 @@ interceptor chooses to service this RPC, or None otherwise.

if action == "continue": continue # Loop back for next question - if action == "cancel": - os.write(child_fd, b'\x15\r') + if action in ("cancel", "web_cancel"): + if action == "web_cancel": + os.write(child_fd, b'\x05') + else: + os.write(child_fd, b'\x15\r') return except asyncio.TimeoutError: os.write(child_fd, b'\x15\r') return + def on_inject(cmd): + response_queue.put(connpy_pb2.InteractResponse(copilot_injected_command=cmd)) + if action == "send_all": commands = result.get("commands", []) - os.write(child_fd, b'\x15') # Ctrl+U to clear line - await asyncio.sleep(0.1) - - # Prepend screen length command to avoid pagination - if "screen_length_command" in n.tags: - os.write(child_fd, (n.tags["screen_length_command"] + "\n").encode()) - response_queue.put(connpy_pb2.InteractResponse(copilot_injected_command=n.tags["screen_length_command"])) - await asyncio.sleep(0.8) - - for cmd in commands: - os.write(child_fd, (cmd + "\n").encode()) - response_queue.put(connpy_pb2.InteractResponse(copilot_injected_command=cmd)) - await asyncio.sleep(0.8) + await n.inject_commands(commands, child_fd, on_inject=on_inject) return elif action.startswith("custom:"): - custom_cmds = action[7:] - os.write(child_fd, b'\x15') - await asyncio.sleep(0.1) - - # Prepend screen length command to avoid pagination - if "screen_length_command" in n.tags: - os.write(child_fd, (n.tags["screen_length_command"] + "\n").encode()) - response_queue.put(connpy_pb2.InteractResponse(copilot_injected_command=n.tags["screen_length_command"])) - await asyncio.sleep(0.8) - - for cmd in custom_cmds.split('\n'): - if cmd.strip(): - os.write(child_fd, (cmd.strip() + "\n").encode()) - response_queue.put(connpy_pb2.InteractResponse(copilot_injected_command=cmd.strip())) - await asyncio.sleep(0.8) + custom_cmds_raw = action[7:] + custom_cmds = [cmd.strip() for cmd in custom_cmds_raw.split('\n') if cmd.strip()] + await n.inject_commands(custom_cmds, child_fd, on_inject=on_inject) return else: os.write(child_fd, b'\x15\r') diff --git a/docs/connpy/grpc_layer/stubs.html b/docs/connpy/grpc_layer/stubs.html index 21edc1b..d133dcd 100644 --- a/docs/connpy/grpc_layer/stubs.html +++ b/docs/connpy/grpc_layer/stubs.html @@ -104,7 +104,6 @@ el.replaceWith(d); import queue from rich.prompt import Prompt from rich.text import Text - from rich.live import Live from rich.panel import Panel from rich.markdown import Markdown @@ -135,7 +134,7 @@ el.replaceWith(d); responses = self.stub.ask(request_generator()) full_content = "" - live_display = None + header_printed = False final_result = {"response": "", "chat_history": []} # Background thread to pull responses from gRPC into a local queue @@ -200,69 +199,53 @@ el.replaceWith(d); if response.debug_message: if debug: - if live_display: - try: live_display.stop() - except: pass if status: try: status.stop() except: pass printer.console.print(Text.from_ansi(response.debug_message)) - if live_display: - try: live_display.start() - except: pass - elif status: + if status: try: status.start() except: pass continue if response.important_message: - if live_display: - try: live_display.stop() - except: pass if status: try: status.stop() except: pass printer.console.print(Text.from_ansi(response.important_message)) - if live_display: - try: live_display.start() - except: pass - elif status: + if status: try: status.start() except: pass continue if not response.is_final: if response.text_chunk: - full_content += response.text_chunk - - if not live_display: + if not header_printed: if status: try: status.stop() except: pass from rich.console import Console as RichConsole - from ..printer import connpy_theme, get_original_stdout + from rich.rule import Rule + from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout()) - # We default to Engineer title during stream, final result will correct it if needed - live_display = Live( - Panel(Markdown(full_content), title="[bold engineer]Network Engineer[/bold engineer]", border_style="engineer", expand=False), - console=stable_console, - refresh_per_second=8, - transient=False - ) - live_display.start() - else: - live_display.update( - Panel(Markdown(full_content), title="[bold engineer]Network Engineer[/bold engineer]", border_style="engineer", expand=False) - ) + # Print header on first chunk + stable_console.print(Rule("[bold engineer]Network Engineer[/bold engineer]", style="engineer")) + header_printed = True + + # Initialize parser + md_parser = IncrementalMarkdownParser(console=stable_console) + + full_content += response.text_chunk + md_parser.feed(response.text_chunk) continue if response.is_final: - if live_display: - try: live_display.stop() - except: pass - # Final stop for status to ensure it disappears before the panel + if header_printed: + from rich.rule import Rule + md_parser.flush() + if status: try: status.stop() except: pass @@ -273,13 +256,14 @@ el.replaceWith(d); role_label = "Network Architect" if responder == "architect" else "Network Engineer" title = f"[bold {alias}]{role_label}[/bold {alias}]" - content_to_print = full_content or final_result.get("response", "") - if content_to_print: - if live_display: - # Re-render the final frame with correct title/colors - live_display.update(Panel(Markdown(content_to_print), title=title, border_style=alias, expand=False)) - else: - printer.console.print(Panel(Markdown(content_to_print), title=title, border_style=alias, expand=False)) + if header_printed: + from rich.console import Console as RichConsole + from ..printer import connpy_theme, get_original_stdout + stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout()) + stable_console.print(Rule(style=alias)) + elif not full_content and final_result.get("response"): + # If nothing streamed but we have response (e.g. error or direct guide) + printer.console.print(Panel(Markdown(final_result["response"]), title=title, border_style=alias, expand=False)) break except Exception as e: # Check if it was a gRPC error that we should let handle_errors catch @@ -342,7 +326,6 @@ def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debu import queue from rich.prompt import Prompt from rich.text import Text - from rich.live import Live from rich.panel import Panel from rich.markdown import Markdown @@ -373,7 +356,7 @@ def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debu responses = self.stub.ask(request_generator()) full_content = "" - live_display = None + header_printed = False final_result = {"response": "", "chat_history": []} # Background thread to pull responses from gRPC into a local queue @@ -438,69 +421,53 @@ def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debu if response.debug_message: if debug: - if live_display: - try: live_display.stop() - except: pass if status: try: status.stop() except: pass printer.console.print(Text.from_ansi(response.debug_message)) - if live_display: - try: live_display.start() - except: pass - elif status: + if status: try: status.start() except: pass continue if response.important_message: - if live_display: - try: live_display.stop() - except: pass if status: try: status.stop() except: pass printer.console.print(Text.from_ansi(response.important_message)) - if live_display: - try: live_display.start() - except: pass - elif status: + if status: try: status.start() except: pass continue if not response.is_final: if response.text_chunk: - full_content += response.text_chunk - - if not live_display: + if not header_printed: if status: try: status.stop() except: pass from rich.console import Console as RichConsole - from ..printer import connpy_theme, get_original_stdout + from rich.rule import Rule + from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout()) - # We default to Engineer title during stream, final result will correct it if needed - live_display = Live( - Panel(Markdown(full_content), title="[bold engineer]Network Engineer[/bold engineer]", border_style="engineer", expand=False), - console=stable_console, - refresh_per_second=8, - transient=False - ) - live_display.start() - else: - live_display.update( - Panel(Markdown(full_content), title="[bold engineer]Network Engineer[/bold engineer]", border_style="engineer", expand=False) - ) + # Print header on first chunk + stable_console.print(Rule("[bold engineer]Network Engineer[/bold engineer]", style="engineer")) + header_printed = True + + # Initialize parser + md_parser = IncrementalMarkdownParser(console=stable_console) + + full_content += response.text_chunk + md_parser.feed(response.text_chunk) continue if response.is_final: - if live_display: - try: live_display.stop() - except: pass - # Final stop for status to ensure it disappears before the panel + if header_printed: + from rich.rule import Rule + md_parser.flush() + if status: try: status.stop() except: pass @@ -511,13 +478,14 @@ def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debu role_label = "Network Architect" if responder == "architect" else "Network Engineer" title = f"[bold {alias}]{role_label}[/bold {alias}]" - content_to_print = full_content or final_result.get("response", "") - if content_to_print: - if live_display: - # Re-render the final frame with correct title/colors - live_display.update(Panel(Markdown(content_to_print), title=title, border_style=alias, expand=False)) - else: - printer.console.print(Panel(Markdown(content_to_print), title=title, border_style=alias, expand=False)) + if header_printed: + from rich.console import Console as RichConsole + from ..printer import connpy_theme, get_original_stdout + stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout()) + stable_console.print(Rule(style=alias)) + elif not full_content and final_result.get("response"): + # If nothing streamed but we have response (e.g. error or direct guide) + printer.console.print(Panel(Markdown(final_result["response"]), title=title, border_style=alias, expand=False)) break except Exception as e: # Check if it was a gRPC error that we should let handle_errors catch @@ -1024,7 +992,7 @@ def set_reserved_names(self, names): self.remote_host = remote_host self.config = config - def _handle_remote_copilot(self, res, request_queue, response_queue, client_buffer_bytes, cmd_byte_positions, pause_generator, resume_generator, old_tty): + def _handle_remote_copilot(self, res, request_queue, response_queue, client_buffer_bytes, pause_generator, resume_generator, old_tty): import json, asyncio, termios, sys, tty, queue from ..core import copilot_terminal_mode from . import connpy_pb2 @@ -1032,6 +1000,10 @@ def set_reserved_names(self, names): pause_generator() termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty) + + node_info = json.loads(res.copilot_node_info_json) if res.copilot_node_info_json else {} + blocks = node_info.get("context_blocks", []) + interface = CopilotInterface( self.config, history=getattr(self, 'copilot_history', None), @@ -1039,8 +1011,6 @@ def set_reserved_names(self, names): ) self.copilot_history = interface.history self.copilot_state = interface.session_state - - node_info = json.loads(res.copilot_node_info_json) if res.copilot_node_info_json else {} async def on_ai_call_remote(active_buffer, question, chunk_callback, merged_node_info): # Send request to server @@ -1066,9 +1036,9 @@ def set_reserved_names(self, names): while True: action, commands, custom_cmd = await interface.run_session( raw_bytes=bytes(client_buffer_bytes), - cmd_byte_positions=cmd_byte_positions, node_info=node_info, - on_ai_call=on_ai_call_remote + on_ai_call=on_ai_call_remote, + blocks=blocks ) if action == "continue": @@ -1081,6 +1051,7 @@ def set_reserved_names(self, names): with copilot_terminal_mode(): action, commands, custom_cmd = asyncio.run(run_remote_copilot()) + print("\033[2m Returning to session...\033[0m", flush=True) # Prepare final action for server action_sent = "cancel" if action == "send_all" and commands: @@ -1105,7 +1076,6 @@ def set_reserved_names(self, names): request_queue = queue.Queue() client_buffer_bytes = bytearray() - cmd_byte_positions = [(0, None)] pause_stdin = [False] wake_r, wake_w = os.pipe() @@ -1152,8 +1122,6 @@ def set_reserved_names(self, names): data = os.read(sys.stdin.fileno(), 1024) if not data: break - if b'\r' in data or b'\n' in data: - cmd_byte_positions.append((len(client_buffer_bytes), None)) yield connpy_pb2.InteractRequest(stdin_data=data) except OSError: break @@ -1227,14 +1195,11 @@ def set_reserved_names(self, names): if res.copilot_prompt: self._handle_remote_copilot( res, request_queue, response_queue, - client_buffer_bytes, cmd_byte_positions, + client_buffer_bytes, pause_generator, resume_generator, old_tty ) continue - if res.copilot_injected_command: - cmd_byte_positions.append((len(client_buffer_bytes), res.copilot_injected_command)) - if res.stdout_data: os.write(sys.stdout.fileno(), res.stdout_data) client_buffer_bytes.extend(res.stdout_data) @@ -1256,7 +1221,6 @@ def set_reserved_names(self, names): params_json = json.dumps(connection_params) request_queue = queue.Queue() client_buffer_bytes = bytearray() - cmd_byte_positions = [(0, None)] pause_stdin = [False] wake_r, wake_w = os.pipe() @@ -1304,8 +1268,6 @@ def set_reserved_names(self, names): data = os.read(sys.stdin.fileno(), 1024) if not data: break - if b'\r' in data or b'\n' in data: - cmd_byte_positions.append((len(client_buffer_bytes), None)) yield connpy_pb2.InteractRequest(stdin_data=data) except OSError: break @@ -1378,14 +1340,11 @@ def set_reserved_names(self, names): if res.copilot_prompt: self._handle_remote_copilot( res, request_queue, response_queue, - client_buffer_bytes, cmd_byte_positions, + client_buffer_bytes, pause_generator, resume_generator, old_tty ) continue - if res.copilot_injected_command: - cmd_byte_positions.append((len(client_buffer_bytes), res.copilot_injected_command)) - if res.stdout_data: os.write(sys.stdout.fileno(), res.stdout_data) client_buffer_bytes.extend(res.stdout_data) @@ -1553,7 +1512,6 @@ def connect_dynamic(self, connection_params, debug=False): params_json = json.dumps(connection_params) request_queue = queue.Queue() client_buffer_bytes = bytearray() - cmd_byte_positions = [(0, None)] pause_stdin = [False] wake_r, wake_w = os.pipe() @@ -1601,8 +1559,6 @@ def connect_dynamic(self, connection_params, debug=False): data = os.read(sys.stdin.fileno(), 1024) if not data: break - if b'\r' in data or b'\n' in data: - cmd_byte_positions.append((len(client_buffer_bytes), None)) yield connpy_pb2.InteractRequest(stdin_data=data) except OSError: break @@ -1675,14 +1631,11 @@ def connect_dynamic(self, connection_params, debug=False): if res.copilot_prompt: self._handle_remote_copilot( res, request_queue, response_queue, - client_buffer_bytes, cmd_byte_positions, + client_buffer_bytes, pause_generator, resume_generator, old_tty ) continue - if res.copilot_injected_command: - cmd_byte_positions.append((len(client_buffer_bytes), res.copilot_injected_command)) - if res.stdout_data: os.write(sys.stdout.fileno(), res.stdout_data) client_buffer_bytes.extend(res.stdout_data) @@ -1713,7 +1666,6 @@ def connect_node(self, unique_id, sftp=False, debug=False, logger=None): request_queue = queue.Queue() client_buffer_bytes = bytearray() - cmd_byte_positions = [(0, None)] pause_stdin = [False] wake_r, wake_w = os.pipe() @@ -1760,8 +1712,6 @@ def connect_node(self, unique_id, sftp=False, debug=False, logger=None): data = os.read(sys.stdin.fileno(), 1024) if not data: break - if b'\r' in data or b'\n' in data: - cmd_byte_positions.append((len(client_buffer_bytes), None)) yield connpy_pb2.InteractRequest(stdin_data=data) except OSError: break @@ -1835,14 +1785,11 @@ def connect_node(self, unique_id, sftp=False, debug=False, logger=None): if res.copilot_prompt: self._handle_remote_copilot( res, request_queue, response_queue, - client_buffer_bytes, cmd_byte_positions, + client_buffer_bytes, pause_generator, resume_generator, old_tty ) continue - if res.copilot_injected_command: - cmd_byte_positions.append((len(client_buffer_bytes), res.copilot_injected_command)) - if res.stdout_data: os.write(sys.stdout.fileno(), res.stdout_data) client_buffer_bytes.extend(res.stdout_data) diff --git a/docs/connpy/index.html b/docs/connpy/index.html index 1585ac4..5b68952 100644 --- a/docs/connpy/index.html +++ b/docs/connpy/index.html @@ -737,6 +737,7 @@ class ai: - COMPLETE MISSIONS: Execute ALL steps of a mission before reporting back. - DIAGRAM: Use ASCII art or Unicode box-drawing characters directly in your responses to visualize topologies or paths when helpful. - EVIDENCE: Include 'Key Snippets' from tool outputs. Be token-efficient. + - LANGUAGE: You MUST respond in the same language used by the user in their question or instruction. - NO WANDERING: Do not speculate. If stuck, report attempts. - SAFETY: When you use 'run_commands' with configuration commands, the system automatically prompts the user for confirmation. Just execute - don't ask permission first. {architect_instructions} @@ -754,6 +755,7 @@ class ai: - ENGINEER CAPABILITIES: Your Engineer can: * Filter nodes (list_nodes), Run CLI commands (run_commands), Get metadata (get_node_info). - ANALYSIS: Review technical findings to identify patterns or design failures. + - LANGUAGE: You MUST respond in the same language used by the user in their question or instruction. - MEMORY: Update long-term facts ONLY when the user explicitly requests it. CRITICAL - EFFICIENT DELEGATION: @@ -829,7 +831,6 @@ class ai: - response: reconstructed ModelResponse (same as non-streaming) - streamed: True if text was rendered to console during streaming """ - from rich.live import Live stream_resp = completion(model=model, messages=messages, tools=tools, api_key=api_key, stream=True, **kwargs) @@ -837,7 +838,7 @@ class ai: full_content = "" is_streaming_text = False has_tool_calls = False - live_display = None + header_printed = False # Determine styling based on current brain role_label = "Network Architect" if "architect" in label.lower() else "Network Engineer" @@ -866,7 +867,6 @@ class ai: if not chunk_callback: if not is_streaming_text: - # Stop spinner definitively if status: try: status.stop() @@ -875,35 +875,28 @@ class ai: # Create a stable, direct Console to bypass _ConsoleProxy recreation bugs from rich.console import Console as RichConsole - from .printer import connpy_theme, get_original_stdout + from rich.rule import Rule + from .printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout()) - live_display = Live( - Panel(Markdown(full_content), title=title, border_style=border, expand=False), - console=stable_console, - refresh_per_second=8, - transient=False - ) - live_display.start() + stable_console.print(Rule(f"[bold {border}]{title}[/bold {border}]", style=border)) + header_printed = True + md_parser = IncrementalMarkdownParser(console=stable_console) is_streaming_text = True - else: - live_display.update( - Panel(Markdown(full_content), title=title, border_style=border, expand=False) - ) + + md_parser.feed(delta.content) except Exception as e: if not chunks: raise finally: - if live_display: - # Render final state with complete content + if header_printed: try: - live_display.update( - Panel(Markdown(full_content), title=title, border_style=border, expand=False) - ) - except Exception: - pass - try: - live_display.stop() + md_parser.flush() + from rich.console import Console as RichConsole + from rich.rule import Rule + from .printer import connpy_theme, get_original_stdout + stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout()) + stable_console.print(Rule(style=border)) except Exception: pass @@ -1852,11 +1845,13 @@ class ai: if persona == "architect": system_prompt = f"""Role: NETWORK ARCHITECT. You act as a senior strategic advisor during a live SSH session. Rules: -1. Answer the user's question directly based on the Terminal Context. -2. Focus on the "why" and "how". Analyze topologies, design patterns, and validate configurations. -3. Do NOT provide commands to execute unless specifically requested. Instead, explain the consequences and best practices. -4. Keep your guide concise and authoritative. -5. You MUST output your response in the following strict format: +1. MANDATORY: You MUST respond in the same language used by the user in their question. +2. Answer the user's question directly and EXCLUSIVELY based on the Terminal Context. +3. NO HALLUCINATIONS. The Terminal Context is a live buffer. If it contains only a shell prompt (like 'iol#' or 'admin@vrouter>') and no command output, it means YOU DON'T HAVE DATA. In this case, YOU MUST NOT invent any information. +4. Focus on the "why" and "how". Analyze topologies, design patterns, and validate configurations. +5. Do NOT provide commands to execute unless specifically requested. Instead, explain the consequences and best practices. +6. Keep your guide concise and authoritative. +7. You MUST output your response in the following strict format: <guide> Your brief tactical guide in markdown. </guide> @@ -1865,7 +1860,7 @@ Your brief tactical guide in markdown. <risk> low </risk> -6. Risk level is usually "low" for read-only/no commands. +8. Risk level is usually "low" for read-only/no commands. Terminal Context: {terminal_buffer} @@ -1875,11 +1870,13 @@ Node: {node_name}""" else: system_prompt = f"""Role: TERMINAL COPILOT. You assist a network engineer during a live SSH session. Rules: -1. Answer the user's question directly based on the Terminal Context. -2. If the user asks you to analyze, parse, or extract data from the Terminal Context, DO IT directly in the <guide> section (you can use markdown tables or lists). Do NOT just give them a command to do it themselves. -3. If the user wants to execute an action, provide the required CLI commands inside a <commands> block, one command per line. If no commands are needed, leave it empty or omit the block. -4. ULTRA-CONCISE. Keep your guide to the point. -5. You MUST output your response in the following strict format: +1. MANDATORY: You MUST respond in the same language used by the user in their question. +2. EXTREMELY IMPORTANT: Answer EXCLUSIVELY based on the provided Terminal Context. +3. NO HALLUCINATIONS. The Terminal Context is a live buffer. If it contains only a shell prompt (like 'iol#' or 'admin@vrouter>') and no command output, it means YOU DON'T HAVE DATA. In this case, YOU MUST NOT invent any information. Instead, explicitly state that you don't see the data and offer the correct CLI commands to retrieve it. +4. If the user asks you to analyze, parse, or extract data from the Terminal Context, DO IT directly in the <guide> section (you can use markdown tables or lists). Do NOT just give them a command to do it themselves. +5. If the user wants to execute an action, provide the required CLI commands inside a <commands> block, one command per line. If no commands are needed, leave it empty or omit the block. +6. ULTRA-CONCISE. Keep your guide to the point. +7. You MUST output your response in the following strict format: <guide> Your brief tactical guide in markdown. 3-4 sentences max. </guide> @@ -1890,7 +1887,7 @@ command 2 <risk> low, high, or destructive </risk> -6. Risk level: "low" for read-only/no commands, "high" for config changes, "destructive" for potentially dangerous ops. +8. Risk level: "low" for read-only/no commands, "high" for config changes, "destructive" for potentially dangerous ops. Terminal Context: {terminal_buffer} @@ -1932,6 +1929,7 @@ Node: {node_name}""" try: while iteration < max_iterations: iteration += 1 + response = await acompletion( model=current_model, messages=messages, @@ -2177,11 +2175,13 @@ def engineer_system_prompt(self): if persona == "architect": system_prompt = f"""Role: NETWORK ARCHITECT. You act as a senior strategic advisor during a live SSH session. Rules: -1. Answer the user's question directly based on the Terminal Context. -2. Focus on the "why" and "how". Analyze topologies, design patterns, and validate configurations. -3. Do NOT provide commands to execute unless specifically requested. Instead, explain the consequences and best practices. -4. Keep your guide concise and authoritative. -5. You MUST output your response in the following strict format: +1. MANDATORY: You MUST respond in the same language used by the user in their question. +2. Answer the user's question directly and EXCLUSIVELY based on the Terminal Context. +3. NO HALLUCINATIONS. The Terminal Context is a live buffer. If it contains only a shell prompt (like 'iol#' or 'admin@vrouter>') and no command output, it means YOU DON'T HAVE DATA. In this case, YOU MUST NOT invent any information. +4. Focus on the "why" and "how". Analyze topologies, design patterns, and validate configurations. +5. Do NOT provide commands to execute unless specifically requested. Instead, explain the consequences and best practices. +6. Keep your guide concise and authoritative. +7. You MUST output your response in the following strict format: <guide> Your brief tactical guide in markdown. </guide> @@ -2190,7 +2190,7 @@ Your brief tactical guide in markdown. <risk> low </risk> -6. Risk level is usually "low" for read-only/no commands. +8. Risk level is usually "low" for read-only/no commands. Terminal Context: {terminal_buffer} @@ -2200,11 +2200,13 @@ Node: {node_name}""" else: system_prompt = f"""Role: TERMINAL COPILOT. You assist a network engineer during a live SSH session. Rules: -1. Answer the user's question directly based on the Terminal Context. -2. If the user asks you to analyze, parse, or extract data from the Terminal Context, DO IT directly in the <guide> section (you can use markdown tables or lists). Do NOT just give them a command to do it themselves. -3. If the user wants to execute an action, provide the required CLI commands inside a <commands> block, one command per line. If no commands are needed, leave it empty or omit the block. -4. ULTRA-CONCISE. Keep your guide to the point. -5. You MUST output your response in the following strict format: +1. MANDATORY: You MUST respond in the same language used by the user in their question. +2. EXTREMELY IMPORTANT: Answer EXCLUSIVELY based on the provided Terminal Context. +3. NO HALLUCINATIONS. The Terminal Context is a live buffer. If it contains only a shell prompt (like 'iol#' or 'admin@vrouter>') and no command output, it means YOU DON'T HAVE DATA. In this case, YOU MUST NOT invent any information. Instead, explicitly state that you don't see the data and offer the correct CLI commands to retrieve it. +4. If the user asks you to analyze, parse, or extract data from the Terminal Context, DO IT directly in the <guide> section (you can use markdown tables or lists). Do NOT just give them a command to do it themselves. +5. If the user wants to execute an action, provide the required CLI commands inside a <commands> block, one command per line. If no commands are needed, leave it empty or omit the block. +6. ULTRA-CONCISE. Keep your guide to the point. +7. You MUST output your response in the following strict format: <guide> Your brief tactical guide in markdown. 3-4 sentences max. </guide> @@ -2215,7 +2217,7 @@ command 2 <risk> low, high, or destructive </risk> -6. Risk level: "low" for read-only/no commands, "high" for config changes, "destructive" for potentially dangerous ops. +8. Risk level: "low" for read-only/no commands, "high" for config changes, "destructive" for potentially dangerous ops. Terminal Context: {terminal_buffer} @@ -2257,6 +2259,7 @@ Node: {node_name}""" try: while iteration < max_iterations: iteration += 1 + response = await acompletion( model=current_model, messages=messages, @@ -4119,6 +4122,7 @@ class node: self.output = "" self.status = 1 self.result = {} + self.cmd_byte_positions = [(0, None)] @MethodHook def _passtx(self, passwords, *, keyfile=None): @@ -4293,9 +4297,9 @@ class node: loop = asyncio.get_running_loop() child_reader_queue = asyncio.Queue() - # Track command byte positions for copilot context navigation + # Reset and track command byte positions for copilot context navigation # Each entry is (byte_position, command_text_or_None) - cmd_byte_positions = [(0, None)] + self.cmd_byte_positions = [(self.mylog.tell() if hasattr(self, 'mylog') else 0, None)] def _child_read_ready(): try: @@ -4336,7 +4340,7 @@ class node: node_info["prompt"] = to_str(self.tags.get("prompt", r'>$|#$|\$$|>.$|#.$|\$.$')) # Invoke copilot (async callback handles UI) - await copilot_handler(self.mylog.getvalue(), node_info, local_stream, child_fd, cmd_byte_positions) + await copilot_handler(self.mylog.getvalue(), node_info, local_stream, child_fd, self.cmd_byte_positions) continue # Remove any stray \x00 bytes and forward normally @@ -4344,10 +4348,9 @@ class node: if clean_data: # Track command boundaries when user hits Enter if hasattr(self, 'mylog') and (b'\r' in clean_data or b'\n' in clean_data): - cmd_byte_positions.append((self.mylog.tell(), None)) - - try: - os.write(child_fd, clean_data) + self.cmd_byte_positions.append((self.mylog.tell(), None)) + + try: os.write(child_fd, clean_data) except OSError: break self.lastinput = time() @@ -4469,6 +4472,45 @@ class node: finally: local_stream.teardown() + @MethodHook + async def inject_commands(self, commands, child_fd, on_inject=None): + """ + Inject a list of commands into the node's PTY. + Handles screen_length_command, history tracking and delays. + """ + if not commands: + return + + # 0. Clear line + os.write(child_fd, b'\x15') + await asyncio.sleep(0.1) + + # 1. Prepare list (prepend screen_length if exists) + slc = self.tags.get("screen_length_command") if hasattr(self, 'tags') and isinstance(self.tags, dict) else None + + to_send = list(commands) + if slc and slc not in to_send: # avoid duplicates if already there + to_send.insert(0, slc) + + # 2. Inject one by one + for cmd in to_send: + # Register in node's official history (SKIP if it's the administrative screen length command) + if cmd != slc and hasattr(self, 'cmd_byte_positions') and self.cmd_byte_positions is not None: + log_pos = self.mylog.tell() if hasattr(self, 'mylog') else 0 + self.cmd_byte_positions.append((log_pos, cmd)) + + # Write physically to PTY + os.write(child_fd, (cmd + "\n").encode()) + + # Notify (e.g., for gRPC or logs) - SKIP for administrative SLC + if on_inject and cmd != slc: + if asyncio.iscoroutinefunction(on_inject): + await on_inject(cmd) + else: + on_inject(cmd) + + # Delay to avoid overwhelming the router + await asyncio.sleep(0.8) @MethodHook def interact(self, debug=False, logger=None): @@ -4550,7 +4592,7 @@ class node: while True: action, commands, custom_cmd = await interface.run_session( raw_bytes=raw_bytes, - cmd_byte_positions=cmd_byte_positions, + cmd_byte_positions=self.cmd_byte_positions, node_info=node_info, on_ai_call=on_ai_call ) @@ -4558,6 +4600,7 @@ class node: continue break finally: + print("\033[2m Returning to session...\033[0m", flush=True) # Reiniciar el lector de la terminal para volver al modo interactivo SSH/Telnet if hasattr(stream, 'start_reading'): stream.start_reading() @@ -4566,20 +4609,7 @@ class node: if action in ("send_all", "custom"): cmds_to_send = commands if action == "send_all" else custom_cmd - - if cmds_to_send: - os.write(child_fd, b'\x15') # Ctrl+U - await asyncio.sleep(0.1) - - # Prepend screen length command to avoid pagination - if "screen_length_command" in self.tags: - cmds_to_send.insert(0, self.tags["screen_length_command"]) - - for cmd in cmds_to_send: - if cmd_byte_positions is not None: - cmd_byte_positions.append((self.mylog.tell(), cmd)) - os.write(child_fd, (cmd + "\n").encode()) - await asyncio.sleep(0.8) + await self.inject_commands(cmds_to_send, child_fd) else: os.write(child_fd, b'\x15\r') except Exception as e: @@ -5073,6 +5103,57 @@ class node:

Methods

+
+async def inject_commands(self, commands, child_fd, on_inject=None) +
+
+
+ +Expand source code + +
@MethodHook
+async def inject_commands(self, commands, child_fd, on_inject=None):
+    """
+    Inject a list of commands into the node's PTY.
+    Handles screen_length_command, history tracking and delays.
+    """
+    if not commands:
+        return
+
+    # 0. Clear line
+    os.write(child_fd, b'\x15')
+    await asyncio.sleep(0.1)
+
+    # 1. Prepare list (prepend screen_length if exists)
+    slc = self.tags.get("screen_length_command") if hasattr(self, 'tags') and isinstance(self.tags, dict) else None
+    
+    to_send = list(commands)
+    if slc and slc not in to_send: # avoid duplicates if already there
+         to_send.insert(0, slc)
+
+    # 2. Inject one by one
+    for cmd in to_send:
+        # Register in node's official history (SKIP if it's the administrative screen length command)
+        if cmd != slc and hasattr(self, 'cmd_byte_positions') and self.cmd_byte_positions is not None:
+            log_pos = self.mylog.tell() if hasattr(self, 'mylog') else 0
+            self.cmd_byte_positions.append((log_pos, cmd))
+        
+        # Write physically to PTY
+        os.write(child_fd, (cmd + "\n").encode())
+        
+        # Notify (e.g., for gRPC or logs) - SKIP for administrative SLC
+        if on_inject and cmd != slc:
+            if asyncio.iscoroutinefunction(on_inject):
+                await on_inject(cmd)
+            else:
+                on_inject(cmd)
+        
+        # Delay to avoid overwhelming the router
+        await asyncio.sleep(0.8)
+
+

Inject a list of commands into the node's PTY. +Handles screen_length_command, history tracking and delays.

+
def interact(self, debug=False, logger=None)
@@ -6189,6 +6270,7 @@ def test(self, commands, expected, vars = None,*, folder = None, prompt = None,

  • node

      +
    • inject_commands
    • interact
    • run
    • test
      • diff --git a/docs/connpy/services/ai_service.html b/docs/connpy/services/ai_service.html index b3a41c0..4f0169c 100644 --- a/docs/connpy/services/ai_service.html +++ b/docs/connpy/services/ai_service.html @@ -58,10 +58,10 @@ el.replaceWith(d); 
      class AIService(BaseService):
     """Business logic for interacting with AI agents and LLM configurations."""
 
-    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) -> list:
+    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict, last_line: str = "") -> list:
         """Identifies command blocks in the terminal history."""
         blocks = []
-        if not (cmd_byte_positions and len(cmd_byte_positions) >= 2 and raw_bytes):
+        if not raw_bytes:
             return blocks
             
         default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$'
@@ -72,29 +72,63 @@ el.replaceWith(d);
         except Exception:
             prompt_re = re.compile(re.sub(r'(?<!\\)\$', '', default_prompt))
             
-        for i in range(1, len(cmd_byte_positions)):
-            pos, known_cmd = cmd_byte_positions[i]
-            prev_pos = cmd_byte_positions[i-1][0]
-            
-            if known_cmd:
-                prev_chunk = raw_bytes[prev_pos:pos]
-                prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
-                prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
-                prompt_text = prev_lines[-1].strip() if prev_lines else ""
-                preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
-                blocks.append((pos, preview[:80]))
-            else:
-                chunk = raw_bytes[prev_pos:pos]
-                cleaned = log_cleaner(chunk.decode(errors='replace'))
-                lines = [l for l in cleaned.split('\n') if l.strip()]
-                preview = lines[-1].strip() if lines else ""
+        parsed_positions = []
+        if cmd_byte_positions and len(cmd_byte_positions) >= 1:
+            for i in range(1, len(cmd_byte_positions)):
+                pos, known_cmd = cmd_byte_positions[i]
+                prev_pos = cmd_byte_positions[i-1][0]
                 
-                if preview:
-                    match = prompt_re.search(preview)
-                    if match:
-                        cmd_text = preview[match.end():].strip()
-                        if cmd_text:
-                            blocks.append((pos, preview[:80]))
+                if known_cmd:
+                    prev_chunk = raw_bytes[prev_pos:pos]
+                    prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
+                    prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
+                    prompt_text = prev_lines[-1].strip() if prev_lines else ""
+                    preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
+                    parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+                else:
+                    chunk = raw_bytes[prev_pos:pos]
+                    cleaned = log_cleaner(chunk.decode(errors='replace'))
+                    lines = [l for l in cleaned.split('\n') if l.strip()]
+                    preview = lines[-1].strip() if lines else ""
+                    
+                    if preview:
+                        match = prompt_re.search(preview)
+                        if match:
+                            cmd_text = preview[match.end():].strip()
+                            if cmd_text:
+                                parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+                            else:
+                                parsed_positions.append({"pos": pos, "type": "EMPTY_PROMPT", "preview": ""})
+                        else:
+                            parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+                    else:
+                        parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+
+        last_newline = raw_bytes.rfind(b'\n')
+        current_prompt_pos = last_newline + 1 if last_newline != -1 else 0
+        current_end = len(raw_bytes)
+
+        for i, item in enumerate(parsed_positions):
+            if item["type"] == "VALID_CMD":
+                start_pos = item["pos"]
+                preview = item["preview"]
+                
+                # Find the end position: next VALID_CMD or EMPTY_PROMPT
+                end_pos = current_prompt_pos
+                for j in range(i + 1, len(parsed_positions)):
+                    next_item = parsed_positions[j]
+                    if next_item["type"] in ("VALID_CMD", "EMPTY_PROMPT"):
+                        end_pos = next_item["pos"]
+                        break
+                
+                blocks.append((start_pos, end_pos, preview))
+
+        # Always ensure there is a final block representing the current prompt
+        if not blocks:
+            blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+        elif blocks[-1][0] < current_prompt_pos:
+            blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+
         return blocks
 
     def process_copilot_input(self, input_text: str, session_state: dict) -> dict:
@@ -317,17 +351,17 @@ el.replaceWith(d);
 
      Ask the AI copilot for terminal assistance.
    • -def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) ‑> list +def build_context_blocks(self,
    raw_bytes: bytes,
    cmd_byte_positions: list,
    node_info: dict,
    last_line: str = '') ‑> list
    Expand source code -
    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) -> list:
+
    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict, last_line: str = "") -> list:
     """Identifies command blocks in the terminal history."""
     blocks = []
-    if not (cmd_byte_positions and len(cmd_byte_positions) >= 2 and raw_bytes):
+    if not raw_bytes:
         return blocks
         
     default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$'
@@ -338,29 +372,63 @@ el.replaceWith(d);
     except Exception:
         prompt_re = re.compile(re.sub(r'(?<!\\)\$', '', default_prompt))
         
-    for i in range(1, len(cmd_byte_positions)):
-        pos, known_cmd = cmd_byte_positions[i]
-        prev_pos = cmd_byte_positions[i-1][0]
-        
-        if known_cmd:
-            prev_chunk = raw_bytes[prev_pos:pos]
-            prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
-            prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
-            prompt_text = prev_lines[-1].strip() if prev_lines else ""
-            preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
-            blocks.append((pos, preview[:80]))
-        else:
-            chunk = raw_bytes[prev_pos:pos]
-            cleaned = log_cleaner(chunk.decode(errors='replace'))
-            lines = [l for l in cleaned.split('\n') if l.strip()]
-            preview = lines[-1].strip() if lines else ""
+    parsed_positions = []
+    if cmd_byte_positions and len(cmd_byte_positions) >= 1:
+        for i in range(1, len(cmd_byte_positions)):
+            pos, known_cmd = cmd_byte_positions[i]
+            prev_pos = cmd_byte_positions[i-1][0]
             
-            if preview:
-                match = prompt_re.search(preview)
-                if match:
-                    cmd_text = preview[match.end():].strip()
-                    if cmd_text:
-                        blocks.append((pos, preview[:80]))
+            if known_cmd:
+                prev_chunk = raw_bytes[prev_pos:pos]
+                prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
+                prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
+                prompt_text = prev_lines[-1].strip() if prev_lines else ""
+                preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
+                parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+            else:
+                chunk = raw_bytes[prev_pos:pos]
+                cleaned = log_cleaner(chunk.decode(errors='replace'))
+                lines = [l for l in cleaned.split('\n') if l.strip()]
+                preview = lines[-1].strip() if lines else ""
+                
+                if preview:
+                    match = prompt_re.search(preview)
+                    if match:
+                        cmd_text = preview[match.end():].strip()
+                        if cmd_text:
+                            parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+                        else:
+                            parsed_positions.append({"pos": pos, "type": "EMPTY_PROMPT", "preview": ""})
+                    else:
+                        parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+                else:
+                    parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+
+    last_newline = raw_bytes.rfind(b'\n')
+    current_prompt_pos = last_newline + 1 if last_newline != -1 else 0
+    current_end = len(raw_bytes)
+
+    for i, item in enumerate(parsed_positions):
+        if item["type"] == "VALID_CMD":
+            start_pos = item["pos"]
+            preview = item["preview"]
+            
+            # Find the end position: next VALID_CMD or EMPTY_PROMPT
+            end_pos = current_prompt_pos
+            for j in range(i + 1, len(parsed_positions)):
+                next_item = parsed_positions[j]
+                if next_item["type"] in ("VALID_CMD", "EMPTY_PROMPT"):
+                    end_pos = next_item["pos"]
+                    break
+            
+            blocks.append((start_pos, end_pos, preview))
+
+    # Always ensure there is a final block representing the current prompt
+    if not blocks:
+        blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+    elif blocks[-1][0] < current_prompt_pos:
+        blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+
     return blocks

    Identifies command blocks in the terminal history.

    diff --git a/docs/connpy/services/index.html b/docs/connpy/services/index.html index 8b626a3..2e52a24 100644 --- a/docs/connpy/services/index.html +++ b/docs/connpy/services/index.html @@ -113,10 +113,10 @@ el.replaceWith(d); 
    class AIService(BaseService):
     """Business logic for interacting with AI agents and LLM configurations."""
 
-    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) -> list:
+    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict, last_line: str = "") -> list:
         """Identifies command blocks in the terminal history."""
         blocks = []
-        if not (cmd_byte_positions and len(cmd_byte_positions) >= 2 and raw_bytes):
+        if not raw_bytes:
             return blocks
             
         default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$'
@@ -127,29 +127,63 @@ el.replaceWith(d);
         except Exception:
             prompt_re = re.compile(re.sub(r'(?<!\\)\$', '', default_prompt))
             
-        for i in range(1, len(cmd_byte_positions)):
-            pos, known_cmd = cmd_byte_positions[i]
-            prev_pos = cmd_byte_positions[i-1][0]
-            
-            if known_cmd:
-                prev_chunk = raw_bytes[prev_pos:pos]
-                prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
-                prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
-                prompt_text = prev_lines[-1].strip() if prev_lines else ""
-                preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
-                blocks.append((pos, preview[:80]))
-            else:
-                chunk = raw_bytes[prev_pos:pos]
-                cleaned = log_cleaner(chunk.decode(errors='replace'))
-                lines = [l for l in cleaned.split('\n') if l.strip()]
-                preview = lines[-1].strip() if lines else ""
+        parsed_positions = []
+        if cmd_byte_positions and len(cmd_byte_positions) >= 1:
+            for i in range(1, len(cmd_byte_positions)):
+                pos, known_cmd = cmd_byte_positions[i]
+                prev_pos = cmd_byte_positions[i-1][0]
                 
-                if preview:
-                    match = prompt_re.search(preview)
-                    if match:
-                        cmd_text = preview[match.end():].strip()
-                        if cmd_text:
-                            blocks.append((pos, preview[:80]))
+                if known_cmd:
+                    prev_chunk = raw_bytes[prev_pos:pos]
+                    prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
+                    prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
+                    prompt_text = prev_lines[-1].strip() if prev_lines else ""
+                    preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
+                    parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+                else:
+                    chunk = raw_bytes[prev_pos:pos]
+                    cleaned = log_cleaner(chunk.decode(errors='replace'))
+                    lines = [l for l in cleaned.split('\n') if l.strip()]
+                    preview = lines[-1].strip() if lines else ""
+                    
+                    if preview:
+                        match = prompt_re.search(preview)
+                        if match:
+                            cmd_text = preview[match.end():].strip()
+                            if cmd_text:
+                                parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+                            else:
+                                parsed_positions.append({"pos": pos, "type": "EMPTY_PROMPT", "preview": ""})
+                        else:
+                            parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+                    else:
+                        parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+
+        last_newline = raw_bytes.rfind(b'\n')
+        current_prompt_pos = last_newline + 1 if last_newline != -1 else 0
+        current_end = len(raw_bytes)
+
+        for i, item in enumerate(parsed_positions):
+            if item["type"] == "VALID_CMD":
+                start_pos = item["pos"]
+                preview = item["preview"]
+                
+                # Find the end position: next VALID_CMD or EMPTY_PROMPT
+                end_pos = current_prompt_pos
+                for j in range(i + 1, len(parsed_positions)):
+                    next_item = parsed_positions[j]
+                    if next_item["type"] in ("VALID_CMD", "EMPTY_PROMPT"):
+                        end_pos = next_item["pos"]
+                        break
+                
+                blocks.append((start_pos, end_pos, preview))
+
+        # Always ensure there is a final block representing the current prompt
+        if not blocks:
+            blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+        elif blocks[-1][0] < current_prompt_pos:
+            blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+
         return blocks
 
     def process_copilot_input(self, input_text: str, session_state: dict) -> dict:
@@ -372,17 +406,17 @@ el.replaceWith(d);
 
    Ask the AI copilot for terminal assistance.
    -def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) ‑> list +def build_context_blocks(self,
    raw_bytes: bytes,
    cmd_byte_positions: list,
    node_info: dict,
    last_line: str = '') ‑> list
    Expand source code -
    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict) -> list:
+
    def build_context_blocks(self, raw_bytes: bytes, cmd_byte_positions: list, node_info: dict, last_line: str = "") -> list:
     """Identifies command blocks in the terminal history."""
     blocks = []
-    if not (cmd_byte_positions and len(cmd_byte_positions) >= 2 and raw_bytes):
+    if not raw_bytes:
         return blocks
         
     default_prompt = r'>$|#$|\$$|>.$|#.$|\$.$'
@@ -393,29 +427,63 @@ el.replaceWith(d);
     except Exception:
         prompt_re = re.compile(re.sub(r'(?<!\\)\$', '', default_prompt))
         
-    for i in range(1, len(cmd_byte_positions)):
-        pos, known_cmd = cmd_byte_positions[i]
-        prev_pos = cmd_byte_positions[i-1][0]
-        
-        if known_cmd:
-            prev_chunk = raw_bytes[prev_pos:pos]
-            prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
-            prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
-            prompt_text = prev_lines[-1].strip() if prev_lines else ""
-            preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
-            blocks.append((pos, preview[:80]))
-        else:
-            chunk = raw_bytes[prev_pos:pos]
-            cleaned = log_cleaner(chunk.decode(errors='replace'))
-            lines = [l for l in cleaned.split('\n') if l.strip()]
-            preview = lines[-1].strip() if lines else ""
+    parsed_positions = []
+    if cmd_byte_positions and len(cmd_byte_positions) >= 1:
+        for i in range(1, len(cmd_byte_positions)):
+            pos, known_cmd = cmd_byte_positions[i]
+            prev_pos = cmd_byte_positions[i-1][0]
             
-            if preview:
-                match = prompt_re.search(preview)
-                if match:
-                    cmd_text = preview[match.end():].strip()
-                    if cmd_text:
-                        blocks.append((pos, preview[:80]))
+            if known_cmd:
+                prev_chunk = raw_bytes[prev_pos:pos]
+                prev_cleaned = log_cleaner(prev_chunk.decode(errors='replace'))
+                prev_lines = [l for l in prev_cleaned.split('\n') if l.strip()]
+                prompt_text = prev_lines[-1].strip() if prev_lines else ""
+                preview = f"{prompt_text}{known_cmd}" if prompt_text else known_cmd
+                parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+            else:
+                chunk = raw_bytes[prev_pos:pos]
+                cleaned = log_cleaner(chunk.decode(errors='replace'))
+                lines = [l for l in cleaned.split('\n') if l.strip()]
+                preview = lines[-1].strip() if lines else ""
+                
+                if preview:
+                    match = prompt_re.search(preview)
+                    if match:
+                        cmd_text = preview[match.end():].strip()
+                        if cmd_text:
+                            parsed_positions.append({"pos": pos, "type": "VALID_CMD", "preview": preview[:80]})
+                        else:
+                            parsed_positions.append({"pos": pos, "type": "EMPTY_PROMPT", "preview": ""})
+                    else:
+                        parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+                else:
+                    parsed_positions.append({"pos": pos, "type": "SCROLLING", "preview": ""})
+
+    last_newline = raw_bytes.rfind(b'\n')
+    current_prompt_pos = last_newline + 1 if last_newline != -1 else 0
+    current_end = len(raw_bytes)
+
+    for i, item in enumerate(parsed_positions):
+        if item["type"] == "VALID_CMD":
+            start_pos = item["pos"]
+            preview = item["preview"]
+            
+            # Find the end position: next VALID_CMD or EMPTY_PROMPT
+            end_pos = current_prompt_pos
+            for j in range(i + 1, len(parsed_positions)):
+                next_item = parsed_positions[j]
+                if next_item["type"] in ("VALID_CMD", "EMPTY_PROMPT"):
+                    end_pos = next_item["pos"]
+                    break
+            
+            blocks.append((start_pos, end_pos, preview))
+
+    # Always ensure there is a final block representing the current prompt
+    if not blocks:
+        blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+    elif blocks[-1][0] < current_prompt_pos:
+        blocks.append((current_prompt_pos, current_end, last_line[:80] if last_line else "CURRENT CONTEXT"))
+
     return blocks

    Identifies command blocks in the terminal history.

    diff --git a/docs/connpy/utils.html b/docs/connpy/utils.html index a75f527..5097c45 100644 --- a/docs/connpy/utils.html +++ b/docs/connpy/utils.html @@ -75,14 +75,34 @@ el.replaceWith(d); elif token in ('\b', '\x7f'): if cursor > 0: cursor -= 1 - elif token == '\x1B[D': # Left Arrow - if cursor > 0: - cursor -= 1 - elif token == '\x1B[C': # Right Arrow - if cursor < len(buffer): - cursor += 1 - elif token == '\x1B[K': # Clear to end of line - buffer = buffer[:cursor] + elif token.startswith('\x1B[') and len(token) >= 3: + # Parse CSI: \x1B[ <params> <final_char> + final = token[-1] + param_str = token[2:-1] + n = int(param_str) if param_str.isdigit() else 1 + + if final == 'D': # CUB – Cursor Back + cursor = max(0, cursor - n) + elif final == 'C': # CUF – Cursor Forward + cursor = min(len(buffer), cursor + n) + elif final == 'K': # EL – Erase in Line + if n == 0 or param_str == '': # Clear to end + buffer = buffer[:cursor] + elif n == 1: # Clear to start + buffer[:cursor] = [' '] * cursor + elif n == 2: # Clear entire line + buffer = [] + cursor = 0 + elif final == 'G': # CHA – Cursor Horizontal Absolute (1-indexed) + cursor = max(0, n - 1) + # Pad buffer if cursor is beyond current length + if cursor > len(buffer): + buffer.extend([' '] * (cursor - len(buffer))) + elif final == 'P': # DCH – Delete Characters + del buffer[cursor:cursor + n] + elif final == '@': # ICH – Insert Characters + buffer[cursor:cursor] = [' '] * n + # All other CSI sequences are silently discarded elif token.startswith('\x1B'): continue elif len(token) == 1 and ord(token) < 32: