host: sausage run with the userland bundle, real shell and services (T13)

- userland/bundle.nu builds the userland and assembles target/userland/bin
  (bin names, tools.txt aliases, brush as sh and bash).
- run and sh take --userland (default SAUSAGE_USERLAND); /bin is read-only.
- The console's bash tool runs through the process table with the
  conversation id; large output spills to the world's /tmp.
- sausage run starts the services supervisor.

P1 exit session (openai/gpt-6.1-sol, fresh world): write, read, edit, git
init/add/commit/diff/log, grep -rn, rg, find and a sed|sort|uniq|wc pipeline.

host/src/harness/environment.rs

@@ -27,11 +27,12 @@ pub async fn op_file(
                 let size = file.stat().await?.size;
                 Ok(json!(file.read(0, size as usize).await?))
             }
-            "write" => {
-                let file = root.open(&path, Open { follow: true, write: true, create: true, truncate: true, ..Default::default() }).await?;
+            "write" | "append" => {
+                let file = root.open(&path, Open { follow: true, write: true, create: true, truncate: kind == "write", ..Default::default() }).await?;
                 let mut offset = 0;
+                let start = if kind == "append" { file.stat().await?.size } else { 0 };
                 while offset < bytes.len() {
-                    let count = file.write(offset as u64, bytes[offset..].to_vec()).await?;
+                    let count = file.write(start + offset as u64, bytes[offset..].to_vec()).await?;
                     if count == 0 { return Err(std::io::ErrorKind::WriteZero.into()); }
                     offset += count;
                 }
@@ -75,6 +76,7 @@ pub fn op_exec(
     #[serde] command: serde_json::Value,
     #[string] cwd: String,
     timeout: Option<f64>,
+    #[string] conversation: Option<String>,
 ) -> Result<u32, JsErrorBox> {
     let exec = state.borrow::<Environment>().exec.clone();
     let command = if let Some(s) = command.as_str() {
@@ -86,7 +88,7 @@ pub fn op_exec(
     };
     let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
     let task = tokio::spawn(async move {
-        let result = exec.exec(command, Options { cwd, timeout: timeout.map(std::time::Duration::from_secs_f64), ..Default::default() }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await;
+        let result = exec.exec(command, Options { cwd, conversation, timeout: timeout.map(std::time::Duration::from_secs_f64), ..Default::default() }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await;
         let _ = sender.send(match result {
             Ok(exit) => json!({"exitCode": exit}),
             Err(e) => json!({"error": e.to_string()}),

host/src/harness/mod.rs

@@ -99,10 +99,10 @@ pub async fn run_with_exec(world: World, exec: Arc<dyn Exec>) -> anyhow::Result<
     .map_err(|error: anyhow::Error| {
         let mut text = error.to_string();
         for key in ["PI_PROXY_BASE_URL", "PI_PROXY_API_KEY"] {
-            if let Ok(value) = std::env::var(key) {
-                if !value.is_empty() {
-                    text = text.replace(&value, "[redacted]");
-                }
+            if let Ok(value) = std::env::var(key)
+                && !value.is_empty()
+            {
+                text = text.replace(&value, "[redacted]");
             }
         }
         anyhow::anyhow!(text)
@@ -193,20 +193,22 @@ pub async fn run_module(world: World, exec: Arc<dyn Exec>, source: &str) -> anyh
     Ok(())
 }
 
-struct Unavailable;
-#[async_trait::async_trait]
-impl Exec for Unavailable {
-    async fn exec(
-        &self,
-        _: crate::exec::Command,
-        _: crate::exec::Options,
-        _: &mut (dyn for<'chunk> FnMut(crate::exec::Stream, &'chunk [u8]) + Send),
-    ) -> anyhow::Result<i32> {
-        anyhow::bail!("process execution is not wired yet")
-    }
-}
-
 /// Open a world and resume its durable root conversation.
 pub async fn run(path: &Path) -> anyhow::Result<()> {
-    run_with_exec(World::open(path)?, Arc::new(Unavailable)).await
+    run_with_userland(path, None).await
+}
+
+/// Start the console with the shared process table and an optional userland bundle.
+pub async fn run_with_userland(path: &Path, userland: Option<&Path>) -> anyhow::Result<()> {
+    let world = World::open_with_userland(path, userland)?;
+    let runtime = crate::process::Runtime::new(World {
+        root: world.root.clone(),
+        host_dir: world.host_dir.clone(),
+        processes: world.processes.clone(),
+    })?;
+    let services = crate::process::supervise(runtime.clone());
+    let result = run_with_exec(world, Arc::new(runtime)).await;
+    services.abort();
+    let _ = services.await;
+    result
 }

host/src/main.rs

@@ -12,21 +12,61 @@ struct Cli {
 #[derive(Subcommand)]
 enum Command {
     /// Start a world: the harness, its services and a console.
-    Run { world: PathBuf },
+    Run {
+        world: PathBuf,
+        #[arg(long)]
+        userland: Option<PathBuf>,
+    },
     /// Run a shell in a world. With `-c`, run one command and exit with its status.
     Sh {
         world: PathBuf,
         #[arg(short = 'c')]
         command: Option<String>,
+        #[arg(long)]
+        userland: Option<PathBuf>,
     },
 }
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     match Cli::parse().command {
-        Command::Run { world } => sausage::harness::run(&world).await,
-        Command::Sh { world, command } => {
-            let status = sausage::process::sh(&world, command).await?;
+        Command::Run { world, userland } => {
+            let bundle =
+                userland.or_else(|| std::env::var_os("SAUSAGE_USERLAND").map(PathBuf::from));
+            sausage::harness::run_with_userland(&world, bundle.as_deref()).await
+        }
+        Command::Sh {
+            world,
+            command,
+            userland,
+        } => {
+            use sausage::exec::{Command, Exec, Options, Stream};
+            let bundle =
+                userland.or_else(|| std::env::var_os("SAUSAGE_USERLAND").map(PathBuf::from));
+            let world = sausage::world::World::open_with_userland(&world, bundle.as_deref())?;
+            let runtime = sausage::process::Runtime::new(world)?;
+            let script =
+                command.ok_or_else(|| anyhow::anyhow!("interactive shell is not implemented"))?;
+            let status = runtime
+                .exec(
+                    Command::Shell(script),
+                    Options {
+                        cwd: "/".into(),
+                        ..Default::default()
+                    },
+                    &mut |stream, bytes| {
+                        use std::io::Write;
+                        match stream {
+                            Stream::Stdout => {
+                                let _ = std::io::stdout().write_all(bytes);
+                            }
+                            Stream::Stderr => {
+                                let _ = std::io::stderr().write_all(bytes);
+                            }
+                        }
+                    },
+                )
+                .await?;
             std::process::exit(status);
         }
     }