@@ -19,7 +19,7 @@ target-side daemon.
 # @local
 VERSION=$(git rev-parse --short HEAD)
 
-# @export VERSION
+# @export VERSION --public
 
 # @copy target/release/myapp-$VERSION -> /tmp/shellflow/myapp-$VERSION @all
 
@@ -64,7 +64,10 @@ deploy.sh` runs it unchanged, and your editor highlights it perfectly.
   warning instead of being performed: a dry run never writes to a target.
 - **State passing** — `@export VAR` captures a variable from a local block and
   injects it into later `@copy` paths, `@remote` blocks, and `@only_if` guards
-  as `$VAR` / `${VAR}` / env.
+  as `$VAR` / `${VAR}` / env. A captured value is **masked in all output** by
+  default (it is a value the tool did not see at parse time, and a playbook
+  routinely exports a credential it just fetched); add `--public` for a value
+  that is meant to be read, such as a version or a commit.
 - **Graceful file copies** — `@copy` uses rsync when the target has it (with a
   `mkdir -p` remote wrapper), and automatically falls back to
   `ssh mkdir -p` + `scp` on hosts without rsync. Destination directories are
@@ -142,7 +145,7 @@ lines belong to the current block.
 | `@local` | `# @local` | Following lines run locally (default) |
 | `@remote` | `# @remote <target>` | Following lines stream to the target (alias, group, or raw spec) |
 | `@copy` | `# @copy <src> -> <dst> @<target> [--delete] [--exclude <glob>]` | Copy a local path to the target (rsync, or scp fallback); a directory source is flattened into `dst` (contents semantics, identical on GNU rsync and macOS openrsync); supports `$VAR` interpolation; `--exclude` is repeatable and honored by rsync (warned as ignored on the scp fallback and in `--local` mode) |
-| `@export` | `# @export <VAR>[,<VAR>…]` | Capture variables from the preceding local block into run state |
+| `@export` | `# @export <VAR>[,<VAR>…] [--public]` | Capture variables from the preceding local block into run state. Captured values are masked in all output by default; `--public` opts out for a value that is meant to be read |
 | `@timeout` | `# @timeout <seconds>` | Per-step timeout for the next block |
 | `@name` | `# @name <label>` | Name the next block (for `--only`/`--skip` and reporting) |
 | `@only_if` | `# @only_if <command>` | Skip the next block where `<command>` fails |
 
@@ -70,7 +70,7 @@ lines belong to the current block.
 | `@local` | `# @local` | Following lines run locally (default) |
 | `@remote` | `# @remote <target>` | Following lines stream to the target (alias, group, raw spec) |
 | `@copy` | `# @copy <src> -> <dst> @<target> [--delete] [--exclude <glob>]` | Copy a local path (rsync, or scp fallback); creates the destination dir; `$VAR` interpolated; `--exclude` is repeatable and honored by rsync. On a host without rsync the scp fallback cannot filter or mirror, so a step that asked for `--delete`/`--exclude` **fails closed** unless `--allow-unfiltered-copy` accepts the downgrade |
-| `@export` | `# @export <VAR>[,<VAR>…]` | Capture variables from the preceding local block into run state |
+| `@export` | `# @export <VAR>[,<VAR>…] [--public]` | Capture variables from the preceding local block into run state. Every captured value is masked in the output (terminal, previews, `--log-file`) unless the line says `--public`; use `--public` for versions, timestamps, paths and other non-secret values |
 | `@timeout` | `# @timeout <seconds>` | Per-step timeout for the next block |
 | `@name` | `# @name <label>` | Name the next block (for `--only`/`--skip`/reporting) |
 | `@only_if` | `# @only_if <command>` | Skip the next block where `<command>` fails |
@@ -112,6 +112,9 @@ Behavioral rules to remember:
 - `@copy` paths are parsed by shellflow, not Bash. Capture values with
   `@export` and reference them as `$VAR` / `${VAR}`. Unresolved variables are
   a hard error in a real run, and a warning + skip in `--dry-run`/`--check`.
