diff --git a/src/pipeline/remote_session.jl b/src/pipeline/remote_session.jl
index cbc4a848..637d82c0 100644
--- a/src/pipeline/remote_session.jl
+++ b/src/pipeline/remote_session.jl
@@ -518,18 +518,29 @@ function run_remote(emit, target, command_for::Function;
                 sleep(_poll_interval)
             end
             emit("  Retrieving $(length(outputs)) output file(s) from $(target.host)")
-            # A failed fetch (a missing required output, a truncated transfer)
-            # throws with the attempt's exit marker still in place. The next run
-            # takes the `exit 0` barrier above, which replays the retained
-            # attempt instead of relaunching it, so a short read never wedges
-            # the stage and its outputs stay available to retrieve again.
+            # A short read is different from a command that never wrote the
+            # file. `replay` retrieves the same attempt; it cannot make a
+            # missing required output appear, so replaying it on every rerun
+            # would report the same error forever. Retire the terminal attempt
+            # (cleanup is allowed because its exit marker is present) and its
+            # journal, so an explicit rerun launches a fresh attempt. If
+            # cleanup fails, keep the journal: the attempt must not be left
+            # without an owner, and the next run can retry the retirement.
             missing = _missing_outputs(ssh, target, dir, outputs)
             if !isempty(missing)
-                journal["failed"] = false
-                _save_journal(journal_path, journal)
+                retired = try
+                    _control(ssh, target, dir, "cleanup")
+                    true
+                catch e
+                    e isa InterruptException && rethrow()
+                    @warn "Could not retire a zero-exit attempt with missing outputs" host=target.host directory=dir exception=e
+                    false
+                end
+                retired && mv(journal_path, journal_path * ".missing-outputs-$(uuid4())")
                 error("Remote $stage finished with status 0 but did not write required output(s): " *
-                      join(missing, ", ") * ". The attempt is retained; run the stage again to " *
-                      "retrieve it, or change its settings to run it afresh.")
+                      join(missing, ", ") * ". " *
+                      (retired ? "The attempt was retired; run the stage again to launch it afresh."
+                               : "Retry to retire the attempt before the stage can run afresh."))
             end
             _fetch_outputs(ssh, scp, target, dir, outputs, optional_outputs)
             journal["retrieved"] = Dict(abspath(path) => (isfile(path) ? _digest(path) : nothing)
diff --git a/test/unit/remote_protocol_tests.jl b/test/unit/remote_protocol_tests.jl
index cbb5e95a..32120940 100644
--- a/test/unit/remote_protocol_tests.jl
+++ b/test/unit/remote_protocol_tests.jl
@@ -22,6 +22,8 @@ function _remote_fixture(f)
         drop_status = Ref(false)
         drop_start_reply = Ref(false)
         fail_cancel = Ref(false)
+        fail_sweep = Ref(false)
+        fail_cleanup = Ref(false)
         retries = Ref(0)
         ssh = function (_, command)
             if probe_launch[] && occursin("worker.sh' start", command)
@@ -32,6 +34,9 @@ function _remote_fixture(f)
                 return `bash -c '(exec 9>"$1"; flock -x 9; touch "$2"; sleep 0.5) & while ! test -f "$2"; do sleep 0.01; done; bash -c "$3"' _ $lockfile $held $command`
             end
             occursin("worker.sh' retry", command) && (retries[] += 1)
+            if fail_sweep[] && occursin("for d in ", command)
+                return `bash -c 'exit 255'`
+            end
             if drop_status[] && occursin("worker.sh' status", command)
                 return `bash -c 'exit 255'`
             end
@@ -42,6 +47,9 @@ function _remote_fixture(f)
             if fail_cancel[] && occursin("worker.sh' cancel", command)
                 return `bash -c 'exit 255'`
             end
+            if fail_cleanup[] && occursin("worker.sh' cleanup", command)
+                return `bash -c 'exit 255'`
+            end
             `bash -c $command`
         end
         scp = function (src, dst)
@@ -73,7 +81,7 @@ function _remote_fixture(f)
         end
         try
             f((; d, target, input, uploads, downloads, fail_download, fail_control, probe_launch, drop_status,
-                drop_start_reply, fail_cancel, retries, invoke, ssh, scp))
+                drop_start_reply, fail_cancel, fail_sweep, fail_cleanup, retries, invoke, ssh, scp))
         finally
             # Never let a failed assertion strand a detached test process.
             for dir in readdir(target.base; join=true)
