Stabilize Jupyter terminal wrapper execution

src/jupyter_runtime.py

@@ -503,6 +503,9 @@ PY
             )
 
     def put_file_bytes(self, api_path: str, data: bytes) -> None:
+        parent_api = '/'.join(api_path.split('/')[:-1])
+        if parent_api:
+            self._ensure_remote_dir(parent_api)
         encoded = base64.b64encode(data).decode('ascii')
         url = f'{self.binding.base_url}/api/contents/{quote(api_path, safe="/")}'
         response = self._session.put(
@@ -538,6 +541,8 @@ PY
         start_marker = f'__ZK_AGENT_START_{uuid.uuid4().hex}__'
         workspace = shlex.quote(self.binding.workspace_cwd)
         run_dir = f'{self.binding.workspace_cwd}/scratchpad/.run_pids'
+        script_dir = self._terminal_script_dir()
+        script_path = f'{script_dir}/{marker}.sh'
         quoted_run_dir = shlex.quote(run_dir)
         pid_file = shlex.quote(f'{run_dir}/{marker}.pid')
         env_prefix = self._remote_env_prefix()
@@ -561,11 +566,20 @@ 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)}'
+        # Keep the terminal input simple. Sending a deeply nested one-line
+        # script directly to the interactive shell is fragile: complex
+        # if/then/fi command strings can trigger continuation prompts, and
+        # background jobs can print "[1] + Done ..." job-control noise. Upload
+        # the wrapper through the Jupyter Contents API, then execute that temp
+        # script in the foreground.
+        self.put_file_bytes(
+            api_path_for_absolute_path(script_path),
+            wrapper_script.encode('utf-8'),
+        )
+        wrapped = build_terminal_script_launcher(
+            script_path=script_path,
+            marker=marker,
+        )
         with self._lock:
             name = self._ensure_terminal()
             ws_url = self._terminal_websocket_url(name)
@@ -599,6 +613,12 @@ PY
                 except Exception:
                     pass
 
+    def _terminal_script_dir(self) -> str:
+        session_part = sanitize_remote_path_part(self.binding.session_id)
+        # Jupyter Contents API usually rejects hidden path components, so this
+        # directory must not start with a dot.
+        return f'{self.binding.workspace_root}/run_scripts/{session_part}'
+
     def run_python(
         self,
         *,
@@ -1295,7 +1315,7 @@ def clean_terminal_output(text: str) -> str:
         for line in cleaned.splitlines()
         if (
             line.strip() not in {'#', '>'}
-            and not re.fullmatch(r'(>\s*)+', line.strip())
+            and not re.fullmatch(r'(#\s*)?(>\s*)+', line.strip())
             and not is_wrapper_job_control_line(line.strip())
         )
     ]
@@ -1314,6 +1334,22 @@ def is_wrapper_job_control_line(line: str) -> bool:
     )
 
 
+def build_terminal_script_launcher(
+    script_path: str,
+    *,
+    marker: str,
+) -> str:
+    quoted_path = shlex.quote(script_path)
+    return (
+        f'bash {quoted_path}; '
+        '__zk_outer_status=$?; '
+        f'rm -f {quoted_path}; '
+        'if [ "$__zk_outer_status" -ne 0 ]; then '
+        f'printf "\\n{marker}:%s\\n" "$__zk_outer_status"; '
+        'fi'
+    )
+
+
 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:

tests/test_jupyter_runtime.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import unittest
 
 from src.jupyter_runtime import (
+    build_terminal_script_launcher,
     clean_terminal_output,
     extract_terminal_command_output,
 )
@@ -43,6 +44,28 @@ class TestJupyterRuntimeOutputFiltering(unittest.TestCase):
 
         self.assertEqual(cleaned, 'real-output')
 
+    def test_clean_terminal_output_drops_continuation_prompt(self) -> None:
+        cleaned = clean_terminal_output('real-output\r\n# >\r\n')
+
+        self.assertEqual(cleaned, 'real-output')
+
+    def test_terminal_script_launcher_hides_complex_wrapper_from_shell(self) -> None:
+        launcher = build_terminal_script_launcher(
+            script_path='/workspace/scratchpad/.run_scripts/run.sh',
+            marker='__ZK_AGENT_EXIT_test__',
+        )
+
+        self.assertNotIn('setsid bash -lc', launcher)
+        self.assertEqual(
+            launcher,
+            'bash /workspace/scratchpad/.run_scripts/run.sh; '
+            '__zk_outer_status=$?; '
+            'rm -f /workspace/scratchpad/.run_scripts/run.sh; '
+            'if [ "$__zk_outer_status" -ne 0 ]; then '
+            'printf "\\n__ZK_AGENT_EXIT_test__:%s\\n" "$__zk_outer_status"; '
+            'fi',
+        )
+
 
 if __name__ == '__main__':
     unittest.main()