Read and change a machine's files from the backend

The first half of EXPLORER.md: server/src/files.rs, which lists a
directory, reads a file, writes one, and creates a file or a directory on
whichever machine a setup names.

Each operation is one small POSIX script run through `Transport`, the way
the import listing and the usage fetch already ask a machine a question,
so the local and the ssh case are one implementation rather than two that
drift. The path crosses as a positional argument and never as script
text; `PATH_PRELUDE` is the one line that gives a leading `~` its
meaning, because a shell expands a tilde in text and not in an argument,
and it is the far machine's home that has to answer.

A read has four answers -- text, binary, tooBig, or the machine's own
error -- because a binary file drawn as text and a big one cut off
silently are both wrong in ways the reader cannot see. A write carries
the sha256 the read reported and is refused with a 409 when the file has
moved on, which is what happens whenever an agent is editing the file
somebody is reading.

`Transport::capture_with_input` is the one description of "run this
there, with this on stdin", and `ship_attachment` moves onto it rather
than assembling a second ssh invocation of its own. It is also the only
capture that hands back the exit status, which is how the write says
"this is not the file you read" without that answer looking like a
failure.

Exercised on both transports against the sandbox -- ssh to this VM with a
throwaway key, since the quoting and the stdin path are what that proves
-- including a filename with an apostrophe, one with a tab, an unreadable
file, a binary one, one over the limit, and the 409.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

server/src/session/transport.rs

@@ -60,6 +60,11 @@ impl Launch {
 pub enum Streams {
     /// Pipes owned by this server; the child is killed when they drop.
     Piped,
+    /// The same, except that stdin is already open on something this
+    /// server holds -- the file being copied to another machine. Bytes
+    /// this process has in memory do not need this: [`Streams::Piped`]
+    /// gives a pipe to write them into as the child reads.
+    PipedFrom(Stdio),
     /// Files -- and, for stdin, a fifo the child itself holds open so it
     /// never reads EOF -- that outlast this process.
     Detached {
@@ -117,6 +122,13 @@ impl Transport {
                     .stderr(Stdio::piped())
                     .kill_on_drop(true);
             }
+            Streams::PipedFrom(stdin) => {
+                command
+                    .stdin(stdin)
+                    .stdout(Stdio::piped())
+                    .stderr(Stdio::piped())
+                    .kill_on_drop(true);
+            }
             Streams::Detached {
                 stdin,
                 stdout,
@@ -174,6 +186,57 @@ impl Transport {
         Ok(String::from_utf8_lossy(&output.stdout).into_owned())
     }
 
+    /// Runs `launch` with `input` on its stdin and reports everything it
+    /// produced -- stdout as bytes, stderr as text, and the exit status.
+    ///
+    /// The one description of "run this there, with this on stdin", so
+    /// that shipping an attachment and writing a file through the explorer
+    /// are the same operation rather than two. It is also the only capture
+    /// that hands back the **status**: a script can then answer with an
+    /// exit code the caller distinguishes (the explorer's write says
+    /// `exit 3` for "this file is not the one you read"), which
+    /// [`Transport::capture`] cannot express because it turns every
+    /// failure into one error.
+    ///
+    /// Bytes rather than a `String`, because a file's contents are not
+    /// text until something has checked, and lossy decoding would replace
+    /// the evidence that they are not.
+    ///
+    /// `Err` means the process could not be started at all; a process that
+    /// ran and failed is a [`Captured`] with a status saying so.
+    pub async fn capture_with_input(&self, launch: &Launch, input: Input) -> Result<Captured> {
+        let (streams, to_write) = match input {
+            Input::None => (Streams::Piped, None),
+            Input::Bytes(bytes) => (Streams::Piped, Some(bytes)),
+            Input::File(file) => (Streams::PipedFrom(file.into()), None),
+        };
+        let mut child = self.spawn(launch, streams)?;
+        if let Some(bytes) = to_write {
+            // Written from a task rather than before the wait, because the
+            // child may not read all of it -- the write script exits
+            // without reading when the file has changed underneath -- and
+            // a caller blocked on filling a pipe nobody is draining would
+            // deadlock instead of getting that answer. The broken pipe is
+            // the expected end of this write, so it is dropped: what
+            // happened is the exit status below.
+            let mut stdin = child.stdin.take().context("the child has no stdin")?;
+            tokio::spawn(async move {
+                use tokio::io::AsyncWriteExt;
+                let _ = stdin.write_all(&bytes).await;
+                let _ = stdin.shutdown().await;
+            });
+        }
+        let output = child
+            .wait_with_output()
+            .await
+            .with_context(|| format!("waiting for \"{}\" {}", launch.program, self.describe()))?;
+        Ok(Captured {
+            status: output.status,
+            stdout: output.stdout,
+            stderr: String::from_utf8_lossy(&output.stderr).trim().to_string(),
+        })
+    }
+
     /// How to say where this runs, for a log line a person reads.
     pub fn describe(&self) -> String {
         match self {
@@ -182,3 +245,41 @@ impl Transport {
         }
     }
 }
+
+/// What a command is given on its standard input.
+///
+/// Three cases rather than an `Option<Stdio>` because they are three
+/// genuinely different arrangements and only this knows which: nothing to
+/// say, bytes this process is holding, or a file it has open. The last one
+/// is how a several-hundred-megabyte attachment reaches another machine
+/// without passing through this server's memory.
+pub enum Input {
+    None,
+    Bytes(Vec<u8>),
+    File(std::fs::File),
+}
+
+/// Everything a finished command produced, including the status.
+pub struct Captured {
+    pub status: std::process::ExitStatus,
+    pub stdout: Vec<u8>,
+    /// Trimmed, and what a failure is reported as: ssh's own refusals and
+    /// a tool's own message about the file it could not open are both the
+    /// useful half of why something did not work, and both are written to
+    /// name the thing.
+    pub stderr: String,
+}
+
+impl Captured {
+    /// The stdout of a command that succeeded, or the machine's own words.
+    pub fn ok(self) -> Result<Vec<u8>> {
+        if self.status.success() {
+            return Ok(self.stdout);
+        }
+        anyhow::bail!(if self.stderr.is_empty() {
+            format!("it failed with no explanation ({})", self.status)
+        } else {
+            self.stderr
+        })
+    }
+}