+  A captured value is masked in every rendered line unless the declaration says
+  `--public`; a secret exported under an ordinary name (`DATABASE_URL`) is
+  therefore still not printed.
 - `@only_if` guards run per host with the same env as the block; a failed
   guard reports `SKIPPED`.
 - `@copy` `src`/`dst` must not contain `->`; the target is the last
@@ -141,7 +144,7 @@ mutations confined to `/tmp` unless the user explicitly asks otherwise.
 STAMP="sf-$(date +%s)"
 WORKDIR="/tmp/${STAMP}"
 mkdir -p "${WORKDIR}"
-# @export STAMP,WORKDIR
+# @export STAMP,WORKDIR --public
 
 # @copy ./playbooks/marker.txt -> $WORKDIR/marker.txt @all
 
 
@@ -23,7 +23,9 @@ use tokio::sync::Semaphore;
 
 use super::stream;
 use crate::{
-    executor::{ArcUi, HostOutcome, LogFile, RunError, StepConfig, cancel::Cancel, fanout_slot},
+    executor::{
+        ArcUi, HostOutcome, LogFile, RunError, StepConfig, cancel::Cancel, fanout_slot, quote,
+    },
     rbatch::{Batch, SshDest},
 };
 
@@ -42,15 +44,10 @@ fn check_budget(check: &Assertion, step_timeout: Option<u64>) -> u64 {
     match check {
         Assertion::Port { .. } => step_timeout.unwrap_or(DEFAULT_PORT_TIMEOUT_SECS),
         Assertion::Health { .. } => step_timeout.unwrap_or(DEFAULT_HEALTH_TIMEOUT_SECS),
-        // The unit-state, enablement, port-free, and rendered checks answer
-        // at once.
-        Assertion::UnitState { .. } |
-        Assertion::Enabled { .. } |
-        Assertion::PortFree { .. } |
-        Assertion::Rendered { .. } |
-        Assertion::NoLeak { .. } |
-        Assertion::Digest { .. } |
-        Assertion::Credstore { .. } => 0,
+        // The unit-state, enablement, port-free, rendered, leak, digest and
+        // credential-store checks all answer at once — and so would a variant
+        // this build does not know (`Assertion` is `#[non_exhaustive]`).
+        _ => 0,
     }
 }
 
@@ -393,6 +390,11 @@ fn command_for(check: &Assertion, step_timeout: Option<u64>) -> String {
             helper = quote(crate::executor::host::HELPER_PATH),
             service = quote(service),
         ),
+        // `Assertion` is `#[non_exhaustive]`: an unknown check cannot be
+        // evaluated, and reporting it as passed would be a lie — the command
+        // fails and says why.
+        _ => "echo 'this build does not recognise the assertion (upgrade shellflow)'; false"
+            .to_string(),
     }
 }
 
@@ -442,14 +444,9 @@ fn status_glob(pattern: &str) -> String {
     pattern.to_string()
 }
 