@@ -101,6 +109,63 @@ function _task_error(t)
     end
 end
 
+# The branches that own and close their own SSH master are only reachable when
+# `_connection` is nothing. A transport that refuses every request is enough to
+# exercise that ownership and its teardown: no sshd, account or second
+# connection is needed, and the fake executables are visible only inside the
+# `withenv` block.
+function _refusing_transport(f)
+    mktempdir() do bin
+        for name in ("ssh", "scp")
+            path = joinpath(bin, name)
+            write(path, "#!/bin/sh\nexit 255\n")
+            chmod(path, 0o700)
+        end
+        withenv("PATH" => bin * ":" * get(ENV, "PATH", "")) do
+            f()
+        end
+    end
+end
+
+# Model ControlMaster=auto opening a master before a remote request fails.
+# Keep per-ControlPath state so an exit request must close an existing master.
+function _owned_transport(f, fixture)
+    mktempdir(fixture.d) do bin
+        events = joinpath(bin, "events")
+        masters = mkpath(joinpath(bin, "masters"))
+        script = raw"""#!/bin/sh
+ctl=
+action=request
+for arg in "$@"; do
+    case "$arg" in
+        ControlPath=*) ctl=${arg#ControlPath=} ;;
+        exit) action=exit ;;
+    esac
+done
+[ -n "$ctl" ] || exit 254
+master="$MM_TEST_MASTERS/${ctl##*/}"
+if [ "$action" = exit ]; then
+    [ -f "$master" ] || exit 254
+    rm -- "$master" || exit 254
+    printf 'closed\t%s\n' "$ctl" >> "$MM_TEST_EVENTS"
+    exit 0
+fi
+if [ ! -f "$master" ]; then
+    : > "$master"
+    printf 'opened\t%s\n' "$ctl" >> "$MM_TEST_EVENTS"
+fi
+printf 'request\t%s\t%s\n' "$ctl" "$*" >> "$MM_TEST_EVENTS"
+exit 255
+"""
+        write(joinpath(bin, "ssh"), script)
+        chmod(joinpath(bin, "ssh"), 0o700)
+        withenv("PATH" => bin * ":" * get(ENV, "PATH", ""),
+                "MM_TEST_MASTERS" => masters, "MM_TEST_EVENTS" => events) do
+            f((; events, masters))
+        end
+    end
+end
+
 @testset "Durable remote execution protocol" begin
     @testset "a transient status probe cannot lose a detached launch" begin
         _remote_fixture() do f
@@ -594,11 +659,13 @@ end
         end
     end
 
-    @testset "a completed attempt without a required output stays retrievable" begin
+    @testset "a zero exit without a required output is retired for a fresh attempt" begin
         _remote_fixture() do f
-            command = d -> "cd '$d' && true"
-            # A zero exit that omits the required file is reported, not accepted
-            # as a successful retrieval, and its attempt is kept.
+            # `replay` cannot recover a file the command never wrote. The same
+            # command must be able to succeed on an explicit rerun, so the
+            # zero-exit attempt is cleaned up and its journal archived.
+            flag = joinpath(f.d, "write-output")
+            command = d -> "cd '$d' && if test -f '$flag'; then cat input > output; fi"
             error_text = try
                 f.invoke(command)
                 ""
@@ -606,18 +673,58 @@ end
                 sprint(showerror, e)
             end
             @test occursin("required output", error_text)
