Make terminal_class configurable and provide a manager to detect terminal status on linux systems

jupyter_server/serverapp.py

@@ -351,6 +351,7 @@ def init_settings(
             server_root_dir=root_dir,
             jinja2_env=env,
             terminals_available=terminado_available and jupyter_app.terminals_enabled,
+            stateful_terminals_enabled=jupyter_app.stateful_terminals_enabled,
             serverapp=jupyter_app,
         )
 
@@ -1671,6 +1672,18 @@ def _update_server_extensions(self, change):
         ),
     )
 
+    stateful_terminals_enabled = Bool(
+        False,
+        config=True,
+        help=_i18n(
+            """Set to True to enable stateful terminals.
+
+            Terminals may also be automatically disabled if the terminado package
+         is not available.
+            """
+        ),
+    )
+
     # Since use of terminals is also a function of whether the terminado package is
     # available, this variable holds the "final indication" of whether terminal functionality
     # should be considered (particularly during shutdown/cleanup).  It is enabled only

jupyter_server/terminal/__init__.py

@@ -12,7 +12,7 @@
 from jupyter_server.utils import url_path_join as ujoin
 from . import api_handlers
 from .handlers import TermSocket
-from .terminalmanager import TerminalManager
+from .terminalmanager import TerminalManager, StatefulTerminalManager
 
 
 def initialize(webapp, root_dir, connection_url, settings):
@@ -29,7 +29,12 @@ def initialize(webapp, root_dir, connection_url, settings):
     # the user has specifically set a preferred shell command.
     if os.name != "nt" and shell_override is None and not sys.stdout.isatty():
         shell.append("-l")
