deno-jupyter: support confirm and prompt in notebooks

Supports `confirm` and `prompt` with custom versions used when inside a
Jupyter Notebook with Deno kernel.

The desired behavior (per python reference and docs):
* confirm or prompt will trigger kernel to request the frontend to get
  user's input and return it to backend for processing

We accomplish this by creating custom versions of confirm and prompt
that call into an op_jupyter_input rust function with access to the
stdin_socket.

`confirm` and `prompt` are instantiated in the jupyter specific TS
interface, so they only override the standard functions in jupyter
context.

Jupyter requires us to clone zmq_identities for this "input_request"
message as documented in comments:

```
* Using with_identities() because of jupyter client docs instruction
* Requires cloning identities per :
* https://jupyter-client.readthedocs.io/en/latest/messaging.html#messages-on-the-stdin-router-dealer-channel
* The stdin socket of the client is required to have the
*  same zmq IDENTITY as the client’s shell socket.
*  Because of this, the input_request must be sent with the same IDENTITY
*  routing prefix as the execute_reply in order for the frontend to receive the message.
```

cli/js/40_jupyter.js

@@ -337,7 +337,14 @@ async function formatInner(obj, raw) {
 internals.jupyter = { formatInner };
 
 function enableJupyter() {
-  const { op_jupyter_broadcast } = core.ops;
+  const { op_jupyter_broadcast, op_jupyter_input } = core.ops;
+
+  async function input(
+    prompt,
+    password,
+  ) {
+    return await op_jupyter_input(prompt, password);
+  }
 
   async function broadcast(
     msgType,
@@ -412,6 +419,25 @@ function enableJupyter() {
     return;
   }
 
+  // Override confirm and prompt because they depend on a tty
+  // and in the Deno.jupyter environment that doesn't exist.
+  async function confirm(message = "Confirm") {
+    const answer = await input(`${message} [y/N] `, false)
+    return answer === "Y" || answer === "y";
+  }
+
+  async function prompt(message = "Prompt", defaultValue = "", { password = false } = {}) {
+    const answer = await input(`${message} [${defaultValue}] `, password)
+
+    if(answer === "") {
+      return defaultValue;
+    }
+
+    return answer;
+  }
+
+  globalThis.confirm = confirm;
+  globalThis.prompt = prompt;
   globalThis.Deno.jupyter = {
     broadcast,
     display,

cli/ops/jupyter.rs

@@ -10,13 +10,15 @@ use crate::tools::jupyter::server::StdioMsg;
 use deno_core::error::AnyError;
 use deno_core::op2;
 use deno_core::serde_json;
+use deno_core::serde_json::json;
 use deno_core::OpState;
 use tokio::sync::mpsc;
 use tokio::sync::Mutex;
 
 deno_core::extension!(deno_jupyter,
   ops = [
     op_jupyter_broadcast,
+    op_jupyter_input,
   ],
   options = {
     sender: mpsc::UnboundedSender<StdioMsg>,
@@ -30,6 +32,61 @@ deno_core::extension!(deno_jupyter,
   },
 );
 
+#[op2(async)]
+#[string]
+pub async fn op_jupyter_input(
+  state: Rc<RefCell<OpState>>,
+  #[string] prompt: String,
+  #[serde] is_password: serde_json::Value,
+) -> Result<Option<String>, AnyError> {
+  let (_iopub_socket, last_execution_request, stdin_socket) = {
+    let s = state.borrow();
+
+    (
+      s.borrow::<Arc<Mutex<Connection<zeromq::PubSocket>>>>()
+        .clone(),
+      s.borrow::<Rc<RefCell<Option<JupyterMessage>>>>().clone(),
+      s.borrow::<Arc<Mutex<Connection<zeromq::RouterSocket>>>>()
+        .clone(),
+    )
+  };
+
+  let mut stdin = stdin_socket.lock().await;
+
+  let maybe_last_request = last_execution_request.borrow().clone();
+  if let Some(last_request) = maybe_last_request {
+    if !last_request.allow_stdin() {
+      return Ok(None);
+    }
+
+    /*
+     * Using with_identities() because of jupyter client docs instruction
+     * Requires cloning identities per :
+     * https://jupyter-client.readthedocs.io/en/latest/messaging.html#messages-on-the-stdin-router-dealer-channel
+     * The stdin socket of the client is required to have the
+     *  same zmq IDENTITY as the client’s shell socket.
+     *  Because of this, the input_request must be sent with the same IDENTITY
+     *  routing prefix as the execute_reply in order for the frontend to receive the message.
+     * """
+     */
+    last_request
+      .new_message("input_request")
+      .with_identities(&last_request)
+      .with_content(json!({
+        "prompt": prompt,
+        "password": is_password,
+      }))
+      .send(&mut *stdin)
+      .await?;
+
+    let response = JupyterMessage::read(&mut *stdin).await?;
+
+    return Ok(Some(response.value().to_string()));
+  }
+
+  Ok(None)
+}
+
 #[op2(async)]
 pub async fn op_jupyter_broadcast(
   state: Rc<RefCell<OpState>>,

cli/tools/jupyter/jupyter_msg.rs

@@ -180,6 +180,14 @@ impl JupyterMessage {
     self.content["comm_id"].as_str().unwrap_or("")
   }
 
+  pub(crate) fn allow_stdin(&self) -> bool {
+    self.content["allow_stdin"].as_bool().unwrap_or(false)
+  }
+
+  pub(crate) fn value(&self) -> &str {
+    self.content["value"].as_str().unwrap_or("")
+  }
+
   // Creates a new child message of this message. ZMQ identities are not transferred.
   pub(crate) fn new_message(&self, msg_type: &str) -> JupyterMessage {
     let mut header = self.header.clone();
@@ -235,6 +243,14 @@ impl JupyterMessage {
     self
   }
 
+  pub(crate) fn with_identities(
+    mut self,
+    msg: &JupyterMessage,
+  ) -> JupyterMessage {
+    self.zmq_identities = msg.zmq_identities.clone();
+    self
+  }
+
   pub(crate) async fn send<S: zeromq::SocketSend>(
     &self,
     connection: &mut Connection<S>,

cli/tools/jupyter/server.rs

@@ -25,6 +25,7 @@ use super::jupyter_msg::Connection;
 use super::jupyter_msg::JupyterMessage;
 use super::ConnectionSpec;
 
+
 pub enum StdioMsg {
   Stdout(String),
   Stderr(String),
@@ -51,19 +52,21 @@ impl JupyterServer {
       bind_socket::<zeromq::RouterSocket>(&spec, spec.shell_port).await?;
     let control_socket =
       bind_socket::<zeromq::RouterSocket>(&spec, spec.control_port).await?;
-    let _stdin_socket =
+    let stdin_socket =
       bind_socket::<zeromq::RouterSocket>(&spec, spec.stdin_port).await?;
     let iopub_socket =
       bind_socket::<zeromq::PubSocket>(&spec, spec.iopub_port).await?;
     let iopub_socket = Arc::new(Mutex::new(iopub_socket));
+    let stdin_socket = Arc::new(Mutex::new(stdin_socket));
     let last_execution_request = Rc::new(RefCell::new(None));
 
-    // Store `iopub_socket` in the op state so it's accessible to the runtime API.
+    // Store `iopub_socket` and `stdin_socket` in the op state for access to the runtime API.
     {
       let op_state_rc = repl_session.worker.js_runtime.op_state();
       let mut op_state = op_state_rc.borrow_mut();
       op_state.put(iopub_socket.clone());
       op_state.put(last_execution_request.clone());
+      op_state.put(stdin_socket.clone());
     }
 
     let cancel_handle = CancelHandle::new_rc();

runtime/js/99_main.js

@@ -591,6 +591,7 @@ const NOT_IMPORTED_OPS = [
 
   // Related to `Deno.jupyter` API
   "op_jupyter_broadcast",
+  "op_jupyter_input",
 
   // Related to `Deno.test()` API
   "op_test_event_step_result_failed",