-            dir = _dir(f)
-            @test _RX._remote_state(f.ssh, f.target, dir) == "exit 0"
-            @test !haskey(_journal(f), "retrieved")
+            @test occursin("retired", error_text)
+            @test !isfile(joinpath(f.d, "stage.log.remote.json"))
+            @test any(n -> startswith(n, "stage.log.remote.json.missing-outputs-"),
+                      readdir(f.d))
+            @test isempty(readdir(f.target.base))
             uploads = f.uploads[]
-            # The next run retrieves from the same attempt: no upload, no
-            # relaunch, and the file that was missing is fetched and verified.
-            write(joinpath(dir, "output"), "known input\n")
+            touch(flag)
             f.invoke(command)
-            @test f.uploads[] == uploads
+            @test f.uploads[] > uploads
             @test read(joinpath(f.d, "stage.out"), String) == "known input\n"
             @test haskey(_journal(f), "retrieved")
+        end
+    end
+
+    @testset "a refused retirement keeps the journal so the attempt keeps its owner" begin
+        _remote_fixture() do f
+            # If the host refuses the cleanup, the attempt still exists there.
+            # The journal is the only record that names it, so it must stay
+            # until a later run retires the attempt.
+            flag = joinpath(f.d, "write-output")
+            command = d -> "cd '$d' && if test -f '$flag'; then cat input > output; fi"
+            journal = joinpath(f.d, "stage.log.remote.json")
+            failure() = try
+                f.invoke(command)
+                ""
+            catch e
+                sprint(showerror, e)
+            end
+            f.fail_cleanup[] = true
+            error_text = failure()
+            @test occursin("required output", error_text)
+            @test occursin("Retry to retire the attempt", error_text)
+            @test !occursin("was retired", error_text)
+            @test isfile(journal)
+            @test !any(n -> occursin(".missing-outputs-", n), readdir(f.d))
+            dir = _dir(f)
+            @test isdir(dir)
+            # The host accepts the cleanup again: the next run retires the same
+            # attempt and does not launch the command a second time.
+            f.fail_cleanup[] = false
+            uploads = f.uploads[]
+            error_text = failure()
+            @test occursin("was retired", error_text)
+            @test f.uploads[] == uploads
             @test !isdir(dir)
+            @test !isfile(journal)
+            @test any(n -> startswith(n, "stage.log.remote.json.missing-outputs-"), readdir(f.d))
+            # Only then does a run launch afresh.
+            touch(flag)
+            f.invoke(command)
+            @test f.uploads[] > uploads
+            @test read(joinpath(f.d, "stage.out"), String) == "known input\n"
         end
     end
 
@@ -626,8 +733,13 @@ end
             # A retained attempt carries the diagnostics of the run whose
             # outputs are being recovered; a stage that reopened its log keeps
             # its header and gains the remote text once, however often the
-            # recovery path is reached.
-            @test_throws ErrorException f.invoke(d -> "cd '$d' && echo stage-diagnostics && true")
+            # recovery path is reached. A broken download (not a missing
+            # output) is what keeps this attempt: a command that omits a
+            # required file is retired for a fresh launch instead.
+            command = d -> "cd '$d' && echo stage-diagnostics && cat input > output"
+            f.fail_download[] = true
+            @test_throws Exception f.invoke(command)
+            f.fail_download[] = false
             dir = _dir(f)
             log = joinpath(f.d, "stage.log")
             write(log, "=== denoise ===\n")
@@ -650,6 +762,175 @@ end
         end
     end
 