-    terminal_manager = webapp.settings["terminal_manager"] = TerminalManager(
+    terminal_class = (
+        StatefulTerminalManager
+        if webapp.settings["stateful_terminals_enabled"]
+        else TerminalManager
+    )
+    terminal_manager = webapp.settings["terminal_manager"] = terminal_class(
         shell_command=shell,
         extra_env={
             "JUPYTER_SERVER_ROOT": root_dir,

jupyter_server/terminal/handlers.py

@@ -27,6 +27,7 @@ def get(self, *args, **kwargs):
     def on_message(self, message):
         super(TermSocket, self).on_message(message)
         self._update_activity()
+        self._set_state_busy_if_stateful()
 
     def write_message(self, message, binary=False):
         super(TermSocket, self).write_message(message, binary=binary)
@@ -37,3 +38,9 @@ def _update_activity(self):
         # terminal may not be around on deletion/cull
         if self.term_name in self.terminal_manager.terminals:
             self.terminal_manager.terminals[self.term_name].last_activity = utcnow()
+
+    def _set_state_busy_if_stateful(self):
+        if not hasattr(self.terminal_manager, "set_state_idle_if_return"):
+            return
+        if self.term_name in self.terminal_manager.terminals:
+            self.terminal_manager.terminals[self.term_name].execution_state = "busy"

jupyter_server/terminal/terminalmanager.py

@@ -7,6 +7,7 @@
 from datetime import timedelta
 
 import terminado
+from terminado.management import _poll
 from tornado import web
 from tornado.ioloop import IOLoop
 from tornado.ioloop import PeriodicCallback
@@ -102,7 +103,7 @@ def get_terminal_model(self, name):
     def _check_terminal(self, name):
         """Check a that terminal 'name' exists and raise 404 if not."""
         if name not in self.terminals:
-            raise web.HTTPError(404, u"Terminal not found: %s" % name)
+            raise web.HTTPError(404, "Terminal not found: %s" % name)
 
     def _initialize_culler(self):
         """Start culler if 'cull_inactive_timeout' is greater than zero.
@@ -165,3 +166,57 @@ async def _cull_inactive_terminal(self, name):
                     "Culling terminal '%s' due to %s seconds of inactivity.", name, inactivity
                 )
                 await self.terminate(name, force=True)
+
+
+class StatefulTerminalManager(TerminalManager):
+    # patch execution_state into terminal
+    def pty_read(self, fd, events=None):
+        """Called by the event loop when there is pty data ready to read."""
+        # prevent blocking on fd
+        if not _poll(fd, timeout=0.1):  # 100ms
+            self.log.debug(f"Spurious pty_read() on fd {fd}")
+            return
+        ptywclients = self.ptys_by_fd[fd]
+        try:
+            s = ptywclients.ptyproc.read(65536)
+            client_list = ptywclients.clients
+            ptywclients.read_buffer.append(s)
+            # hook for set terminal status
+            self.set_state_idle_if_return(ptywclients, s)
+            if not client_list:
+                # No one to consume our output: buffer it.
+                ptywclients.preopen_buffer.append(s)
+                return
+            for client in ptywclients.clients:
+                client.on_pty_read(s)
+        except EOFError:
+            self.on_eof(ptywclients)
+            for client in ptywclients.clients:
+                client.on_pty_died()
+
+    def set_state_idle_if_return(self, ptywclients, s):
+        first_stdout = getattr(ptywclients, "first_stdout", "")
+
+        if not first_stdout:
+            # Record the first output to identify the terminal return
+            # It works well for jupyterhub-singleuser and should also work for other debian-based mirrors
+            # fixme: May fail if terminal is not properly separated with ':' or change user after connect
+            #        (Any change to the user, hostname or environment may render it invalid)
+            first_stdout = s.split(":")[0].lstrip()
+            ptywclients.first_stdout = first_stdout
+            self.log.debug(f'take "{first_stdout}" as terminal returned')
+        if s.lstrip().startswith(first_stdout):
+            self._set_state_idle(ptywclients)
+
+    def _set_state_idle(self, ptywclients):
+        self.log.debug("set terminal execution_state as idle")
+        ptywclients.execution_state = "idle"
+
+    def get_terminal_model(self, name):
+        """Return a JSON-safe dict representing a terminal.
+        For use in representing terminals in the JSON APIs.
+        """
+        model = super(StatefulTerminalManager, self).get_terminal_model(name)
+        term = self.terminals[name]
+        model.setdefault("execution_state", getattr(term, "execution_state", "not connected yet"))
+        return model

jupyter_server/tests/test_stateful_terminal.py

@@ -0,0 +1,123 @@
+import asyncio
+import json
+import os
+import platform
+import shutil
+
+import pytest
+from traitlets.config import Config
+
+
+@pytest.fixture
+def terminal_path(tmp_path):
+    subdir = tmp_path.joinpath("terminal_path")
+    subdir.mkdir()
+
+    yield subdir
+
+    shutil.rmtree(str(subdir), ignore_errors=True)
+
+
+CULL_TIMEOUT = 10
+CULL_INTERVAL = 3
+
+
+@pytest.fixture
+def jp_server_config():
+    return Config(
+        {
+            "ServerApp": {
+                "stateful_terminals_enabled": True,
+            }
+        }
+    )
+
+
+async def test_set_idle(jp_fetch, jp_ws_fetch, jp_cleanup_subprocesses, jp_serverapp):
+    if platform.system().lower() != "linux":
+        return
+
+    # disable man sudo_root
+    os.system(f"touch {os.path.expanduser('~/.sudo_as_admin_successful')}")
+
+    resp = await jp_fetch(
+        "api",
+        "terminals",
+        method="POST",
+        allow_nonstandard_methods=True,
+    )
+    term = json.loads(resp.body.decode())
+    term_1 = term["name"]
+    ws = await jp_ws_fetch("terminals", "websocket", term_1)
+    setup = ["set_size", 0, 0, 80, 32]
+    await ws.write_message(json.dumps(setup))
+    while True:
+        try:
+            await asyncio.wait_for(ws.read_message(), timeout=1)
+        except asyncio.TimeoutError:
+            break
+    sleep_2_msg = ["stdin", "python -c 'import time;time.sleep(2)'\r\n"]
+    await ws.write_message(json.dumps(sleep_2_msg))
+    await asyncio.sleep(1)
+    assert (
+        jp_serverapp.web_app.settings["terminal_manager"].terminals[term_1].execution_state
+        == "busy"
+    )
+    await asyncio.sleep(2)
+    message_stdout = ""
+    while True:
+        try:
+            message = await asyncio.wait_for(ws.read_message(), timeout=5.0)
+        except asyncio.TimeoutError:
+            break
+
+        message = json.loads(message)
+
+        if message[0] == "stdout":
+            message_stdout += message[1]
+    print(message_stdout)
+    assert (
+        jp_serverapp.web_app.settings["terminal_manager"].terminals[term_1].execution_state
+        == "idle"
+    )
+    await jp_cleanup_subprocesses()
+
+
+async def test_set_idle_disconnect(jp_fetch, jp_ws_fetch, jp_cleanup_subprocesses, jp_serverapp):
+    if os.name != "nt":
+        pytest.skip('Feature not supported on Windows')
+
+    # disable man sudo_root
+    os.system(f"touch {os.path.expanduser('~/.sudo_as_admin_successful')}")
+
+    resp = await jp_fetch(
+        "api",
+        "terminals",
+        method="POST",
+        allow_nonstandard_methods=True,
+    )
+    term = json.loads(resp.body.decode())
+    term_1 = term["name"]
+    ws = await jp_ws_fetch("terminals", "websocket", term_1)
+    setup = ["set_size", 0, 0, 80, 32]
+    await ws.write_message(json.dumps(setup))
+    while True:
+        try:
+            await asyncio.wait_for(ws.read_message(), timeout=1)
+        except asyncio.TimeoutError:
+            break
+    sleep_2_msg = ["stdin", "python -c 'import time;time.sleep(2)'\r\n"]
+    await ws.write_message(json.dumps(sleep_2_msg))
+    ws.close()
+    await asyncio.sleep(1)
+    assert (
+        jp_serverapp.web_app.settings["terminal_manager"].terminals[term_1].execution_state
+        == "busy"
+    )
+    await asyncio.sleep(2)
+    assert not jp_serverapp.web_app.settings["terminal_manager"].terminals[term_1].clients
+    assert (
+        jp_serverapp.web_app.settings["terminal_manager"].terminals[term_1].execution_state
+        == "idle"
+    )
+    await jp_cleanup_subprocesses()