-/// Quote a value for the remote shell.
-///
-/// The value comes from the playbook (a unit name, a state, a URL), so it
-/// is quoted rather than trusted: a unit name with a space must not become
-/// two arguments.
-fn quote(value: &str) -> String {
-    shellflow_core::shell_quote_single(value)
-}
+// `quote` is the executor-wide helper (`super::quote`): the value comes from
+// the playbook (a unit name, a state, a URL), so it is quoted rather than
+// trusted — a unit name with a space must not become two arguments.
 
 #[cfg(test)]
 mod tests {
 
@@ -7,7 +7,7 @@
 //! runner, so a new service adds one declarative block instead of a new
 //! build script.
 
-use super::host::quote;
+use crate::executor::quote;
 
 /// The remote command for `@build`.
 ///
@@ -161,22 +161,27 @@ pub(super) fn build_command(build: &shellflow_core::BuildStep) -> String {
 mod tests {
     use super::build_command;
 
+    /// Build a `BuildStep` through the parser.
+    ///
+    /// The payload structs are `#[non_exhaustive]`, so the DSL is the only
+    /// constructor a consumer has — which is the point: a step carries
+    /// invariants the parser enforced (a declared target, a required
+    /// `--build`), and a struct literal would skip them.
+    fn step(directive: &str) -> shellflow_core::BuildStep {
+        let script = format!("# @server api api\n{directive}\n");
+        let plan = shellflow_core::parse_script(&script).expect("the fixture parses");
+        match &plan.steps()[0] {
+            shellflow_core::Step::Build(step) => step.clone(),
+            other => panic!("expected a build step, got {other:?}"),
+        }
+    }
+
     fn sample() -> String {
-        build_command(&shellflow_core::BuildStep {
-            name: None,
-            target: shellflow_core::Target::new("api"),
-            repo: "https://github.com/x/y.git".to_string(),
-            git_ref: "abc123".to_string(),
-            build_cmd: "cargo build --release -p app".to_string(),
-            prep_cmd: None,
-            workdir: None,
-            artifacts: vec!["target/release/app".to_string()],
-            install_dirs: vec!["/opt/app/bin".to_string()],
-            units: vec!["app.service".to_string()],
-            version_flag: "--version".to_string(),
-            min_free_gb: 8,
-            timeout: None,
-        })
+        build_command(&step(
+            "# @build https://github.com/x/y.git @api --ref abc123 \
+             --build 'cargo build --release -p app' --artifacts target/release/app \
+             --install-dirs /opt/app/bin --units app.service --version-flag=--version",
+        ))
     }
 
     #[test]
@@ -209,41 +214,19 @@ mod tests {
         let cmd = sample();
         assert!(cmd.contains("--version"), "{cmd}");
         // …and falls back to the flag chain when no flag is declared.
-        let noflag = build_command(&shellflow_core::BuildStep {
-            name: None,
-            target: shellflow_core::Target::new("api"),
-            repo: "https://github.com/x/y.git".to_string(),
-            git_ref: "r".to_string(),
-            build_cmd: "make".to_string(),
-            prep_cmd: None,
-            workdir: None,
-            artifacts: vec!["app".to_string()],
-            install_dirs: vec!["/opt/a".to_string()],
-            units: vec![],
-            version_flag: String::new(),
-            min_free_gb: 8,
-            timeout: None,
-        });
+        let noflag = build_command(&step(
+            "# @build https://github.com/x/y.git @api --ref r --build make \
+             --artifacts app --install-dirs /opt/a",
+        ));
         assert!(noflag.contains("--version -V -v -version version"), "{noflag}");
     }
 
     #[test]
     fn build_values_are_quoted() {
-        let cmd = build_command(&shellflow_core::BuildStep {
-            name: None,
-            target: shellflow_core::Target::new("api"),
-            repo: "https://github.com/x/y.git".to_string(),
-            git_ref: "r; evil".to_string(),
-            build_cmd: "make".to_string(),
-            prep_cmd: None,
-            workdir: None,
-            artifacts: vec!["a".to_string()],
-            install_dirs: vec!["/opt/a".to_string()],
-            units: vec![],
-            version_flag: String::new(),
-            min_free_gb: 8,
-            timeout: None,
-        });
+        let cmd = build_command(&step(
+            "# @build https://github.com/x/y.git @api --ref 'r; evil' --build make \
+             --artifacts a --install-dirs /opt/a",
+        ));
         assert!(cmd.contains("'r; evil'"), "{cmd}");
     }
 
@@ -255,21 +238,12 @@ mod tests {
 
     #[test]
     fn build_installs_renamed_artifacts_under_the_new_name() {
-        let mut step = shellflow_core::BuildStep {
-            name: None,
-            target: shellflow_core::Target::new("ai"),
-            repo: "https://github.com/bnb-chain/reth.git".to_string(),
-            git_ref: "5710c1f".to_string(),
-            build_cmd: "cargo build --release -p reth".to_string(),
-            prep_cmd: None,
-            workdir: None,
-            artifacts: vec!["target/release/reth=reth-bsc".to_string()],
-            install_dirs: vec!["/usr/local/bin".to_string()],
-            units: vec!["bsc-reth-testnet.service".to_string()],
-            version_flag: "--version".to_string(),
-            min_free_gb: 20,
-            timeout: None,
-        };
+        let mut step = step(
+            "# @build https://github.com/bnb-chain/reth.git @api --ref 5710c1f \
+             --build 'cargo build --release -p reth' \
+             --artifacts target/release/reth=reth-bsc --install-dirs /usr/local/bin \
+             --units bsc-reth-testnet.service --version-flag=--version --min-free-gb 20",
+        );
         let cmd = build_command(&step);
         // The installed file and its backup carry the *new* name; the
         // built file keeps the workspace name.
 
@@ -8,9 +8,9 @@ use shellflow_core::{CopyStep, ResolvedHost};
 use tokio::{process::Command, task::JoinSet};
 
 use super::{
-    ArcUi, CopyStepArgs, DiffFilter, HostOutcome, LogFile, StepConfig, fanout_slot,
+    ArcUi, CopyStepArgs, DiffFilter, HostOutcome, LogFile, RunError, StepConfig, fanout_slot,
     process::{GroupGuard, isolate},
-    remote::{RunError, resolve_remote_env},
+    remote::resolve_remote_env,
     stream,
 };
 
@@ -340,7 +340,7 @@ async fn run_copy_step_local(args: LocalCopyArgs<'_>) -> Result<Vec<(String, Hos
         // after `cp` succeeded. Wiping first (the previous behaviour) meant a
         // failed copy left the destination empty — the worst possible outcome
         // for a `--delete` deploy.
-        return run_local_copy_delete(src, dst, config, cancel).await;
+        return run_local_copy_delete(src, dst, config, cancel, log_file).await;
     }
 
     if let Some(parent) = std::path::Path::new(dst).parent() &&
@@ -354,7 +354,7 @@ async fn run_copy_step_local(args: LocalCopyArgs<'_>) -> Result<Vec<(String, Hos
     let mut cmd = Command::new("cp");
     cmd.arg("-a").arg(src).arg(dst);
     cmd.kill_on_drop(true);
-    let status = run_command_status(cmd, config.timeout, "local copy", cancel).await?;
+    let status = run_command_status(cmd, config.timeout, "local copy", cancel, log_file).await?;
     if !status.success() {
         bail!("local copy exited with status {status}");
     }
@@ -372,6 +372,7 @@ async fn run_local_copy_delete(
     dst: &str,
     config: StepConfig,
     cancel: &super::Cancel,
+    log_file: &LogFile,
 ) -> Result<Vec<(String, HostOutcome)>> {
     let dst_path = std::path::Path::new(dst);
     if let Some(parent) = dst_path.parent().filter(|p| !p.as_os_str().is_empty()) {
@@ -392,7 +393,7 @@ async fn run_local_copy_delete(
     let mut cmd = Command::new("cp");
     cmd.arg("-a").arg(src).arg(&staging);
     cmd.kill_on_drop(true);
-    let status = run_command_status(cmd, config.timeout, "local copy", cancel).await?;
+    let status = run_command_status(cmd, config.timeout, "local copy", cancel, log_file).await?;
     if !status.success() {
         // The old destination is untouched; drop the half-built staging tree.
         remove_path(&staging).await?;
@@ -430,6 +431,7 @@ async fn run_command_status(
     timeout: Option<u64>,
     label: &str,
     cancel: &super::Cancel,
+    log_file: &LogFile,
 ) -> Result<std::process::ExitStatus> {
     isolate(&mut cmd);
     cmd.kill_on_drop(true);
@@ -447,7 +449,7 @@ async fn run_command_status(
             return Err(err).wrap_err(format!("failed to run {label}"));
         }
         super::process::ChildOutcome::TimedOut => {
-            crate::ui::warn(&format!("{label} timed out — killed"), &None);
+            crate::ui::warn(&format!("{label} timed out — killed"), log_file);
             super::process::terminate(&mut child).await;
             guard.disarm();
             bail!("{label} timed out");
@@ -479,8 +481,13 @@ struct RsyncHostArgs<'a> {
 async fn run_rsync_host(args: RsyncHostArgs<'_>) -> (String, HostOutcome) {
     let RsyncHostArgs { ui, host, srcs, dst, opts, config, has_rsync, log_file, cancel } = args;
     // Never a bare `ssh`: a transport without BatchMode can stop at a
-    // password or host-key prompt in the middle of an unattended deploy.
-    let remote_shell = host.spec.rsync_transport();
+    // password or host-key prompt in the middle of an unattended deploy. The
+    // string comes from the one policy builder, so `--no-ssh-reuse` and the
+    // connect bound reach the transfer too.
+    let remote_shell = super::transport::rsync_transport(
+        super::transport::SshPolicy::step(config.ssh_reuse, false),
+        &host.spec,
+    );
 
     if config.verbose >= 2 {
         crate::ui::note(
@@ -510,7 +517,7 @@ async fn run_rsync_host(args: RsyncHostArgs<'_>) -> (String, HostOutcome) {
                 ),
                 log_file,
             );
-            return (host.alias.clone(), HostOutcome::Failed(super::remote::RunError::Config));
+            return (host.alias.clone(), HostOutcome::Failed(super::RunError::Config));
         }
         crate::ui::warn(
             &format!(
@@ -615,8 +622,9 @@ async fn run_rsync_transfer(args: RsyncTransferArgs<'_>) -> std::result::Result<
             _ => dst.as_str(),
         }
     };
-    let quoted = dir.replace('\'', "'\\''");
-    let mkdir_rsync = format!("mkdir -p '{quoted}' && rsync");
+    // The remote shell parses `--rsync-path` verbatim, so the directory goes
+    // through the one quoting helper instead of a hand-rolled `'\''` swap.
+    let mkdir_rsync = format!("mkdir -p -- {} && rsync", super::quote(dir));
     cmd.args(["--rsync-path", &mkdir_rsync, "-e", args.remote_shell]);
     let dest = format!("{}:{dst}", args.host.spec.dest());
     // Several sources land in one destination directory; rsync keeps each
@@ -727,12 +735,11 @@ async fn run_scp_transfer(
     let scp_srcs: Vec<String> =
         if srcs.len() == 1 && is_dir_src { vec![format!("{first}/.")] } else { srcs.to_vec() };
 
-    let quoted_dir = dir.replace('\'', "'\\''");
-    let mut mkdir = Command::new("ssh");
-    // Fail closed on an unknown host key or a missing key: never prompt.
-    mkdir.args(["-o", "BatchMode=yes"]);
-    mkdir.args(host.spec.to_ssh_args()).arg("-T");
-    mkdir.arg(format!("mkdir -p '{quoted_dir}'"));
+    let mut mkdir = Command::from(super::transport::ssh_command(
+        super::transport::SshPolicy::step(config.ssh_reuse, false),
+        &host.spec,
+    ));
+    mkdir.arg(format!("mkdir -p -- {}", super::quote(&dir)));
     mkdir
         .stdin(std::process::Stdio::null())
         .stdout(std::process::Stdio::null())
@@ -744,7 +751,7 @@ async fn run_scp_transfer(
     // Routed through `run_command_status` so the mkdir honours the per-step
     // timeout and owns a `GroupGuard`: an unbounded `ssh` here used to hang a
     // `--timeout`-bounded copy step forever.
-    let status = run_command_status(mkdir, mkdir_timeout, "remote mkdir", &cancel)
+    let status = run_command_status(mkdir, mkdir_timeout, "remote mkdir", &cancel, log_file)
         .await
         .map_err(|err| CopyFailure::transport(format!("failed to create remote dir: {err:#}")))?;
     if !status.success() {