+    @testset "the fetch reports a required output it cannot see" begin
+        _remote_fixture() do f
+            out = joinpath(f.d, "never.out")
+            caught = try
+                _RX._fetch_outputs(f.ssh, f.scp, f.target,
+                    joinpath(f.target.base, "absent"), ["output" => out],
+                    Pair{String,String}[])
+                nothing
+            catch e
+                e
+            end
+            @test caught isa ErrorException
+            @test occursin("finished without required output", sprint(showerror, caught))
+            @test !isfile(out)
+        end
+    end
+
+    @testset "a durable cancel request is honoured before a receipt is reused" begin
+        _remote_fixture() do f
+            command = d -> "cd '$d' && cat input > output"
+            f.invoke(command)
+            journal = _journal(f)
+            journal["cancel_requested"] = true
+            write(joinpath(f.d, "stage.log.remote.json"), JSON3.write(journal))
+            uploads = f.uploads[]
+            f.invoke(command)
+            @test read(joinpath(f.d, "stage.out"), String) == "known input\n"
+            @test f.uploads[] > uploads
+            @test _journal(f)["cancel_requested"] == false
+            @test haskey(_journal(f), "retrieved")
+            @test isempty(readdir(f.target.base))
+        end
+    end
+
+    @testset "a durable cancel request stops the old attempt of the same stage" begin
+        _remote_fixture() do f
+            # The host did not acknowledge the stop, so the attempt is still
+            # running there. The SAME stage runs again (no setting changed):
+            # the pending request must stop that attempt before a new one
+            # launches, or two commands of one stage would be live at once.
+            flag = joinpath(f.d, "finish-now")
+            command = d -> "cd '$d' && if test -f '$flag'; then cat input > output; " *
+                           "else echo started > started; sleep 300; fi"
+            t = @async f.invoke(command)
+            _wait_remote(() -> isfile(joinpath(f.d, "stage.log.remote.json")) &&
+                isfile(joinpath(_dir(f), "started")))
+            dir = _dir(f)
+            f.fail_cancel[] = true
+            schedule(t, InterruptException(); error=true)
+            @test !isnothing(_task_error(t))
+            @test _journal(f)["cancel_requested"]
+            @test isdir(dir)
+            f.fail_cancel[] = false
+            touch(flag)
+            uploads = f.uploads[]
+            f.invoke(command)
+            @test !isdir(dir)
+            @test _dir(f) != dir
+            @test f.uploads[] > uploads
+            @test read(joinpath(f.d, "stage.out"), String) == "known input\n"
+            @test !_journal(f)["cancel_requested"]
+            @test isempty(readdir(f.target.base))
+        end
+    end
+
+    @testset "receipt recovery cleans up a retained attempt" begin
+        _remote_fixture() do f
+            command = d -> "cd '$d' && echo stage-diagnostics && cat input > output"
+            f.fail_download[] = true
+            @test_throws Exception f.invoke(command)
+            dir = _dir(f)
+            @test _RX._remote_state(f.ssh, f.target, dir) == "exit 0"
+            out = joinpath(f.d, "stage.out")
+            write(out, "known input\n")
+            journal = _journal(f)
+            journal["retrieved"] = Dict(abspath(out) => _RX._digest(out))
+            write(joinpath(f.d, "stage.log.remote.json"), JSON3.write(journal))
+            # The local log is gone, as after a crash before it reached disk.
+            # Cleanup removes the remote copy, so recovery must replay it first.
+            log = joinpath(f.d, "stage.log")
+            rm(log; force=true)
+            f.fail_download[] = false
+            uploads = f.uploads[]
+            f.invoke(command)
+            @test f.uploads[] == uploads
+            @test read(out, String) == "known input\n"
+            @test occursin("stage-diagnostics", read(log, String))
+            @test !isdir(dir)
+            @test haskey(_journal(f), "retrieved")
+        end
+    end
+
+    @testset "a failed stale-attempt sweep cannot fail a finished stage" begin
+        _remote_fixture() do f
+            f.fail_sweep[] = true
+            f.invoke(d -> "cd '$d' && cat input > output")
+            @test read(joinpath(f.d, "stage.out"), String) == "known input\n"
+            @test isempty(readdir(f.target.base))
+        end
+    end
+
+    @testset "an owned transport is closed even when the connection refuses" begin
+        _remote_fixture() do f
+            log = joinpath(f.d, "own.log")
+            _refusing_transport() do
+                @test_throws Exception begin
+                    _RX.run_remote(_ -> nothing, f.target,
+                        d -> "cd '$d' && cat input > output";
+                        inputs=[f.input => "input"],
+                        outputs=["output" => joinpath(f.d, "own.out")],
+                        log_path=log, stage="denoise", _poll_interval=0.02)
+                end
+            end
+            @test isfile(log * ".remote.json")
+        end
+    end
+
+    @testset "a superseded journal closes the old master it opened" begin
+        _remote_fixture() do f
+            for cancel_requested in (false, true)
+                tag = string(cancel_requested)
+                log = joinpath(f.d, "superseded-" * tag * ".log")
+                journal_path = log * ".remote.json"
+                old_identity = _RX._identity(f.target, "denoise", d -> "cd '$d' && true",
+                    [f.input => "input"],
+                    ["output" => joinpath(f.d, "old-" * tag * ".out")],
+                    Pair{String,String}[], ["old"])
+                journal = Dict{String,Any}(
+                    "identity" => old_identity,
+                    "directory" => joinpath(f.target.base,
+                        "mm_" * "0"^64 * "_00000000-0000-4000-8000-000000000000"),
+                    "identity_file" => nothing,
+                    "failed" => false,
+                    "cancel_requested" => cancel_requested)
+                write(journal_path, JSON3.write(journal))
+                original_journal = read(journal_path)
+                _owned_transport(f) do transport
+                    @test_throws ProcessFailedException begin
+                        _RX.run_remote(_ -> nothing, f.target,
+                            d -> "cd '$d' && cat input > output";
+                            inputs=[f.input => "input"],
+                            outputs=["output" => joinpath(f.d, "superseded-" * tag * ".out")],
+                            log_path=log, stage="denoise", context=["new"],
+                            _poll_interval=0.02)
+                    end
+                    calls = split.(readlines(transport.events), '\t'; limit=3)
+                    # The first request uses the old journal's master. It opens
+                    # that master, fails its status probe, then must close it.
+                    @test calls[1][1] == "opened"
+                    old_ctl = calls[1][2]
+                    @test calls[2][1:2] == ["request", old_ctl]
+                    @test occursin("ControlMaster=auto", calls[2][3])
+                    @test occursin(journal["directory"], calls[2][3])
+                    @test occursin("worker.sh' status", calls[2][3])
+                    @test calls[3] == ["closed", old_ctl]
+                    @test isempty(readdir(transport.masters))
+                    if cancel_requested
+                        @test length(calls) == 3
+                        @test read(journal_path) == original_journal
+                    end
+                end
+                if !cancel_requested
+                    @test any(n -> startswith(n, basename(journal_path) * ".superseded-"),
+                              readdir(f.d))
+                end
+            end
+        end
+    end
+
     @testset "path traversal and reserved control files are refused" begin
         for path in ("/absolute", "../outside", "a/../../b", "a//b", ".metamanifold/pid", "a\nb", "a\\b")
             @test_throws ErrorException _RX._relative(path)
