Merge v3-t32: pi-durable patch (P3-a), headless harness with channel input (P3-b) (T32)

# Conflicts:
#	host/src/harness/mod.rs

host/src/api/server.rs

@@ -125,12 +125,7 @@ pub async fn run_with_userland(
         tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()?
-            .block_on(crate::harness::run_module_with_credentials(
-                world,
-                Arc::new(runtime),
-                cred,
-                include_str!("../../js/bundle.js"),
-            ))
+            .block_on(crate::harness::run_console(world, Arc::new(runtime), cred))
     })
     .await
     .map_err(anyhow::Error::from)

host/src/harness/environment.rs

@@ -129,19 +129,6 @@ pub async fn op_exec_next(
     Ok(value)
 }
 
-#[op2]
-#[string]
-pub async fn op_input() -> Result<Option<String>, JsErrorBox> {
-    use tokio::io::AsyncBufReadExt;
-    let mut reader = tokio::io::BufReader::with_capacity(1, tokio::io::stdin());
-    let mut line = String::new();
-    let n = reader
-        .read_line(&mut line)
-        .await
-        .map_err(|e| JsErrorBox::generic(e.to_string()))?;
-    Ok((n != 0).then_some(line))
-}
-
 #[op2(fast)]
 pub fn op_output(#[string] text: &str) {
     print!("{text}");

host/src/harness/input.rs

@@ -0,0 +1,74 @@
+//! Channel input for a headless harness. The console and stop signals are clients.
+use deno_core::{OpState, op2};
+use deno_error::JsErrorBox;
+use serde::Serialize;
+use serde_json::Value;
+use std::{cell::RefCell, rc::Rc};
+use tokio::sync::mpsc;
+
+#[derive(Serialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub enum Input {
+    Submit {
+        conversation: u64,
+        content: Value,
+        #[serde(rename = "requestId")]
+        request_id: String,
+    },
+    Stop,
+}
+
+pub type Sender = mpsc::UnboundedSender<Input>;
+pub type Receiver = mpsc::UnboundedReceiver<Input>;
+
+pub fn channel() -> (Sender, Receiver) {
+    mpsc::unbounded_channel()
+}
+
+pub(super) struct Inbox(pub tokio::sync::Mutex<Receiver>);
+
+#[op2]
+#[serde]
+pub async fn op_input(state: Rc<RefCell<OpState>>) -> Result<serde_json::Value, JsErrorBox> {
+    let inbox = state.borrow().borrow::<Rc<Inbox>>().clone();
+    serde_json::to_value(inbox.0.lock().await.recv().await)
+        .map_err(|error| JsErrorBox::generic(error.to_string()))
+}
+
+pub(super) fn console(sender: Sender) {
+    std::thread::spawn(move || {
+        use std::io::BufRead;
+        for line in std::io::stdin().lock().lines() {
+            let Ok(line) = line else { break };
+            if line.trim().is_empty() {
+                continue;
+            }
+            if sender
+                .send(Input::Submit {
+                    conversation: 1,
+                    content: Value::String(line),
+                    request_id: uuid::Uuid::new_v4().to_string(),
+                })
+                .is_err()
+            {
+                break;
+            }
+        }
+        // EOF closes only this client. The signal client keeps the harness alive.
+    });
+}
+
+pub(super) fn stop_signal(
+    sender: Sender,
+) -> std::io::Result<impl std::future::Future<Output = ()> + Send> {
+    // Register before V8 starts, so a stop cannot arrive before the listener.
+    let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
+    let mut interrupt = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
+    Ok(async move {
+        tokio::select! {
+            _ = interrupt.recv() => {}
+            _ = terminate.recv() => {}
+        }
+        let _ = sender.send(Input::Stop);
+    })
+}

host/src/harness/mod.rs

@@ -1,6 +1,7 @@
 //! Durable V8 harness, host-side SQLite storage, and world tool operations.
 mod credentials;
 mod environment;
+pub mod input;
 use crate::{exec::Exec, world::World};
 use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
 use deno_error::JsErrorBox;
@@ -85,7 +86,7 @@ deno_core::extension!(
         environment::op_file,
         environment::op_exec,
         environment::op_exec_next,
-        environment::op_input,
+        input::op_input,
         environment::op_output
     ]
 );
@@ -105,12 +106,7 @@ async fn run_with_shared_credentials(
         tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()?
-            .block_on(run_module_with_credentials(
-                world,
-                exec,
-                cred,
-                include_str!("../../js/bundle.js"),
-            ))
+            .block_on(run_console(world, exec, cred))
     })
     .await?
     .map_err(|error: anyhow::Error| {
@@ -139,6 +135,19 @@ pub async fn run_module_with_credentials(
     exec: Arc<dyn Exec>,
     cred: Arc<crate::cred::store::Cred>,
     source: &str,
+) -> anyhow::Result<()> {
+    let (_, receiver) = input::channel();
+    run_module_with_input(world, exec, cred, receiver, source).await
+}
+
+/// Run without the console. Submissions and stop requests arrive through the channel.
+/// The caller must use a current-thread Tokio runtime.
+pub async fn run_module_with_input(
+    world: World,
+    exec: Arc<dyn Exec>,
+    cred: Arc<crate::cred::store::Cred>,
+    receiver: input::Receiver,
+    source: &str,
 ) -> anyhow::Result<()> {
     let proxy = credentials::proxy(&cred)?;
     let mut runtime = JsRuntime::new(RuntimeOptions {
@@ -196,6 +205,12 @@ pub async fn run_module_with_credentials(
         .borrow_mut()
         .put(credentials::Notifications::new(&cred));
     runtime.op_state().borrow_mut().put(cred);
+    runtime
+        .op_state()
+        .borrow_mut()
+        .put(std::rc::Rc::new(input::Inbox(tokio::sync::Mutex::new(
+            receiver,
+        ))));
     runtime
         .op_state()
         .borrow_mut()
@@ -237,6 +252,25 @@ pub async fn run_module_with_credentials(
     })
 }
 
+/// Run the headless harness with console input and signal clients.
+/// The caller must use a current-thread Tokio runtime.
+pub async fn run_console(
+    world: World,
+    exec: Arc<dyn Exec>,
+    cred: Arc<crate::cred::store::Cred>,
+) -> anyhow::Result<()> {
+    let (sender, receiver) = input::channel();
+    input::console(sender.clone());
+    let signals = tokio::spawn(input::stop_signal(sender)?);
+    let source = concat!(
+        "globalThis.consoleEnabled = true;\n",
+        include_str!("../../js/bundle.js")
+    );
+    let result = run_module_with_input(world, exec, cred, receiver, source).await;
+    signals.abort();
+    result
+}
+
 /// Open a world and resume its durable root conversation.
 pub async fn run(path: &Path) -> anyhow::Result<()> {
     run_with_userland(path, None).await