diff --git a/src/jupyter_runtime.py b/src/jupyter_runtime.py index c5df8bb..268b5b0 100644 --- a/src/jupyter_runtime.py +++ b/src/jupyter_runtime.py @@ -542,7 +542,7 @@ PY pid_file = shlex.quote(f'{run_dir}/{marker}.pid') env_prefix = self._remote_env_prefix() inner_command = f'{env_prefix}{command}' - wrapped = ( + wrapper_script = ( f'mkdir -p {workspace} {quoted_run_dir} && ' f'cd {workspace} && ' f'printf "\\n{start_marker}\\n"; ' @@ -561,6 +561,11 @@ PY f'rm -f {pid_file}; ' f'printf "\\n{marker}:%s\\n" "$__zk_status"' ) + # Run the wrapper inside a foreground non-interactive bash. Sending the + # whole background-job wrapper directly to the terminal shell can make + # the interactive shell print job-control lines such as + # "[1] + Done (setsid bash -lc ...)", which can hide the real stdout. + wrapped = f'bash -lc {shlex.quote(wrapper_script)}' with self._lock: name = self._ensure_terminal() ws_url = self._terminal_websocket_url(name) @@ -1288,11 +1293,27 @@ def clean_terminal_output(text: str) -> str: lines = [ line.rstrip('\r') for line in cleaned.splitlines() - if line.strip() not in {'#', '>'} and not re.fullmatch(r'(>\s*)+', line.strip()) + if ( + line.strip() not in {'#', '>'} + and not re.fullmatch(r'(>\s*)+', line.strip()) + and not is_wrapper_job_control_line(line.strip()) + ) ] return '\n'.join(lines).strip('\r\n') +def is_wrapper_job_control_line(line: str) -> bool: + """Return true for terminal job-control notices created by our wrapper.""" + if 'setsid bash -lc' not in line: + return False + return bool( + re.match( + r'^\[\d+\]\s*[+-]?\s*(Done|done|Terminated|terminated|Killed|killed|Exit \d+)', + line, + ) + ) + + def extract_terminal_command_output(text: str, *, start_marker: str) -> str: """Drop terminal echo/prompt text before the wrapper's real start marker.""" if not start_marker: diff --git a/tests/test_jupyter_runtime.py b/tests/test_jupyter_runtime.py index b00d7c6..fd34875 100644 --- a/tests/test_jupyter_runtime.py +++ b/tests/test_jupyter_runtime.py @@ -35,6 +35,14 @@ class TestJupyterRuntimeOutputFiltering(unittest.TestCase): self.assertEqual(output, 'actual') + def test_clean_terminal_output_drops_wrapper_job_control_line(self) -> None: + cleaned = clean_terminal_output( + 'real-output\r\n' + '[1] + Done (setsid bash -lc "export A=1; lscpu")\r\n' + ) + + self.assertEqual(cleaned, 'real-output') + if __name__ == '__main__': unittest.main()