[luv-38] Restore the terminal when a connection dies mid-session (#38)

* [luv-38] Restore the terminal when a connection dies mid-session

A dropped SSH connection left the local terminal in the remote program's
modes: mouse tracking on, so every mouse move typed "35;22;1M" at the
shell prompt, plus bracketed paste, a stuck alternate screen and raw mode.
tmux and the agent TUI turn those on and only turn them off when they exit
cleanly; a broken pipe skips that, and exec_ssh's execv meant luv had
already become ssh, so nothing of ours was left alive to clean up.

Add hand_over(), which keeps luv alive as the child's parent purely to
restore the terminal afterwards — termios settings plus the DEC private
modes only the remote program knew it had set. The child stays in the same
foreground process group, so Ctrl-C, SIGWINCH and exit-code passthrough
behave as they did under execv. The -nit pipe path keeps plain execv:
no terminal there to leave dirty.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012UYshGx4wKQ4covMGukc4r

* [luv-38] Bump version to 0.2.1

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012UYshGx4wKQ4covMGukc4r

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: NiveditJain

docs/remote-sessions.md

@@ -343,6 +343,19 @@ happen. Type the prompt into the running agent instead.
 It shows the last known state rather than pruning; nothing is lost. When a host
 is gone for good, `luv ls --prune` forgets its entries.
 
+**Junk like `35;22;1M` appears at my prompt after a dropped connection**
+
+That's mouse-tracking reports. The remote tmux/agent turned mouse tracking on
+in *your* terminal and, having been killed along with the connection, never
+turned it off — so every mouse move now types coordinates at your shell.
+Bracketed paste (`200~` around pastes), a missing cursor, and a stuck
+alternate screen come from the same cause.
+
+luv now cleans this up itself: it stays alive as the parent of `ssh` and
+restores the terminal whatever happens to the connection. If you land in this
+state from something else, `reset` (or `stty sane` plus
+`printf '\033[?1003l\033[?1006l\033[?2004l'`) clears it.
+
 **A session is wedged**
 
 ```bash

luv/__init__.py

@@ -12,6 +12,11 @@
 import time
 from pathlib import Path
 
+try:
+    import termios
+except ImportError:  # not POSIX — the ssh/tmux handoffs don't run there either
+    termios = None
+
 LUV_DIR = Path.home() / ".luv"
 CONFIG_FILE = LUV_DIR / "config.json"
 SESSIONS_FILE = LUV_DIR / "sessions.json"
@@ -1153,21 +1158,109 @@ def open_pr(org: str, repo: str, number: int, prompt: str | None, nav_mode: bool
         launch(clone_dir, prompt, plan_mode=plan_mode, non_interactive=non_interactive, extra_env=extra_env, model=model, agent=agent)
 
 
-def exec_ssh(hc: dict, remote_cmd: str, *, tty: bool = True) -> None:
-    """Hand the terminal to ssh, replacing this process.
+# Terminal modes a full-screen program switches on and is expected to switch
+# off again on its way out. A connection that dies mid-session never gets to,
+# and the leftovers are user-visible: mouse tracking turns every mouse move
+# into "35;22;1M" junk at the shell prompt, bracketed paste wraps pastes in
+# "200~", and the alternate screen swallows the scrollback.
+TERM_RESET = (
+    "\x1b[?1000l\x1b[?1001l\x1b[?1002l\x1b[?1003l"   # mouse tracking off
+    "\x1b[?1004l"                                    # focus reporting off
+    "\x1b[?1005l\x1b[?1006l\x1b[?1015l\x1b[?1016l"   # mouse report encodings off
+    "\x1b[?2004l"                                    # bracketed paste off
+    "\x1b[?1049l"                                    # back to the primary screen
+    "\x1b[?1l\x1b>"                                  # normal cursor keys, keypad
+    "\x1b[r"                                         # full-height scroll region
+    "\x1b[?7h\x1b[4l\x1b[?25h\x1b[0m"                # wrap, replace, cursor, colours
+)
+
+
+def terminal_fd() -> int | None:
+    """The fd our terminal is on, or None when there isn't one.
+
+    stdin first: it is the one a redirect is least likely to have taken away,
+    and termios wants the terminal device rather than whichever stream happens
+    to still point at it.
+    """
+    for stream in (sys.stdin, sys.stdout, sys.stderr):
+        try:
+            fd = stream.fileno()
+        except (AttributeError, ValueError, OSError):
+            continue
+        if os.isatty(fd):
+            return fd
+    return None
+
+
+@contextlib.contextmanager
+def terminal_guard():
+    """Restore the terminal on the way out, however we get there.
 
-    execv (rather than subprocess.run) is what gives correct TTY handling,
-    Ctrl-C, and exit-code passthrough for free — same as the local agent paths.
+    Two layers, because a killed program skips two different kinds of cleanup:
+    termios settings (raw mode, echo) that ssh itself normally puts back, and
+    the DEC private modes the *remote* program turned on, which nothing on this
+    side knows about. Both are cheap no-ops when nothing was broken.
+    """
+    fd = terminal_fd()
+    saved = None
+    if fd is not None and termios is not None:
+        with contextlib.suppress(termios.error, OSError):
+            saved = termios.tcgetattr(fd)
+    try:
+        yield
+    finally:
+        if fd is not None:
+            for stream in (sys.stdout, sys.stderr):
+                with contextlib.suppress(ValueError, OSError):
+                    stream.flush()
+            if saved is not None:
+                with contextlib.suppress(termios.error, OSError):
+                    termios.tcsetattr(fd, termios.TCSADRAIN, saved)
+            with contextlib.suppress(OSError):
+                os.write(fd, TERM_RESET.encode())
+
+
+def hand_over(argv: list[str], *, restore: bool = True) -> None:
+    """Give the terminal to a child and exit with its status. Never returns.
+
+    Without `restore` this is a plain execv, which is the better deal when
+    there is no terminal to leave in a bad state: no process in the middle, and
+    TTY handling, Ctrl-C and the exit code all pass through for free.
+
+    With it, we stay alive as a parent whose only job is to clean up. That
+    costs one process and buys the case this exists for — ssh dying on a broken
+    pipe, taking a remote tmux and its agent TUI with it, with nobody left to
+    turn mouse tracking back off. The child is still in our foreground process
+    group, so it keeps receiving Ctrl-C and SIGWINCH from the terminal driver
+    exactly as it did under execv.
     """
+    if not restore:
+        os.execv(argv[0], argv)
+    else:
+        with terminal_guard():
+            proc = subprocess.Popen(argv)
+            while True:
+                try:
+                    code = proc.wait()
+                    break
+                except KeyboardInterrupt:
+                    # The child got this same Ctrl-C from the tty and decides
+                    # for itself what to do with it; outliving it is the point.
+                    continue
+        sys.exit(code)
+
+
+def exec_ssh(hc: dict, remote_cmd: str, *, tty: bool = True) -> None:
+    """Hand the terminal to ssh. Never returns."""
     ssh_bin = shutil.which("ssh")
     if not ssh_bin:
         die("'ssh' not found in PATH")
     argv = ssh_base(hc, tty=tty) + [remote_shell(remote_cmd)]
-    os.execv(ssh_bin, [ssh_bin] + argv[1:])
+    hand_over([ssh_bin] + argv[1:], restore=tty)
 
 
 def attach_session(hc: dict | None, name: str) -> None:
-    """Attach to a tmux session, locally or over ssh. Replaces this process.
+    """Attach to a tmux session, locally or over ssh. Never returns.
 
     -d detaches other clients so the pane isn't size-clamped to a stale window
     left open elsewhere; these are all the same user's sessions.
@@ -1176,7 +1269,8 @@ def attach_session(hc: dict | None, name: str) -> None:
         tmux_bin = shutil.which("tmux")
         if not tmux_bin:
             die("'tmux' not found in PATH")
-        os.execv(tmux_bin, [tmux_bin, "attach", "-d", "-t", name])
+        hand_over([tmux_bin, "attach", "-d", "-t", name])
+        return
     print(f"luv: attaching {name} on {hc['host']}")
     exec_ssh(hc, shlex.join(["tmux", "attach", "-d", "-t", name]))
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "luv-cli"
-version = "0.2.0"
+version = "0.2.1"
 description = "Launch Claude Code or Codex agents on GitHub repos with isolated workspaces and optional Docker dev environments"
 requires-python = ">=3.10"
 license = "MIT"

tests/test_agents.py

@@ -122,11 +122,11 @@ def _dispatch(self, argv, env=None):
         with (patch.object(sys, "argv", ["luv"] + argv),
               patch.dict(luv.os.environ, env, clear=False),
               patch.object(luv.shutil, "which", side_effect=lambda n: f"/bin/{n}"),
-              patch.object(luv.os, "execv") as execv,
+              patch.object(luv, "hand_over") as hand_over,
               contextlib.redirect_stdout(io.StringIO()),
               contextlib.redirect_stderr(io.StringIO())):
             luv.main()
-        return execv.call_args.args[1] if execv.called else None
+        return hand_over.call_args.args[0] if hand_over.called else None
 
     def test_new_workspace_gets_pending_session(self):
         argv = self._dispatch(["myrepo", "fix it"])
@@ -219,6 +219,129 @@ def test_local_flag_rejects_ssh_flags(self):
             self._dispatch(["--local", "-s", "gpu", "myrepo"])
 
 
+class _FakeProc:
+    """A child that can raise KeyboardInterrupt before it finally exits."""
+
+    def __init__(self, returncode=0, interrupts=0):
+        self.returncode = returncode
+        self.interrupts = interrupts
+
+    def wait(self):
+        if self.interrupts:
+            self.interrupts -= 1
+            raise KeyboardInterrupt
+        return self.returncode
+
+
+class TerminalRestoreTests(unittest.TestCase):
+    """A connection that dies must not leave the terminal in the remote
+    program's modes — that's the "35;22;1M" junk at the shell prompt."""
+
+    FD = 42
+    SAVED = ["saved", "termios", "attrs"]
+
+    def _hand_over(self, argv, returncode=0, interrupts=0, **kwargs):
+        proc = _FakeProc(returncode, interrupts)
+        self.writes = []
+        with (patch.object(luv, "terminal_fd", return_value=self.FD),
+              patch.object(luv.subprocess, "Popen", return_value=proc) as popen,
+              patch.object(luv.termios, "tcgetattr", return_value=self.SAVED),
+              patch.object(luv.termios, "tcsetattr") as tcsetattr,
+              patch.object(luv.os, "execv") as execv,
+              patch.object(luv.os, "write",
+                           side_effect=lambda fd, data: self.writes.append((fd, data))),
+              self.assertRaises(SystemExit) as exit_ctx):
+            luv.hand_over(argv, **kwargs)
+        self.popen, self.tcsetattr, self.execv = popen, tcsetattr, execv
+        return exit_ctx.exception.code
+
+    def _reset_bytes(self):
+        return b"".join(data for fd, data in self.writes if fd == self.FD)
+
+    def test_broken_connection_still_restores_the_terminal(self):
+        # 255 is what ssh exits with when the connection drops under it.
+        code = self._hand_over(["/bin/ssh", "box", "tmux attach"], returncode=255)
+
+        self.assertEqual(code, 255, "the child's exit code must still pass through")
+        self.assertTrue(self.popen.called, "restore mode must not exec the child away")
+        self.assertIn(b"\x1b[?1003l", self._reset_bytes(), "mouse tracking left on")
+        self.assertIn(b"\x1b[?1006l", self._reset_bytes(), "SGR mouse reports left on")
+        self.assertIn(b"\x1b[?2004l", self._reset_bytes(), "bracketed paste left on")
+        self.assertIn(b"\x1b[?1049l", self._reset_bytes(), "alternate screen left on")
+        self.assertEqual(self.tcsetattr.call_args.args[0], self.FD)
+        self.assertEqual(self.tcsetattr.call_args.args[2], self.SAVED)
+
+    def test_clean_exit_restores_too(self):
+        code = self._hand_over(["/bin/tmux", "attach"], returncode=0)
+
+        self.assertEqual(code, 0)
+        self.assertIn(b"\x1b[?1003l", self._reset_bytes())
+
+    def test_ctrl_c_does_not_kill_the_parent_before_cleanup(self):
+        # Ctrl-C reaches the child through the tty; the parent must outlive it
+        # or there is nobody left to clean up after it.
+        code = self._hand_over(["/bin/ssh", "box"], returncode=130, interrupts=2)
+
+        self.assertEqual(code, 130)
+        self.assertIn(b"\x1b[?1003l", self._reset_bytes())
+
+    def test_no_tty_handoff_still_execs(self):
+        # -nit streams stream-json into a pipe: no terminal to restore, so keep
+        # the cheaper exec and don't leave a process in the middle.
+        writes = []
+        with (patch.object(luv, "terminal_fd", return_value=self.FD),
+              patch.object(luv.os, "execv") as execv,
+              patch.object(luv.os, "write", side_effect=writes.append),
+              patch.object(luv.subprocess, "Popen") as popen):
+            luv.hand_over(["/bin/ssh", "box"], restore=False)
+
+        self.assertEqual(execv.call_args.args, ("/bin/ssh", ["/bin/ssh", "box"]))
+        self.assertFalse(popen.called)
+        self.assertEqual(writes, [])
+
+    def test_guard_is_a_noop_without_a_terminal(self):
+        with (patch.object(luv, "terminal_fd", return_value=None),
+              patch.object(luv.os, "write") as write,
+              patch.object(luv.termios, "tcsetattr") as tcsetattr):
+            with luv.terminal_guard():
+                pass
+
+        self.assertFalse(write.called)
+        self.assertFalse(tcsetattr.called)
+
+    def test_restore_survives_a_child_that_never_started(self):
+        # An OSError out of Popen must not skip the cleanup either.
+        writes = []
+        with (patch.object(luv, "terminal_fd", return_value=self.FD),
+              patch.object(luv.termios, "tcgetattr", return_value=self.SAVED),
+              patch.object(luv.termios, "tcsetattr"),
+              patch.object(luv.os, "write",
+                           side_effect=lambda fd, data: writes.append(data)),
+              patch.object(luv.subprocess, "Popen", side_effect=OSError("boom"))):
+            with self.assertRaises(OSError):
+                luv.hand_over(["/bin/ssh", "box"])
+
+        self.assertIn(b"\x1b[?1003l", b"".join(writes))
+
+    def test_local_attach_goes_through_the_guard(self):
+        with (patch.object(luv.shutil, "which", side_effect=lambda n: f"/bin/{n}"),
+              patch.object(luv, "hand_over") as hand_over):
+            luv.attach_session(None, "luv-myrepo-42")
+
+        self.assertEqual(hand_over.call_args.args[0],
+                         ["/bin/tmux", "attach", "-d", "-t", "luv-myrepo-42"])
+
+    def test_remote_attach_asks_for_a_tty_and_restores(self):
+        with (patch.object(luv.shutil, "which", side_effect=lambda n: f"/bin/{n}"),
+              patch.object(luv, "hand_over") as hand_over,
+              contextlib.redirect_stdout(io.StringIO())):
+            luv.attach_session({"host": "box"}, "luv-myrepo-42")
+
+        argv = hand_over.call_args.args[0]
+        self.assertIn("-t", argv)
+        self.assertNotEqual(hand_over.call_args.kwargs.get("restore"), False)
+
+
 class SessionNameTests(unittest.TestCase):
     def test_illegal_tmux_characters_are_replaced(self):
         self.assertEqual(luv.tmux_session_name("foo.js-7"), "luv-foo_js-7")