diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index a30fd0b..09e8623 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -98,7 +98,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: } ) - # Replay login transcript once, then draw a live prompt so first keystrokes match later lines. + # Replay full login transcript (kept for StrictMode remount / brief reconnect). bootstrap = bytes(sess.bootstrap_output or b"") if bootstrap: try: @@ -107,8 +107,6 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: ) except Exception: _log.debug("webcrt bootstrap send failed session=%s", session_id, exc_info=True) - # Only replay login banner on the first attach (StrictMode remount / blip). - sess.bootstrap_output = b"" stop = asyncio.Event() diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index 0b2620e..10e57c9 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -158,24 +158,13 @@ def normalize_cli_transcript(text: str) -> str: return "\n".join(lines) -def strip_trailing_prompt_lines(text: str) -> str: - """Remove final prompt line(s) so a live RETURN can paint the interactive prompt.""" - lines = str(text or "").split("\n") - while lines and not str(lines[-1]).strip(): - lines.pop() - while lines and _looks_like_cli_prompt(lines[-1]): - lines.pop() - while lines and not str(lines[-1]).strip(): - lines.pop() - return "\n".join(lines) - - def prepare_bootstrap_output(text: str) -> str: - """Login transcript for UI replay; ends with newline, without the final prompt.""" - body = strip_trailing_prompt_lines(normalize_cli_transcript(text)) - if not body: - return "" - return body if body.endswith("\n") else body + "\n" + """Full login transcript for UI replay; keep final prompt, no trailing newline after it. + + Trailing newline would leave the cursor on a blank line so the first typed line + looks wrong; cursor should sit after the prompt like a real CRT. + """ + return normalize_cli_transcript(text) def _prime_interactive_channel(conn: ConnectHandler) -> None: @@ -491,9 +480,7 @@ def create_session( except Exception: pass - # Prefer session_log (full login transcript). Prime channel, then strip the final - # prompt so WebSocket attach can paint a live interactive prompt (first keystrokes - # then behave like subsequent lines). + # Prefer session_log (full login transcript including final prompt). _prime_interactive_channel(conn) bootstrap = prepare_bootstrap_output(_session_log_text(log_buf)) if not bootstrap.strip(): @@ -522,7 +509,8 @@ def create_session( cli_keymap=cli_keymap, conn=conn, bootstrap_output=str(bootstrap or "").encode("utf-8", errors="replace"), - needs_live_prompt=True, + # Only nudge a live prompt when transcript has no recognizable prompt yet. + needs_live_prompt=not _looks_like_cli_prompt(bootstrap), ) # Keep bootstrap for WS attach replay; do not rely solely on out_queue (StrictMode remount). sess.start_reader() diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index a701f33..fe4ce07 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -79,9 +79,8 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(svc.map_network_cli_enter("\r", telnet), "\r\n") # type: ignore[arg-type] self.assertEqual(svc.normalize_cli_transcript("R2#R2#\nR2#"), "R2#") self.assertEqual(svc.normalize_cli_transcript("banner\nR2#R2#"), "banner\nR2#") - self.assertEqual(svc.prepare_bootstrap_output("login\nR2#\nR2#"), "login\n") - self.assertTrue(svc.prepare_bootstrap_output("login\nR2#").endswith("\n")) - self.assertFalse(svc._looks_like_cli_prompt(svc.prepare_bootstrap_output("login\nR2#").rstrip("\n") or "x")) + self.assertEqual(svc.prepare_bootstrap_output("login\nR2#\nR2#"), "login\nR2#") + self.assertTrue(svc.prepare_bootstrap_output("login\nR2#").endswith("R2#")) @patch.object(svc, "_audit") @patch.object(svc, "open_netmiko_connection") @@ -172,9 +171,8 @@ class WebcrtServiceTests(unittest.TestCase): assert sess is not None boot = sess.bootstrap_output.decode("utf-8", errors="replace") self.assertIn("Username:huawei", boot) - # Final prompt is stripped; live RETURN on WS attach paints the interactive prompt. - self.assertNotIn("", boot) - self.assertTrue(sess.needs_live_prompt) + self.assertIn("", boot) + self.assertFalse(sess.needs_live_prompt) before = list(fake.written) sess.write_stdin("\n") self.assertEqual(fake.written[len(before) :], ["\n"])