diff --git a/test/unit/test_phylogeny.jl b/test/unit/test_phylogeny.jl
index 22f03b1b..e41a148a 100644
--- a/test/unit/test_phylogeny.jl
+++ b/test/unit/test_phylogeny.jl
@@ -59,6 +59,56 @@ _ph_factory() = YAML.load_file(joinpath(@__DIR__, "..", "..", "config", "default
     @test !("-redo" in split(remote_command))
 end
 
+@testset "run_workflow keeps a retained remote log and clears only unverified outputs" begin
+    # Both halves of the new branch run before a step can fail, so a step whose
+    # command cannot run is enough to observe what each half did to the stage
+    # log, the outputs and the per-step key.
+    mktempdir() do root
+        dir = joinpath(root, "reference")
+        files = _PH.reference_files(dir)
+        cfg = _ph_factory()
+        align_dir = mkpath(dirname(files["reference.aln.fasta"]))
+        trim_dir = mkpath(dirname(files["reference.trim.fasta"]))
+        write(files["reference.aln.fasta"], ">a\nACGT\n")
+        write(files["references.fasta"], ">a\nACGT\n")
+
+        # Local: reset the log and clear the old outputs, then fail on a step
+        # with no tool behind it (nothing must be installed for this test).
+        local_step = _PH.Step("trim", ("reference", "trim"), nothing, String[],
+            ["reference.aln.fasta"],
+            ["reference.trim.fasta", "reference.columns"], String[])
+        local_log = joinpath(dir, "logs", "trim.log")
+        mkpath(dirname(local_log))
+        write(local_log, "sentinel that a local rerun must truncate")
+        write(files["reference.trim.fasta"], "stale")
+        write(files["reference.columns"], "stale")
+        write(_PH._key_path(files, local_step), "stale key")
+        @test_throws Exception _PH.run_workflow([local_step], dir, files, cfg)
+        @test !occursin("sentinel", read(local_log, String))
+        @test !isfile(files["reference.trim.fasta"])
+        @test !isfile(_PH._key_path(files, local_step))
+        @test _PH.read_status(dir)["state"] == "failed"
+
+        # Remote: a retained journal means the runner may replay the remote
+        # diagnostics, so the log must not be truncated; only the key is
+        # cleared before the runner checks the receipt under its owner lock.
+        remote_step = _PH.Step("align", ("reference", "align"), "phylogeny_align",
+            ["mafft"], ["references.fasta"], ["reference.aln.fasta"], String[])
+        remote_log = joinpath(dir, "logs", "align.log")
+        mkpath(dirname(remote_log))
+        write(remote_log, "sentinel that a retained journal must keep")
+        write(remote_log * ".remote.json", "{}")
+        write(_PH._key_path(files, remote_step), "stale key")
+        cfg["remote"] = Dict("host" => "127.0.0.1",
+            "staging_dir" => mkpath(joinpath(root, "remote")),
+            "stages" => ["phylogeny_align"])
+        @test_throws Exception _PH.run_workflow([remote_step], dir, files, cfg)
+        @test occursin("sentinel", read(remote_log, String))
+        @test !isfile(_PH._key_path(files, remote_step))
+        @test _PH.read_status(dir)["state"] == "failed"
+    end
+end
+
 @testset "Remote phylogeny clears outputs without a matching verified receipt" begin
     if Sys.islinux()
         for changed in (:settings, :input, :output)
diff --git a/test/unit/test_remote_stages.jl b/test/unit/test_remote_stages.jl
index 508d3c53..bed89158 100644
--- a/test/unit/test_remote_stages.jl
+++ b/test/unit/test_remote_stages.jl
@@ -297,6 +297,27 @@ const _R = MetaManifold.DADA2
         @test isempty(errors)
     end
 
+    @testset "verify_remote_sizes quotes a remote path with an apostrophe" begin
+        # `_sq` escapes the single quote in the name, so `wc -c` sees one path
+        # and the byte count is compared against the right local file.
+        mktempdir() do d
+            staging = mkpath(joinpath(d, "staging"))
+            rel = "sample's reads.fastq.gz"
+            local_path = joinpath(d, rel)
+            write(local_path, "0123456789")
+            write(joinpath(staging, rel), "0123456789")
+            ssh = (_, cmd) -> `bash -c $cmd`
+            emitted = String[]
+            rx = MetaManifold.RemoteExec
+            rx.verify_remote_sizes(ssh, (; host="fixture"), staging,
+                [local_path => rel], msg -> push!(emitted, msg))
+            @test any(m -> occursin("1 uploaded file", m), emitted)
+            write(joinpath(staging, rel), "0123")
+            @test_throws ErrorException rx.verify_remote_sizes(ssh, (; host="fixture"),
+                staging, [local_path => rel], _ -> nothing)
+        end
+    end
+
     @testset "every offloadable stage has a remote script" begin
         # REMOTE_STAGES is what a config may name; a stage listed there with no
         # script behind it would fail only once someone offloaded it.