.gitea/workflows/agents.yml

@@ -0,0 +1,49 @@
+name: Agents
+run-name: >-
+  AI bot at
+  #${{ gitea.event.issue.number || gitea.event.pull_request.number }}
+
+on:
+  issues:
+    types: [opened]
+  pull_request:
+    types: [opened, review_requested]
+  issue_comment:
+    types: [created]
+  pull_request_review_comment:
+    types: [created]
+
+# New human events replace stale runs for the same work item. AI-bot comments
+# get unique groups so their skipped workflows cannot cancel an active login.
+concurrency:
+  group: >-
+    agents-${{ gitea.event.sender.login == 'bot' && gitea.run_id ||
+    gitea.event.issue.number || gitea.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  bot:
+    # Nothing the bot account posts can start a run, so it cannot trigger itself.
+    if: >
+      gitea.event.sender.login != 'bot' &&
+      ((gitea.event_name == 'issues' &&
+      contains(gitea.event.issue.body, '@bot')) ||
+      (gitea.event_name == 'pull_request' &&
+      ((gitea.event.action == 'opened' &&
+      contains(gitea.event.pull_request.body, '@bot')) ||
+      gitea.event.requested_reviewer.login == 'bot')) ||
+      (gitea.event_name == 'issue_comment' &&
+      contains(gitea.event.comment.body, '@bot')) ||
+      (gitea.event_name == 'pull_request_review_comment' &&
+      contains(gitea.event.review.content, '@bot')))
+    runs-on: self-hosted
+    container:
+      volumes:
+        - bot-home:/root
+    steps:
+      - name: Run AI bot
+        # Intentionally follow protected main so centrally managed profiles and
+        # quota fallbacks update without per-repository credential changes.
+        uses: https://git.capsulizers.com/commons/bot-agents@main
+        with:
+          master-key: ${{ secrets.BOT_AGENTS_MASTER_KEY }}

.gitea/workflows/ci.yml

@@ -0,0 +1,51 @@
+name: CI
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+
+concurrency:
+  group: ${{ gitea.workflow }}-${{ gitea.ref }}
+  cancel-in-progress: true
+
+jobs:
+  checks:
+    runs-on: self-hosted
+    container:
+      volumes:
+        - memona-plugin-target:/cargo-target
+    env:
+      CARGO_BUILD_JOBS: 2
+      CARGO_TARGET_DIR: /cargo-target/serialized
+      CARGO_HOME: /cargo-target/cargo-home
+      RUSTUP_HOME: /cargo-target/rustup-home
+      RUSTUP_TOOLCHAIN: stable
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Install system packages
+        run: >-
+          apt-get update && apt-get install -y --no-install-recommends
+          build-essential ca-certificates curl util-linux
+
+      # The same lock protects the shared toolchain and every Cargo command.
+      # Keep its target and lock paths in sync with Memona's plugin checks.
+      - name: Check and build plugin
+        run: |
+          flock --no-fork "$CARGO_TARGET_DIR.lock" sh -ec '
+            export PATH="$CARGO_HOME/bin:$PATH"
+            if [ ! -x "$CARGO_HOME/bin/rustup" ]; then
+              mkdir -p "$CARGO_HOME"
+              curl --proto =https --tlsv1.2 -fsSL https://sh.rustup.rs -o "$CARGO_HOME/rustup-init.sh"
+              sh "$CARGO_HOME/rustup-init.sh" -y --no-modify-path --default-toolchain none
+              rm "$CARGO_HOME/rustup-init.sh"
+            fi
+            rustup toolchain install stable --profile minimal --component clippy,rustfmt --target wasm32-wasip2 --no-self-update
+            cargo fmt --check
+            cargo clippy --locked --all-targets -- -D warnings
+            cargo test --locked --lib
+            cargo build --locked --release --target wasm32-wasip2
+          '

Cargo.lock

@@ -14,6 +14,12 @@ version = "2.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
 
+[[package]]
+name = "cfg-if"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
+
 [[package]]
 name = "equivalent"
 version = "1.0.2"
@@ -26,6 +32,18 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
 
+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi",
+ "wasip2",
+]
+
 [[package]]
 name = "hashbrown"
 version = "0.17.1"
@@ -71,6 +89,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
 
+[[package]]
+name = "libc"
+version = "0.2.189"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+
 [[package]]
 name = "log"
 version = "0.4.34"
@@ -96,18 +120,26 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
 
 [[package]]
 name = "memona-plugin-ftp"
-version = "0.1.2"
+version = "0.1.3"
 dependencies = [
+ "getrandom",
  "memona-plugin-sdk",
+ "tokio",
 ]
 
 [[package]]
 name = "memona-plugin-sdk"
 version = "0.1.0"
 dependencies = [
- "wit-bindgen",
+ "wit-bindgen 0.62.0",
 ]
 
+[[package]]
+name = "pin-project-lite"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+
 [[package]]
 name = "prettyplease"
 version = "0.3.0"
@@ -136,6 +168,12 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
 [[package]]
 name = "semver"
 version = "1.0.28"
@@ -195,12 +233,42 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "tokio"
+version = "1.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
+dependencies = [
+ "pin-project-lite",
+ "tokio-macros",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "2.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
 
+[[package]]
+name = "wasip2"
+version = "1.0.1+wasi-0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7"
+dependencies = [
+ "wit-bindgen 0.46.0",
+]
+
 [[package]]
 name = "wasm-encoder"
 version = "0.259.0"
@@ -235,6 +303,12 @@ dependencies = [
  "semver",
 ]
 
+[[package]]
+name = "wit-bindgen"
+version = "0.46.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59"
+
 [[package]]
 name = "wit-bindgen"
 version = "0.62.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "memona-plugin-ftp"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2024"
 license = "MIT"
 
@@ -8,8 +8,12 @@ license = "MIT"
 crate-type = ["cdylib", "rlib"]
 
 [dependencies]
+getrandom = "0.3"
 memona-plugin-sdk = { path = "sdk" }
 
+[dev-dependencies]
+tokio = { version = "1", features = ["macros", "rt"] }
+
 [lints]
 workspace = true
 

README.md

@@ -8,12 +8,22 @@ The plugin is a WebAssembly component built against the Memona filesystem
 contract, `memona:filesystem@0.1.0`. It uses only the host's sockets; it has no
 ambient filesystem, environment or network access of its own.
 
+This package requires a Memona version that supports plugin manifest schema 2.
+
 ## What it registers
 
 One provider, `ftp`, for the `ftp` and `ftps` schemes. A connection is
 configured with a server name, an optional port, a TLS toggle for FTPS, and an
 optional username and password. An empty username signs in anonymously.
 
+With the compatible schema 2 Memona version required by this package, public,
+LAN, VPN, link-local and localhost addresses work directly from the server
+name; that API has no separate connection network grant.
+
+The TLS toggle currently starts TLS on the control connection only. Protected
+data connections and explicit `AUTH TLS` negotiation are not implemented, so
+the plugin does not yet provide a complete FTPS connection.
+
 Listing and stat need a server that answers `MLSD` and `MLST`, the
 machine-readable listing commands. A server without them reports the folder as
 unsupported rather than guessing at the older, human-formatted listing or
@@ -23,15 +33,13 @@ so a conditional write is reported as unsupported rather than simulated.
 ## Passive mode and the port range
 
 Transfers are passive: the plugin asks the server for a data port with `EPSV`
-or `PASV` and connects out to it. The server picks a fresh port for every
+and connects out to it; `PASV` fallback is not implemented. The server picks a fresh port for every
 transfer and announces it only as the transfer begins, so the range that has to
 be reachable is wide.
 
-`manifest.json` declares `networkPorts` as 21 and 1024-65535, and `networkHosts`
+The package declares socket ports 21 and 1024-65535, which Memona applies
-as `*` because a server can be anywhere. Declaring is not granting: a
+automatically to each connection. The control port and passive data ports
-connection reaches only the host and ports the person grants it, and a grant
+must be within these ranges.
-never accepts a wildcard host. Grant the control port and the span the server
-uses for its data channel, usually `21` and `1024-65535`.
 
 ## Build
 

manifest.json

@@ -1,13 +1,12 @@
 {
-  "schemaVersion": 1,
+  "schemaVersion": 2,
   "id": "memona/ftp",
   "name": "FTP",
-  "version": "0.1.2",
+  "version": "0.1.3",
   "apiVersion": "0.1.0",
   "description": "Open an FTP server as a folder, and let agents work in it.",
   "license": "MIT",
   "entry": "plugin.wasm",
-  "networkHosts": ["*"],
   "networkPorts": [
     { "from": 21, "to": 21 },
     { "from": 1024, "to": 65535 }

sdk/wit/filesystem.wit

@@ -35,11 +35,11 @@ interface host {
         /// `respond`. An empty result means the body is complete.
         pull: func(limit: u32) -> result<list<u8>, host-error>;
     }
-    /// Begins a request to a host both declared by the package and granted on
+    /// Begins a request to any HTTP or HTTPS host, including LAN and loopback
-    /// the connection. Each redirect and DNS result is checked, and DNS is
+    /// addresses. This API has no per-connection network grant. Each redirect
-    /// pinned. Private/LAN addresses additionally require the private-network
+    /// and DNS result is checked, and DNS is pinned. Cross-origin redirects
-    /// grant. Cross-origin redirects remove request headers; HTTPS cannot
+    /// remove request headers; HTTPS cannot downgrade. The host bounds total
-    /// downgrade. The host bounds total bytes per exchange, not buffer size.
+    /// bytes per exchange, not buffer size.
     http: func(request: request) -> result<exchange, host-error>;
 
     /// One open TCP connection. Dropping the handle closes the socket, and the
@@ -51,13 +51,13 @@ interface host {
         /// Writes at most 1,048,576 bytes and returns how many were accepted.
         write: func(data: list<u8>) -> result<u32, host-error>;
     }
-    /// Opens a socket to a host and port both declared by the package and
+    /// Opens a socket to any host, including LAN and loopback addresses, on a
-    /// granted on the connection. The host resolves and pins DNS, applies the
+    /// package-declared port. This API has no per-connection network grant.
-    /// same private-address rule as `http`, and starts TLS when asked. Ports
+    /// The host resolves and pins DNS and starts TLS when asked. Ports
-    /// negotiated at runtime, such as an FTP data channel, still need a grant.
+    /// negotiated at runtime, such as an FTP data channel, must be declared.
     connect: func(host: string, port: u16, tls: bool) -> result<tcp, host-error>;
 
-    /// One datagram socket connected to a single granted peer.
+    /// One datagram socket connected to a single peer.
     resource udp {
         /// Sends one datagram of at most 65,507 bytes.
         send: func(data: list<u8>) -> result<u32, host-error>;
@@ -65,8 +65,8 @@ interface host {
         /// means none arrived inside the host's wait, not that the peer left.
         receive: func(limit: u32) -> result<list<u8>, host-error>;
     }
-    /// Connects a datagram socket under the same declaration and grant rules
+    /// Connects a datagram socket under the same address and port rules as
-    /// as `connect`, for the few filesystem protocols that still need one.
+    /// `connect`, for the few filesystem protocols that still need one.
     bind: func(host: string, port: u16) -> result<udp, host-error>;
 
     /// A configuration value, typed as the package's manifest declared it.
@@ -81,7 +81,8 @@ interface host {
 
 /// Filesystem operations for one connection. Paths are absolute within its
 /// root, starting with `/`; they are never local OS paths or resource URIs.
-/// Dot segments, backslashes, percent escapes, colons, and controls are invalid.
+/// Colons and percent signs are literal; paths must never be URL-decoded.
+/// Dot segments, backslashes, and controls are invalid.
 /// Every invocation is bounded by host fuel, memory, time, and cancellation.
 interface provider {
     use host.{setting};

src/control.rs

@@ -1,6 +1,9 @@
 //! The FTP control channel: one socket, line replies, and passive transfers.
 
+#[cfg(test)]
+use crate::test_host as host;
 use memona_plugin_sdk::ErrorCode;
+#[cfg(not(test))]
 use memona_plugin_sdk::memona::filesystem::host;
 
 const MAX_LINE: usize = 8192;
@@ -64,7 +67,9 @@ fn names_a_conflict(text: &str) -> bool {
     .collect();
   words.contains(&"exists")
     || words.contains(&"nonempty")
-    || words.windows(2).any(|pair| matches!(pair, ["not" | "non", "empty"]))
+    || words
+      .windows(2)
+      .any(|pair| matches!(pair, ["not" | "non", "empty"]))
 }
 
 /// What the first reply to a transfer command means. A transfer answers with
@@ -86,6 +91,11 @@ pub struct Control {
   buffer: Vec<u8>,
 }
 
+struct DispatchError {
+  code: ErrorCode,
+  dispatched: bool,
+}
+
 impl Control {
   pub fn open(
     host_name: String,
@@ -150,27 +160,61 @@ impl Control {
     Err(ErrorCode::Unavailable)
   }
 
-  /// Sends one command. A newline in an argument would forge a second command,
+  /// Writes one complete command and records whether any of it reached the
-  /// so it is refused rather than escaped.
+  /// socket. A newline in an argument would forge a second command, so it is
-  pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
+  /// refused before dispatch rather than escaped.
+  fn dispatch(&mut self, command: &str) -> Result<(), DispatchError> {
     if command.contains(['\r', '\n']) || command.len() > MAX_LINE {
-      return Err(ErrorCode::InvalidInput);
+      return Err(DispatchError {
+        code: ErrorCode::InvalidInput,
+        dispatched: false,
+      });
     }
     let line = format!("{command}\r\n");
     let mut sent = 0;
     while sent < line.len() {
-      let written = self
+      let written =
+        self
           .socket
           .write(&line.as_bytes()[sent..])
-        .map_err(transport)? as usize;
+          .map_err(|error| DispatchError {
+            code: transport(error),
+            dispatched: sent > 0,
+          })? as usize;
       if written == 0 {
-        return Err(ErrorCode::Unavailable);
+        return Err(DispatchError {
+          code: ErrorCode::Unavailable,
+          dispatched: sent > 0,
+        });
       }
       sent += written;
     }
+    Ok(())
+  }
+
+  /// Sends one command and reads its reply.
+  pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
+    self.dispatch(command).map_err(|error| error.code)?;
     self.reply()
   }
 
+  /// Sends a command that changes remote state. Once any command byte was
+  /// dispatched, a write interruption or missing reply makes its outcome
+  /// unknowable even when the underlying transport reports cancellation.
+  pub fn send_mutation(
+    &mut self,
+    command: &str,
+  ) -> Result<(u16, String), ErrorCode> {
+    self.dispatch(command).map_err(|error| {
+      if error.dispatched {
+        ErrorCode::OutcomeUnknown
+      } else {
+        error.code
+      }
+    })?;
+    self.reply().map_err(|_| ErrorCode::OutcomeUnknown)
+  }
+
   pub fn expect(&mut self, command: &str) -> Result<String, ErrorCode> {
     let (code, text) = self.send(command)?;
     outcome(code)?;
@@ -255,7 +299,10 @@ mod tests {
     ));
     // The same two conditions as other servers phrase them.
     assert!(matches!(
-      refusal(550, "550 Cannot create a file when that file already exists."),
+      refusal(
+        550,
+        "550 Cannot create a file when that file already exists."
+      ),
       Err(ErrorCode::Conflict)
     ));
     assert!(matches!(

src/lib.rs

@@ -4,8 +4,14 @@ mod control;
 mod listing;
 mod transfer;
 
+#[cfg(test)]
+mod session_tests;
+#[cfg(test)]
+mod test_host;
+
 use control::{Control, outcome, preliminary, refusal, transport};
 use listing::{machine_info, page};
+#[cfg(not(test))]
 use memona_plugin_sdk::memona::filesystem::host;
 use memona_plugin_sdk::{
   Capabilities, ChangePage, EntryKind, ErrorCode, FileInfo, Guest, GuestReader,
@@ -14,14 +20,18 @@ use memona_plugin_sdk::{
 use std::cell::{Cell, RefCell};
 use std::collections::HashMap;
 use std::rc::Rc;
+#[cfg(test)]
+use test_host as host;
 use transfer::Buffered;
 
 const MAX_CHUNK: u32 = 1024 * 1024;
 /// One listing may be this large, and so may every listing kept together.
 const MAX_LISTING_BYTES: usize = 8 * 1024 * 1024;
+const MAX_STAGE_ATTEMPTS: usize = 64;
 
 thread_local! {
   static CONTROL: RefCell<Option<Control>> = const { RefCell::new(None) };
+  static SESSION: RefCell<Option<Session>> = const { RefCell::new(None) };
   /// Raw MLSD listings by path, so later pages of a folder are parsed from
   /// memory instead of downloaded and parsed whole again.
   static LISTINGS: RefCell<HashMap<String, Rc<[u8]>>> = RefCell::new(HashMap::new());
@@ -29,12 +39,38 @@ thread_local! {
 
 struct Ftp;
 
+#[derive(Clone)]
+struct Session {
+  host: String,
+  port: u16,
+  tls: bool,
+  user: String,
+  password: String,
+}
+
+impl Session {
+  fn connect(&self) -> Result<Control, ErrorCode> {
+    let mut control = Control::open(self.host.clone(), self.port, self.tls)?;
+    let (code, _) = control.send(&format!("USER {}", self.user))?;
+    if code == 331 {
+      let (code, _) = control.send(&format!("PASS {}", self.password))?;
+      outcome(code)?;
+    } else {
+      outcome(code)?;
+    }
+    // Binary throughout: text mode would rewrite line endings inside files.
+    control.expect("TYPE I")?;
+    Ok(control)
+  }
+}
+
 fn text(setting: Option<&Setting>) -> Option<String> {
   match setting {
-    Some(Setting::Text(value)) | Some(Setting::Secret(value)) => {
+    Some(Setting::Text(value)) => {
       let value = value.trim().to_owned();
       (!value.is_empty()).then_some(value)
     }
+    Some(Setting::Secret(value)) => Some(value.clone()),
     Some(Setting::Toggle(_)) | None => None,
   }
 }
@@ -49,12 +85,43 @@ fn route(target: &Locator) -> Result<String, ErrorCode> {
 fn with_control<T>(
   action: impl FnOnce(&mut Control) -> Result<T, ErrorCode>,
 ) -> Result<T, ErrorCode> {
-  CONTROL.with_borrow_mut(|slot| match slot.as_mut() {
+  CONTROL.with_borrow_mut(|slot| {
+    let result = match slot.as_mut() {
       Some(control) => action(control),
       None => Err(ErrorCode::Internal),
+    };
+    if matches!(
+      result,
+      Err(
+        ErrorCode::Unavailable
+          | ErrorCode::OutcomeUnknown
+          | ErrorCode::Internal
+      )
+    ) {
+      *slot = None;
+    }
+    result
   })
 }
 
+/// Checks a control channel only before a fresh filesystem operation. A lost
+/// idle session is replaced and logged in again. Nothing inside a transfer or
+/// mutation retries, because its missing reply may hide a completed change.
+fn ready() -> Result<(), ErrorCode> {
+  if !CONTROL.with_borrow(Option::is_none) {
+    match with_control(|control| control.expect("NOOP")) {
+      Ok(_) => return Ok(()),
+      Err(ErrorCode::Unavailable) => {}
+      Err(error) => return Err(error),
+    }
+  }
+  let session = SESSION
+    .with_borrow(Clone::clone)
+    .ok_or(ErrorCode::Internal)?;
+  CONTROL.set(Some(session.connect()?));
+  Ok(())
+}
+
 /// Starts a passive transfer and leaves the data socket open for the caller,
 /// who owes the server one more reply once that socket is closed: either the
 /// completion, read by `settle`, or the answer to an `abort`.
@@ -67,6 +134,17 @@ fn transfer(command: &str) -> Result<host::Tcp, ErrorCode> {
   })
 }
 
+/// Starts a passive upload. Losing the command reply is ambiguous because an
+/// FTP server may have created or truncated the destination on receipt.
+fn mutation_transfer(command: &str) -> Result<host::Tcp, ErrorCode> {
+  with_control(|control| {
+    let data = control.passive()?;
+    let (code, _) = control.send_mutation(command)?;
+    preliminary(code)?;
+    Ok(data)
+  })
+}
+
 /// Consumes the completion reply of a transfer whose data socket is closed.
 fn settle() -> Result<(), ErrorCode> {
   with_control(|control| {
@@ -75,6 +153,16 @@ fn settle() -> Result<(), ErrorCode> {
   })
 }
 
+/// Consumes an upload's completion reply. The data and command are already
+/// dispatched, so an interrupted reply cannot say whether the file committed.
+fn settle_mutation() -> Result<(), ErrorCode> {
+  let (code, _) = with_control(|control| {
+    control.reply().map_err(|_| ErrorCode::OutcomeUnknown)
+  })
+  .map_err(|_| ErrorCode::OutcomeUnknown)?;
+  outcome(code)
+}
+
 /// Ends a transfer early, after its data socket is closed.
 fn abort() -> Result<(), ErrorCode> {
   with_control(Control::abort)
@@ -167,20 +255,62 @@ impl Drop for FtpReader {
 
 /// Writes one file to a passive data socket.
 struct FtpWriter {
-  path: String,
+  target: String,
+  staged: RefCell<Option<String>>,
   socket: RefCell<Option<host::Tcp>>,
   written: Cell<u64>,
 }
 
+impl FtpWriter {
+  /// Removes the private staging file when possible. A missing stage means a
+  /// prior promotion or cleanup already completed.
+  fn cleanup_stage(&self) {
+    let Some(staged) = self.staged.borrow().clone() else {
+      return;
+    };
+    if ready().is_err() {
+      return;
+    }
+    let removed = with_control(|control| {
+      let (code, _) = control.send_mutation(&format!("DELE {staged}"))?;
+      outcome(code)
+    });
+    if matches!(removed, Ok(()) | Err(ErrorCode::NotFound)) {
+      self.staged.take();
+    }
+  }
+
+  /// A failed data write can leave both the remote file and its pending
+  /// completion reply in an unknown state. Close both channels, then make a
+  /// best-effort cleanup of the private stage over a fresh session.
+  fn discard_uncertain_upload(&self) -> ErrorCode {
+    drop(self.socket.take());
+    CONTROL.set(None);
+    self.cleanup_stage();
+    ErrorCode::OutcomeUnknown
+  }
+}
+
 impl GuestWriter for FtpWriter {
   fn push(&self, data: Vec<u8>) -> Result<(), ErrorCode> {
-    let socket = self.socket.borrow();
+    if self.socket.borrow().is_none() {
-    let socket = socket.as_ref().ok_or(ErrorCode::InvalidInput)?;
+      return Err(ErrorCode::InvalidInput);
+    }
     let mut sent = 0;
     while sent < data.len() {
-      let written = socket.write(&data[sent..]).map_err(transport)? as usize;
+      let result = {
+        let socket = self.socket.borrow();
+        let Some(socket) = socket.as_ref() else {
+          return Err(ErrorCode::InvalidInput);
+        };
+        socket.write(&data[sent..])
+      };
+      let written = match result {
+        Ok(written) => written as usize,
+        Err(_) => return Err(self.discard_uncertain_upload()),
+      };
       if written == 0 {
-        return Err(ErrorCode::Unavailable);
+        return Err(self.discard_uncertain_upload());
       }
       sent += written;
     }
@@ -188,10 +318,30 @@ impl GuestWriter for FtpWriter {
     Ok(())
   }
   /// Closing the data socket is what ends an FTP upload; only then does the
-  /// server answer, so the socket is given up before the reply is read.
+  /// server answer. A confirmed stage is promoted only after that reply, so an
+  /// interrupted transfer never exposes a partial destination.
   fn finish(&self) -> Result<Option<FileInfo>, ErrorCode> {
     drop(self.socket.take().ok_or(ErrorCode::InvalidInput)?);
-    settle()?;
+    if let Err(error) = settle_mutation() {
+      self.cleanup_stage();
+      return Err(error);
+    }
+    let staged = self
+      .staged
+      .borrow()
+      .clone()
+      .ok_or(ErrorCode::InvalidInput)?;
+    let promoted = with_control(|control| {
+      control.expect(&format!("RNFR {staged}"))?;
+      let (code, _) =
+        control.send_mutation(&format!("RNTO {}", self.target))?;
+      outcome(code)
+    });
+    if let Err(error) = promoted {
+      self.cleanup_stage();
+      return Err(error);
+    }
+    self.staged.take();
     Ok(Some(FileInfo {
       kind: EntryKind::File,
       size: self.written.get(),
@@ -201,19 +351,64 @@ impl GuestWriter for FtpWriter {
 }
 
 impl Drop for FtpWriter {
-  /// An upload abandoned before it finished must leave no partial file. The
+  /// An abandoned upload closes its data channel, resynchronizes when
-  /// server takes the closed socket as the end of the upload, so what it
+  /// possible, and removes only its private stage. The destination was never
-  /// stored is deleted once the channel is back in sync.
+  /// exposed unless `finish` confirmed the promotion.
   fn drop(&mut self) {
-    let Some(socket) = self.socket.take() else {
+    if let Some(socket) = self.socket.take() {
-      return;
-    };
       drop(socket);
-    if abort().is_ok() {
+      if abort().is_err() {
-      let _ =
+        CONTROL.set(None);
-        with_control(|control| control.expect(&format!("DELE {}", self.path)));
       }
     }
+    self.cleanup_stage();
+  }
+}
+
+fn random_stage_token() -> Result<u128, ErrorCode> {
+  let mut bytes = [0; 16];
+  getrandom::fill(&mut bytes).map_err(|_| ErrorCode::Internal)?;
+  Ok(u128::from_be_bytes(bytes))
+}
+
+/// Chooses an unused, unguessable private sibling for one upload. The remote
+/// probe remains a collision check, while 128 random bits prevent another
+/// client from racing a predictable name between that probe and `STOR`.
+fn stage_path_with(
+  target: &str,
+  mut next_token: impl FnMut() -> Result<u128, ErrorCode>,
+) -> Result<String, ErrorCode> {
+  let (parent, name) =
+    target.rsplit_once('/').ok_or(ErrorCode::InvalidInput)?;
+  if name.is_empty() {
+    return Err(ErrorCode::InvalidInput);
+  }
+  for _ in 0..MAX_STAGE_ATTEMPTS {
+    let token = next_token()?;
+    let staged = format!("{parent}/.memona-upload-{token:032x}");
+    match with_control(|control| probe(control, &staged)) {
+      Err(ErrorCode::NotFound) => return Ok(staged),
+      Ok(_) => {}
+      Err(error) => return Err(error),
+    }
+  }
+  Err(ErrorCode::Conflict)
+}
+
+fn stage_path(target: &str) -> Result<String, ErrorCode> {
+  stage_path_with(target, random_stage_token)
+}
+
+fn start_writer(target: String) -> Result<FtpWriter, ErrorCode> {
+  ready()?;
+  let staged = stage_path(&target)?;
+  let socket = mutation_transfer(&format!("STOR {staged}"))?;
+  Ok(FtpWriter {
+    target,
+    staged: RefCell::new(Some(staged)),
+    socket: RefCell::new(Some(socket)),
+    written: Cell::new(0),
+  })
 }
 
 impl Guest for Ftp {
@@ -240,19 +435,16 @@ impl Guest for Ftp {
       config.iter().find(|(key, _)| key == "tls").map(|(_, v)| v),
       Some(Setting::Toggle(true))
     );
-    let mut control = Control::open(host_name, port, tls)?;
     let user = lookup("username").unwrap_or_else(|| "anonymous".to_owned());
-    let (code, _) = control.send(&format!("USER {user}"))?;
+    let session = Session {
-    if code == 331 {
+      host: host_name,
-      let password = lookup("password").unwrap_or_default();
+      port,
-      let (code, _) = control.send(&format!("PASS {password}"))?;
+      tls,
-      outcome(code)?;
+      user,
-    } else {
+      password: lookup("password").unwrap_or_default(),
-      outcome(code)?;
+    };
-    }
+    CONTROL.set(Some(session.connect()?));
-    // Binary throughout: text mode would rewrite line endings inside files.
+    SESSION.set(Some(session));
-    control.expect("TYPE I")?;
-    CONTROL.set(Some(control));
     Ok(Capabilities {
       write: true,
       mkdir: true,
@@ -273,6 +465,7 @@ impl Guest for Ftp {
       }
       *slot = None;
     });
+    SESSION.set(None);
   }
 
   fn stat(target: Locator) -> Result<FileInfo, ErrorCode> {
@@ -284,6 +477,7 @@ impl Guest for Ftp {
         revision: None,
       });
     }
+    ready()?;
     with_control(|control| probe(control, &path))
   }
 
@@ -305,13 +499,17 @@ impl Guest for Ftp {
     });
     let listing = match kept {
       Some(listing) => listing,
-      None => fetch_listing(&path)?,
+      None => {
+        ready()?;
+        fetch_listing(&path)?
+      }
     };
     page(&listing, start, limit as usize)
   }
 
   fn open_read(target: Locator, offset: u64) -> Result<Reader, ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     if offset > 0 {
       with_control(|control| control.expect(&format!("REST {offset}")))?;
     }
@@ -331,20 +529,16 @@ impl Guest for Ftp {
     if !matches!(condition, WriteCondition::Overwrite) {
       return Err(ErrorCode::Unsupported);
     }
-    let socket = transfer(&format!("STOR {path}"))?;
+    Ok(Writer::new(start_writer(path)?))
-    Ok(Writer::new(FtpWriter {
-      path,
-      socket: RefCell::new(Some(socket)),
-      written: Cell::new(0),
-    }))
   }
 
   /// A refused create is a conflict when the path is already taken, which is
   /// the one thing the server says in words rather than in its reply code.
   fn mkdir(target: Locator) -> Result<FileInfo, ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     with_control(|control| {
-      let (code, text) = control.send(&format!("MKD {path}"))?;
+      let (code, text) = control.send_mutation(&format!("MKD {path}"))?;
       refusal(code, &text)
     })?;
     Ok(FileInfo {
@@ -361,15 +555,18 @@ impl Guest for Ftp {
     destination: String,
   ) -> Result<FileInfo, ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     with_control(|control| {
+      let info = probe(control, &path)?;
       match probe(control, &destination) {
         Ok(_) => return Err(ErrorCode::Conflict),
         Err(ErrorCode::NotFound) => {}
         Err(error) => return Err(error),
       }
       control.expect(&format!("RNFR {path}"))?;
-      control.expect(&format!("RNTO {destination}"))?;
+      let (code, _) = control.send_mutation(&format!("RNTO {destination}"))?;
-      probe(control, &destination)
+      outcome(code)?;
+      Ok(info)
     })
   }
 
@@ -378,12 +575,15 @@ impl Guest for Ftp {
   /// holds children is refused as a conflict, not as an absence.
   fn remove(target: Locator) -> Result<(), ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     with_control(|control| {
-      let (code, _) = control.send(&format!("DELE {path}"))?;
+      let (code, _) = control.send_mutation(&format!("DELE {path}"))?;
-      if (200..300).contains(&code) {
+      match outcome(code) {
-        return Ok(());
+        Ok(()) => return Ok(()),
+        Err(ErrorCode::NotFound) => {}
+        Err(error) => return Err(error),
       }
-      let (code, text) = control.send(&format!("RMD {path}"))?;
+      let (code, text) = control.send_mutation(&format!("RMD {path}"))?;
       refusal(code, &text)
     })
   }
@@ -393,4 +593,26 @@ impl Guest for Ftp {
   }
 }
 
+#[cfg(not(test))]
 memona_plugin_sdk::export!(Ftp);
+
+#[cfg(test)]
+mod tests {
+  use super::text;
+  use memona_plugin_sdk::Setting;
+
+  #[tokio::test]
+  async fn secret_credentials_preserve_significant_whitespace() {
+    for password in [" ftp fixture secret ", "   ", ""] {
+      assert_eq!(
+        text(Some(&Setting::Secret(password.to_owned()))),
+        Some(password.to_owned())
+      );
+    }
+    assert_eq!(
+      text(Some(&Setting::Text(" server.example ".to_owned()))),
+      Some("server.example".to_owned())
+    );
+    assert_eq!(text(Some(&Setting::Text("   ".to_owned()))), None);
+  }
+}

src/listing.rs

@@ -12,7 +12,6 @@ struct Facts<'a> {
 
 fn facts(line: &str) -> Option<Facts<'_>> {
   let (facts, name) = line.split_once(' ')?;
-  let name = name.trim();
   if name.is_empty() {
     return None;
   }
@@ -106,6 +105,21 @@ mod tests {
   use super::{machine_entry, machine_info, page};
   use memona_plugin_sdk::{EntryKind, ErrorCode};
 
+  #[tokio::test]
+  async fn listing_names_preserve_significant_whitespace()
+  -> Result<(), ErrorCode> {
+    let listing = b"type=file;size=1;  report.md \r\ntype=dir;  spaced folder \r\ntype=file;size=2;    \r\n";
+    let listed = page(listing, 0, 10)?;
+    let names: Vec<&str> = listed
+      .entries
+      .iter()
+      .map(|entry| entry.name.as_str())
+      .collect();
+    assert_eq!(names, [" report.md ", " spaced folder ", "   "]);
+    assert!(listed.cursor.is_none());
+    Ok(())
+  }
+
   #[test]
   fn reads_kind_and_size_and_skips_the_directory_itself()
   -> Result<(), &'static str> {

src/session_tests.rs

@@ -0,0 +1,173 @@
+//! FTP command recovery against an in-memory protocol transport. Live QA
+//! separately exercises the compiled WASM component over real host sockets.
+
+use super::{Ftp, stage_path_with, start_writer, test_host};
+use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting};
+
+type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
+
+fn guest<T>(value: Result<T, ErrorCode>) -> TestResult<T> {
+  value.map_err(|error| format!("guest operation failed: {error:?}").into())
+}
+
+fn config() -> Vec<(String, Setting)> {
+  vec![
+    ("host".into(), Setting::Text("127.0.0.1".into())),
+    (
+      "port".into(),
+      Setting::Text(test_host::CONTROL_PORT.to_string()),
+    ),
+    ("username".into(), Setting::Text("fixture".into())),
+    (
+      "password".into(),
+      Setting::Secret(" fixture secret ".into()),
+    ),
+  ]
+}
+
+#[tokio::test]
+async fn a_listing_reconnects_after_the_server_expires_the_control_session()
+-> TestResult {
+  let fixture = test_host::install(test_host::Scenario::ExpiredListing);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
+  assert_eq!(first.entries[0].name, " report.md ");
+  let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
+  assert_eq!(next.entries[0].name, " report.md ");
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 2);
+  assert_eq!(summary.listings, 2);
+  assert_eq!(summary.mkdir_commands, 0);
+  Ok(())
+}
+
+#[tokio::test]
+async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply()
+-> TestResult {
+  let fixture = test_host::install(test_host::Scenario::LostMutationReply);
+  guest(Ftp::open("ftp".into(), config()))?;
+  assert!(matches!(
+    Ftp::mkdir(Locator::Path("/new".into())),
+    Err(ErrorCode::OutcomeUnknown)
+  ));
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 2);
+  assert_eq!(summary.listings, 0);
+  assert_eq!(summary.mkdir_commands, 1);
+  Ok(())
+}
+
+#[tokio::test]
+async fn cancelled_preflight_never_reconnects_or_executes_the_mutation()
+-> TestResult {
+  let fixture = test_host::install(test_host::Scenario::CancelledPreflight);
+  guest(Ftp::open("ftp".into(), config()))?;
+  assert!(matches!(
+    Ftp::mkdir(Locator::Path("/cancelled".into())),
+    Err(ErrorCode::Cancelled)
+  ));
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.listings, 0);
+  assert_eq!(summary.mkdir_commands, 0);
+  Ok(())
+}
+
+#[tokio::test]
+async fn a_lost_upload_completion_reports_an_unknown_outcome() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::LostUploadCompletion);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = guest(start_writer("/target".into()))?;
+  guest(writer.push(b"body".to_vec()))?;
+  assert!(matches!(writer.finish(), Err(ErrorCode::OutcomeUnknown)));
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 2);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
+  assert_eq!(summary.uploaded_bytes, 4);
+  assert_eq!(summary.stage_deletes, 1);
+  assert_eq!(summary.promotions, 0);
+  Ok(())
+}
+
+#[tokio::test]
+async fn an_uncertain_upload_write_discards_the_control_session() -> TestResult
+{
+  let fixture = test_host::install(test_host::Scenario::LostUploadData);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = guest(start_writer("/target".into()))?;
+  assert!(matches!(
+    writer.push(b"body".to_vec()),
+    Err(ErrorCode::OutcomeUnknown)
+  ));
+  let page = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
+  assert_eq!(page.entries[0].name, " report.md ");
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 2);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
+  assert_eq!(summary.uploaded_bytes, 0);
+  assert_eq!(summary.stage_deletes, 1);
+  assert_eq!(summary.promotions, 0);
+  assert_eq!(summary.listings, 1);
+  Ok(())
+}
+
+#[tokio::test]
+async fn a_complete_upload_promotes_its_private_stage() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::SuccessfulUpload);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = guest(start_writer("/target".into()))?;
+  guest(writer.push(b"body".to_vec()))?;
+  let info = guest(writer.finish())?.ok_or("missing upload result")?;
+  assert_eq!(info.size, 4);
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
+  assert_eq!(summary.uploaded_bytes, 4);
+  assert_eq!(summary.stage_deletes, 0);
+  assert_eq!(summary.promotions, 1);
+  Ok(())
+}
+
+#[tokio::test]
+async fn dropping_an_upload_removes_only_its_private_stage() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::AbandonedUpload);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = guest(start_writer("/target".into()))?;
+  guest(writer.push(b"partial".to_vec()))?;
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
+  assert_eq!(summary.uploaded_bytes, 7);
+  assert_eq!(summary.stage_deletes, 1);
+  assert_eq!(summary.promotions, 0);
+  Ok(())
+}
+
+#[tokio::test]
+async fn a_stage_name_collision_draws_a_fresh_random_token() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::StageCollision);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let mut tokens = [1, 2].into_iter();
+  let staged = guest(stage_path_with("/target", || {
+    tokens.next().ok_or(ErrorCode::Internal)
+  }))?;
+  assert_eq!(staged, "/.memona-upload-00000000000000000000000000000002");
+  Ftp::close();
+  assert_eq!(fixture.summary().stage_probes, 2);
+  Ok(())
+}

src/test_host.rs

@@ -0,0 +1,278 @@
+//! In-memory FTP host for the generated synchronous `Guest` API. Production
+//! WASM uses Memona's real async socket host; live QA covers that boundary.
+
+use std::cell::RefCell;
+use std::collections::VecDeque;
+use std::rc::Rc;
+
+pub use memona_plugin_sdk::HostError;
+
+pub const CONTROL_PORT: u16 = 21_210;
+const DATA_PORT: u16 = 31_310;
+const LISTING: &[u8] = b"type=file;size=3;  report.md \r\n";
+
+#[derive(Clone, Copy)]
+pub enum Scenario {
+  AbandonedUpload,
+  CancelledPreflight,
+  ExpiredListing,
+  LostUploadData,
+  LostMutationReply,
+  LostUploadCompletion,
+  StageCollision,
+  SuccessfulUpload,
+}
+
+struct State {
+  scenario: Scenario,
+  control_connections: usize,
+  listings: usize,
+  mkdir_commands: usize,
+  upload_commands: usize,
+  target_upload_commands: usize,
+  uploaded_bytes: usize,
+  stage_deletes: usize,
+  stage_probes: usize,
+  promotions: usize,
+}
+
+thread_local! {
+  static ACTIVE: RefCell<Option<Rc<RefCell<State>>>> = const { RefCell::new(None) };
+}
+
+pub struct Fixture(Rc<RefCell<State>>);
+
+pub struct Summary {
+  pub control_connections: usize,
+  pub listings: usize,
+  pub mkdir_commands: usize,
+  pub upload_commands: usize,
+  pub target_upload_commands: usize,
+  pub uploaded_bytes: usize,
+  pub stage_deletes: usize,
+  pub stage_probes: usize,
+  pub promotions: usize,
+}
+
+pub fn install(scenario: Scenario) -> Fixture {
+  let state = Rc::new(RefCell::new(State {
+    scenario,
+    control_connections: 0,
+    listings: 0,
+    mkdir_commands: 0,
+    upload_commands: 0,
+    target_upload_commands: 0,
+    uploaded_bytes: 0,
+    stage_deletes: 0,
+    stage_probes: 0,
+    promotions: 0,
+  }));
+  ACTIVE.set(Some(state.clone()));
+  Fixture(state)
+}
+
+impl Fixture {
+  pub fn summary(&self) -> Summary {
+    let state = self.0.borrow();
+    Summary {
+      control_connections: state.control_connections,
+      listings: state.listings,
+      mkdir_commands: state.mkdir_commands,
+      upload_commands: state.upload_commands,
+      target_upload_commands: state.target_upload_commands,
+      uploaded_bytes: state.uploaded_bytes,
+      stage_deletes: state.stage_deletes,
+      stage_probes: state.stage_probes,
+      promotions: state.promotions,
+    }
+  }
+}
+
+impl Drop for Fixture {
+  fn drop(&mut self) {
+    ACTIVE.with_borrow_mut(|active| {
+      if active
+        .as_ref()
+        .is_some_and(|state| Rc::ptr_eq(state, &self.0))
+      {
+        *active = None;
+      }
+    });
+  }
+}
+
+enum Kind {
+  Control(usize),
+  Data,
+}
+
+pub struct Tcp {
+  state: Rc<RefCell<State>>,
+  kind: Kind,
+  input: RefCell<VecDeque<u8>>,
+}
+
+pub fn connect(name: &str, port: u16, tls: bool) -> Result<Tcp, HostError> {
+  if tls || name != "127.0.0.1" {
+    return Err(HostError::InvalidInput);
+  }
+  let state = ACTIVE
+    .with_borrow(Clone::clone)
+    .ok_or(HostError::Internal)?;
+  match port {
+    CONTROL_PORT => {
+      let index = {
+        let mut state = state.borrow_mut();
+        let index = state.control_connections;
+        state.control_connections += 1;
+        index
+      };
+      Ok(Tcp {
+        state,
+        kind: Kind::Control(index),
+        input: RefCell::new(VecDeque::from(b"220 Fixture\r\n".to_vec())),
+      })
+    }
+    DATA_PORT => Ok(Tcp {
+      state,
+      kind: Kind::Data,
+      input: RefCell::new(VecDeque::from(LISTING.to_vec())),
+    }),
+    _ => Err(HostError::Denied),
+  }
+}
+
+impl Tcp {
+  pub fn read(&self, limit: u32) -> Result<Vec<u8>, HostError> {
+    let mut input = self.input.borrow_mut();
+    if input.is_empty() && matches!(self.kind, Kind::Control(_)) {
+      return Err(HostError::Unavailable);
+    }
+    let count = usize::min(limit as usize, input.len());
+    Ok(input.drain(..count).collect())
+  }
+
+  pub fn write(&self, bytes: &[u8]) -> Result<u32, HostError> {
+    if matches!(self.kind, Kind::Data) {
+      let mut state = self.state.borrow_mut();
+      if matches!(
+        state.scenario,
+        Scenario::AbandonedUpload
+          | Scenario::LostUploadCompletion
+          | Scenario::SuccessfulUpload
+      ) {
+        state.uploaded_bytes += bytes.len();
+        return Ok(bytes.len() as u32);
+      }
+      return Err(HostError::InvalidInput);
+    }
+    let Kind::Control(index) = self.kind else {
+      unreachable!();
+    };
+    let command = std::str::from_utf8(bytes)
+      .map_err(|_| HostError::InvalidInput)?
+      .strip_suffix("\r\n")
+      .ok_or(HostError::InvalidInput)?;
+    let response = {
+      let mut state = self.state.borrow_mut();
+      match command {
+        "USER fixture" => "331 Password required\r\n",
+        "PASS  fixture secret " => "230 Signed in\r\n",
+        "TYPE I" => "200 Binary\r\n",
+        "QUIT" => "221 Bye\r\n",
+        "EPSV" => "229 Passive (|||31310|)\r\n",
+        "MLSD /"
+          if matches!(
+            state.scenario,
+            Scenario::ExpiredListing | Scenario::LostUploadData
+          ) =>
+        {
+          state.listings += 1;
+          "150 Listing\r\n226 Complete\r\n"
+        }
+        "NOOP" if matches!(state.scenario, Scenario::CancelledPreflight) => {
+          return Err(HostError::Cancelled);
+        }
+        "NOOP"
+          if index == 0
+            && (matches!(state.scenario, Scenario::LostMutationReply)
+              || state.listings > 0) =>
+        {
+          return Err(HostError::Unavailable);
+        }
+        "NOOP" => "200 Still connected\r\n",
+        "MLST /.memona-upload-00000000000000000000000000000001"
+          if matches!(state.scenario, Scenario::StageCollision) =>
+        {
+          state.stage_probes += 1;
+          "250-Begin\r\n type=file;size=3; /.memona-upload-00000000000000000000000000000001\r\n250 End\r\n"
+        }
+        command if command.starts_with("MLST /.memona-upload-") => {
+          state.stage_probes += 1;
+          "550 Missing\r\n"
+        }
+        command
+          if command.starts_with("STOR /.memona-upload-")
+            && matches!(
+              state.scenario,
+              Scenario::AbandonedUpload | Scenario::LostUploadCompletion
+            ) =>
+        {
+          state.upload_commands += 1;
+          "150 Upload started\r\n"
+        }
+        command
+          if command.starts_with("STOR /.memona-upload-")
+            && matches!(state.scenario, Scenario::LostUploadData) =>
+        {
+          state.upload_commands += 1;
+          "150 Upload started\r\n"
+        }
+        command
+          if command.starts_with("STOR /.memona-upload-")
+            && matches!(state.scenario, Scenario::SuccessfulUpload) =>
+        {
+          state.upload_commands += 1;
+          "150 Upload started\r\n226 Upload complete\r\n"
+        }
+        command if command.starts_with("STOR ") => {
+          state.target_upload_commands += 1;
+          "150 Target upload started\r\n"
+        }
+        command if command.starts_with("DELE /.memona-upload-") => {
+          state.stage_deletes += 1;
+          "250 Stage deleted\r\n"
+        }
+        command if command.starts_with("RNFR /.memona-upload-") => {
+          "350 Ready for destination\r\n"
+        }
+        "RNTO /target"
+          if matches!(state.scenario, Scenario::SuccessfulUpload) =>
+        {
+          state.promotions += 1;
+          "250 Promoted\r\n"
+        }
+        "ABOR" if matches!(state.scenario, Scenario::AbandonedUpload) => {
+          "426 Upload abandoned\r\n226 Abort complete\r\n"
+        }
+        "MKD /new"
+          if index == 1
+            && matches!(state.scenario, Scenario::LostMutationReply) =>
+        {
+          state.mkdir_commands += 1;
+          return Ok(bytes.len() as u32);
+        }
+        "MKD /cancelled"
+          if index == 1
+            && matches!(state.scenario, Scenario::CancelledPreflight) =>
+        {
+          state.mkdir_commands += 1;
+          return Err(HostError::Unavailable);
+        }
+        _ => return Err(HostError::InvalidInput),
+      }
+    };
+    self.input.borrow_mut().extend(response.as_bytes());
+    Ok(bytes.len() as u32)
+  }
+}