diff --git a/backend/include/backend/sftp/operation.hpp b/backend/include/backend/sftp/operation.hpp
index bfd03ea6..80e739eb 100644
--- a/backend/include/backend/sftp/operation.hpp
+++ b/backend/include/backend/sftp/operation.hpp
@@ -22,6 +22,18 @@
 #include <system_error>
 #include <vector>
 
+template <>
+struct SecureShell::UnfinishedTaskResult<SharedData::OperationError>
+{
+    SharedData::OperationError operator()() const
+    {
+        return SharedData::OperationError{
+            .type = SharedData::OperationErrorType::ChannelClosed,
+            .extraInfo = "The task did not finish, the SFTP session is closed or its thread stopped",
+        };
+    }
+};
+
 class Operation
 {
   public:
diff --git a/backend/source/backend/sftp/operation_queue.cpp b/backend/source/backend/sftp/operation_queue.cpp
index 1e2a25ef..ddc3591f 100644
--- a/backend/source/backend/sftp/operation_queue.cpp
+++ b/backend/source/backend/sftp/operation_queue.cpp
@@ -1049,6 +1049,7 @@ bool OperationQueue::workQueue(std::deque<std::pair<Ids::OperationId, std::uniqu
     // every eligible op's workInStrand() call. Ops that do not use the strand run on the
     // caller afterwards: their work() blocks on that very thread and would deadlock inside it.
     SecureShell::ProcessingStrand* strand = nullptr;
+    std::size_t strandIndex = 0;
     for (const auto& item : eligible)
     {
         auto* operation = queue[item.index].second.get();
@@ -1056,7 +1057,10 @@ bool OperationQueue::workQueue(std::deque<std::pair<Ids::OperationId, std::uniqu
         if (!operation->hasWake())
             operation->setWake(wake_);
         if (operation->usesStrand() && !strand)
+        {
             strand = operation->strand();
+            strandIndex = item.index;
+        }
     }
 
     struct ItemResult
@@ -1103,17 +1107,42 @@ bool OperationQueue::workQueue(std::deque<std::pair<Ids::OperationId, std::uniqu
     std::vector<ItemResult> results;
     if (strand)
     {
-        auto fut = strand->pushPromiseTask(
-            [&runBatch]()
+        using Batch = std::optional<std::vector<ItemResult>>;
+        auto batch = strand
+                         ->pushPromiseTask(
+                             [&runBatch]() -> Batch
+                             {
+                                 return runBatch(true);
+                             },
+                             []() -> Batch
+                             {
+                                 return std::nullopt;
+                             }
+                         )
+                         .get();
+        // The processing thread no longer runs tasks, the operations that need it fail one per cycle.
+        if (!batch)
+        {
+            results.push_back(
+                {strandIndex,
+                    std::unexpected(
+                        Operation::Error{
+                            .type = Operation::ErrorType::ChannelClosed,
+                            .extraInfo = "The SFTP processing thread no longer runs tasks",
+                        }
+                    )}
+            );
+        }
+        else
+        {
+            results = std::move(batch).value();
+            if (std::ranges::none_of(results, isTerminal))
             {
-                return runBatch(true);
+                auto local = runBatch(false);
+                results.insert(
+                    results.end(), std::make_move_iterator(local.begin()), std::make_move_iterator(local.end())
+                );
             }
-        );
-        results = fut.get();
-        if (std::ranges::none_of(results, isTerminal))
-        {
-            auto local = runBatch(false);
-            results.insert(results.end(), std::make_move_iterator(local.begin()), std::make_move_iterator(local.end()));
         }
     }
     else
diff --git a/ssh/include/ssh/async/processing_strand.hpp b/ssh/include/ssh/async/processing_strand.hpp
index caf95de7..2c4906d0 100644
--- a/ssh/include/ssh/async/processing_strand.hpp
+++ b/ssh/include/ssh/async/processing_strand.hpp
@@ -109,7 +109,8 @@ namespace SecureShell
         }
 
         /**
-         * @brief Pushes a task that makes any further pushes impossible. Also removes all permanent tasks.
+         * @brief Pushes a task that makes any further pushes impossible. Also removes all permanent tasks. The task
+         *        runs even when the thread no longer runs tasks, see ProcessingThread::pushCleanupTask.
          *
          * @param task The task to push.
          */
@@ -117,27 +118,26 @@ namespace SecureShell
         {
             auto* processingThread = finalize();
             if (processingThread != nullptr)
-                processingThread->pushTask(std::move(task));
+                processingThread->pushCleanupTask(std::move(task));
         }
 
         /**
          * @brief Works like pushFinalTask but returns a future for the return value of the task.
          *
-         * @param task The task to run.
+         * @param func The task to run.
+         * @param unfinished Makes the future's value when the task does not finish, also when the strand was finalized
+         *        before. See UnfinishedTaskResult.
          */
-        template <typename FunctionT>
-        auto pushFinalPromiseTask(FunctionT&& func) -> std::future<std::invoke_result_t<std::decay_t<FunctionT>>>
+        template <
+            typename FunctionT,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<FunctionT>>>>
+        auto pushFinalPromiseTask(FunctionT&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<FunctionT>>>
         {
             auto* processingThread = finalize();
             if (processingThread == nullptr)
-            {
-                std::promise<std::invoke_result_t<std::decay_t<FunctionT>>> promise{};
-                promise.set_exception(
-                    std::make_exception_ptr(std::runtime_error("Cannot push task to finalized strand."))
-                );
-                return promise.get_future();
-            }
-            return processingThread->pushPromiseTask(std::forward<FunctionT>(func));
+                return Detail::unfinishedFuture<std::invoke_result_t<std::decay_t<FunctionT>>>(std::move(unfinished));
+            return processingThread->pushCleanupPromiseTask(std::forward<FunctionT>(func), std::move(unfinished));
         }
 
         /**
@@ -164,21 +164,20 @@ namespace SecureShell
          *
          * @tparam Func The type of the task.
          * @param func The task to push.
+         * @param unfinished Makes the future's value when the task does not finish, also when the strand was
+         *        finalized. See UnfinishedTaskResult.
          * @return std::future<std::invoke_result_t<std::decay_t<Func>>> The future of the return value of the task.
          */
-        template <typename Func>
-        auto pushPromiseTask(Func&& func) -> std::future<std::invoke_result_t<std::decay_t<Func>>>
+        template <
+            typename Func,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<Func>>>>
+        auto pushPromiseTask(Func&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<Func>>>
         {
             std::scoped_lock lock(mutex_);
             if (finalized_)
-            {
-                std::promise<std::invoke_result_t<std::decay_t<Func>>> promise{};
-                promise.set_exception(
-                    std::make_exception_ptr(std::runtime_error("Cannot push task to finalized strand."))
-                );
-                return promise.get_future();
-            }
-            return processingThread_->pushPromiseTask(std::forward<Func>(func));
+                return Detail::unfinishedFuture<std::invoke_result_t<std::decay_t<Func>>>(std::move(unfinished));
+            return processingThread_->pushPromiseTask(std::forward<Func>(func), std::move(unfinished));
         }
 
         /**
diff --git a/ssh/include/ssh/async/processing_thread.hpp b/ssh/include/ssh/async/processing_thread.hpp
index 0df4f7a1..66e0f4d1 100644
--- a/ssh/include/ssh/async/processing_thread.hpp
+++ b/ssh/include/ssh/async/processing_thread.hpp
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ssh/async/socket_wait.hpp>
+#include <ssh/async/unfinished_task_result.hpp>
 
 #include <thread>
 #include <atomic>
@@ -12,12 +13,96 @@
 #include <optional>
 #include <future>
 #include <memory>
+#include <type_traits>
+#include <utility>
 #include <vector>
 
 namespace SecureShell
 {
     class ProcessingStrand;
 
+    namespace Detail
+    {
+        /**
+         * @brief The promise of a promise task, shared by the copies of the task or of the callback that completes it.
+         *        When the last copy is gone and the task did not finish, the promise gets the unfinished result.
+         */
+        template <typename ResultT, typename UnfinishedT>
+        class PromiseTaskState
+        {
+          public:
+            explicit PromiseTaskState(UnfinishedT unfinished)
+                : unfinished_{std::move(unfinished)}
+            {}
+            ~PromiseTaskState()
+            {
+                if (!finished_)
+                    complete(promise_, unfinished_);
+            }
+            PromiseTaskState(PromiseTaskState const&) = delete;
+            PromiseTaskState& operator=(PromiseTaskState const&) = delete;
+            PromiseTaskState(PromiseTaskState&&) = delete;
+            PromiseTaskState& operator=(PromiseTaskState&&) = delete;
+
+            std::future<ResultT> future()
+            {
+                return promise_.get_future();
+            }
+
+            template <typename FunctionT>
+            void run(FunctionT& function)
+            {
+                complete(promise_, function);
+                finished_ = true;
+            }
+
+            /**
+             * @brief Sets @p promise to what @p produce returns.
+             */
+            template <typename FunctionT>
+            static void complete(std::promise<ResultT>& promise, FunctionT& produce)
+            {
+                if constexpr (std::is_void_v<ResultT>)
+                {
+                    produce();
+                    promise.set_value();
+                }
+                else
+                {
+                    promise.set_value(produce());
+                }
+            }
+
+          private:
+            std::promise<ResultT> promise_{};
+            UnfinishedT unfinished_;
+            bool finished_{false};
+        };
+
+        /**
+         * @brief Makes a task of @p func that completes the promise of @p state.
+         */
+        template <typename StateT, typename FunctionT>
+        std::function<void()> makePromiseTask(std::shared_ptr<StateT> state, FunctionT&& func)
+        {
+            return [state = std::move(state), func = std::forward<FunctionT>(func)]() mutable
+            {
+                state->run(func);
+            };
+        }
+
+        /**
+         * @brief A future that already holds the unfinished result, for a task refused before it reached a thread.
+         */
+        template <typename ResultT, typename UnfinishedT>
+        std::future<ResultT> unfinishedFuture(UnfinishedT unfinished)
+        {
+            std::promise<ResultT> promise{};
+            PromiseTaskState<ResultT, UnfinishedT>::complete(promise, unfinished);
+            return promise.get_future();
+        }
+    }
+
     /**
      * @brief A processing thread that can be used to execute tasks sequentially in a separate thread.
      * This allows for complex simultaneous operations to be executed in a single thread.
@@ -92,7 +177,8 @@ namespace SecureShell
         std::expected<void, std::string> watchSocket(std::function<SocketWait::Watched()> watched);
 
         /**
-         * @brief Stops the processing thread. Executes all pending tasks.
+         * @brief Stops the processing thread. Executes all pending tasks, on this thread acting as the processing
+         *        thread. Refuses pushes from now until the next start.
          */
         void stop();
 
@@ -109,37 +195,60 @@ namespace SecureShell
          *
          * @param task The task to push.
          * @return true If the task was pushed.
-         * @return false If the task was empty or the processing thread is shutting down.
+         * @return false If the processing thread was stopped and not started again, or its loop ended on an exception.
          */
         bool pushTask(std::function<void()> task);
 
+        /**
+         * @brief Pushes a task that must run, a strand's final task. It is queued like pushTask while the loop or the
+         *        drain of stop() will run it, and still runs when the loop ends on an exception. Once neither will, it
+         *        runs on the calling thread at once, which acts as the processing thread meanwhile.
+         *
+         * @param task The task to push.
+         */
+        void pushCleanupTask(std::function<void()> task);
+
         /**
          * @brief Pushes a task that has a return value to the processing thread.
          * The return value is then accessible through the returned future.
          *
+         * A push refused while stop() drains completes only once the drain reached it, so the future is still a
+         * barrier for the tasks queued before it.
+         *
          * @param func The function to execute.
+         * @param unfinished Makes the future's value when the task does not finish, see UnfinishedTaskResult.
          * @return std::future<std::invoke_result_t<std::decay_t<Func>>> The future that will contain the return value.
          */
-        template <typename Func>
-        auto pushPromiseTask(Func&& func) -> std::future<std::invoke_result_t<std::decay_t<Func>>>
+        template <
+            typename Func,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<Func>>>>
+        auto pushPromiseTask(Func&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<Func>>>
         {
             using ReturnType = std::invoke_result_t<std::decay_t<Func>>;
-            auto promise = std::make_shared<std::promise<ReturnType>>();
-            pushTask(
-                [promise, func = std::forward<Func>(func)]() mutable
-                {
-                    if constexpr (std::is_void_v<ReturnType>)
-                    {
-                        func();
-                        promise->set_value();
-                    }
-                    else
-                    {
-                        promise->set_value(func());
-                    }
-                }
-            );
-            return promise->get_future();
+            auto state = std::make_shared<Detail::PromiseTaskState<ReturnType, UnfinishedFunc>>(std::move(unfinished));
+            auto future = state->future();
+            queuePromiseTask(Detail::makePromiseTask(std::move(state), std::forward<Func>(func)));
+            return future;
+        }
+
+        /**
+         * @brief Works like pushCleanupTask but returns a future for the return value of the task.
+         *
+         * @param func The function to execute.
+         * @param unfinished Makes the future's value when the task throws, see UnfinishedTaskResult.
+         */
+        template <
+            typename Func,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<Func>>>>
+        auto pushCleanupPromiseTask(Func&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<Func>>>
+        {
+            using ReturnType = std::invoke_result_t<std::decay_t<Func>>;
+            auto state = std::make_shared<Detail::PromiseTaskState<ReturnType, UnfinishedFunc>>(std::move(unfinished));
+            auto future = state->future();
+            pushCleanupTask(Detail::makePromiseTask(std::move(state), std::forward<Func>(func)));
+            return future;
         }
 
         /**
@@ -221,6 +330,17 @@ namespace SecureShell
       private:
         void run(std::chrono::milliseconds const& pollInterval);
 
+        /**
+         * @brief Queues a promise task like pushTask. While stop() drains it is queued refused instead: the drain
+         *        drops it in order, so its future completes after the tasks queued before it.
+         */
+        void queuePromiseTask(std::function<void()> task);
+
+        /**
+         * @brief Runs a task outside the loop, where a throwing task must not keep the others from running.
+         */
+        static void runContained(std::function<void()> const& task);
+
         /**
          * @brief What the permanent tasks of one cycle need from the wait after it.
          */
@@ -255,11 +375,45 @@ namespace SecureShell
             std::optional<std::chrono::milliseconds> pollInterval;
         };
 
+        /**
+         * @brief What a queued task is.
+         */
+        enum class TaskKind
+        {
+            Regular,
+            /** @brief Still runs when the loop ends on an exception. */
+            Cleanup,
+            /** @brief A promise task pushed while stop() drains: dropped, not run, when its turn comes. */
+            Refused,
+        };
+
+        struct QueuedTask
+        {
+            std::function<void()> run;
+            TaskKind kind{TaskKind::Regular};
+        };
+
+        /**
+         * @brief What happens to pushed tasks.
+         */
+        enum class Intake
+        {
+            /** @brief Tasks are queued: the thread was constructed or started. */
+            Open,
+            /** @brief stop() is running: tasks are refused, cleanup tasks are queued for its drain to run. */
+            Draining,
+            /** @brief Stopped, or the loop threw: tasks are refused, cleanup tasks run on the caller. */
+            Closed,
+        };
+
       private:
         std::thread thread_{};
+        /** @brief Held while cleanup tasks run outside the loop and while intake_ leaves Closed. Taken first. */
+        std::recursive_mutex cleanupMutex_{};
         mutable std::recursive_mutex taskMutex_{};
         std::atomic<bool> running_ = false;
-        std::atomic<bool> shuttingDown_ = false;
+        /** @brief Under taskMutex_. */
+        Intake intake_{Intake::Open};
         std::atomic_bool permanentTasksAvailable_ = false;
         std::atomic<int> permanentTaskIdCounter_ = 0;
         std::atomic<std::thread::id> processingThreadId_{};
@@ -278,7 +432,7 @@ namespace SecureShell
         std::function<SocketWait::Watched()> watchedSocket_{};
         std::unique_ptr<SocketWait> socketWait_{};
 
-        std::deque<std::function<void()>> tasks_{};
+        std::deque<QueuedTask> tasks_{};
         std::map<PermanentTaskId, PermanentTask> permanentTasks_{};
     };
 }
\ No newline at end of file
diff --git a/ssh/include/ssh/async/unfinished_task_result.hpp b/ssh/include/ssh/async/unfinished_task_result.hpp
new file mode 100644
index 00000000..ca5aa817
--- /dev/null
+++ b/ssh/include/ssh/async/unfinished_task_result.hpp
@@ -0,0 +1,39 @@
+#pragma once
+
+#include <expected>
+
+namespace SecureShell
+{
+    /**
+     * @brief Makes what the future of a promise task holds when the task does not finish: its push was refused, the
+     *        thread dropped it, or it threw. Defined for void (nothing), bool (false) and std::expected of an error
+     *        that defines it (that error), other results pass their own.
+     */
+    template <typename ResultT>
+    struct UnfinishedTaskResult;
+
+    template <>
+    struct UnfinishedTaskResult<void>
+    {
+        void operator()() const noexcept
+        {}
+    };
+
+    template <>
+    struct UnfinishedTaskResult<bool>
+    {
+        bool operator()() const noexcept
+        {
+            return false;
+        }
+    };
+
+    template <typename ValueT, typename ErrorT>
+    struct UnfinishedTaskResult<std::expected<ValueT, ErrorT>>
+    {
+        std::expected<ValueT, ErrorT> operator()() const
+        {
+            return std::unexpected(UnfinishedTaskResult<ErrorT>{}());
+        }
+    };
+}
diff --git a/ssh/include/ssh/sftp_error.hpp b/ssh/include/ssh/sftp_error.hpp
index af03b01b..cda81270 100644
--- a/ssh/include/ssh/sftp_error.hpp
+++ b/ssh/include/ssh/sftp_error.hpp
@@ -1,5 +1,7 @@
 #pragma once
 
+#include <ssh/async/unfinished_task_result.hpp>
+
 #include <fmt/format.h>
 
 #include <string>
@@ -71,4 +73,16 @@ namespace SecureShell
             );
         }
     };
+
+    template <>
+    struct UnfinishedTaskResult<SftpError>
+    {
+        SftpError operator()() const
+        {
+            return SftpError{
+                .message = "The task did not finish, the SFTP session is closed or its thread stopped",
+                .wrapperError = WrapperErrors::TaskPushFailed,
+            };
+        }
+    };
 }
\ No newline at end of file
diff --git a/ssh/include/ssh/sftp_session.hpp b/ssh/include/ssh/sftp_session.hpp
index 6c6b27c5..9a20c44d 100644
--- a/ssh/include/ssh/sftp_session.hpp
+++ b/ssh/include/ssh/sftp_session.hpp
@@ -57,10 +57,13 @@ namespace SecureShell
             strand_->pushTask(std::forward<FunctionT>(func));
         }
 
-        template <typename FunctionT>
-        auto performPromise(FunctionT&& func) -> std::future<std::invoke_result_t<std::decay_t<FunctionT>>>
+        template <
+            typename FunctionT,
+            typename UnfinishedFunc = UnfinishedTaskResult<std::invoke_result_t<std::decay_t<FunctionT>>>>
+        auto performPromise(FunctionT&& func, UnfinishedFunc unfinished = {})
+            -> std::future<std::invoke_result_t<std::decay_t<FunctionT>>>
         {
-            return strand_->pushPromiseTask(std::forward<FunctionT>(func));
+            return strand_->pushPromiseTask(std::forward<FunctionT>(func), std::move(unfinished));
         }
 
         /**
diff --git a/ssh/source/ssh/async/processing_thread.cpp b/ssh/source/ssh/async/processing_thread.cpp
index 49f1a1b3..c923224a 100644
--- a/ssh/source/ssh/async/processing_thread.cpp
+++ b/ssh/source/ssh/async/processing_thread.cpp
@@ -7,6 +7,33 @@
 
 namespace SecureShell
 {
+    namespace
+    {
+        /**
+         * @brief Makes the calling thread act as the processing thread until destroyed, for tasks run outside the loop.
+         */
+        class ActingProcessingThread
+        {
+          public:
+            explicit ActingProcessingThread(std::atomic<std::thread::id>& processingThreadId)
+                : processingThreadId_{processingThreadId}
+                , previous_{processingThreadId.exchange(std::this_thread::get_id())}
+            {}
+            ~ActingProcessingThread()
+            {
+                processingThreadId_.store(previous_);
+            }
+            ActingProcessingThread(ActingProcessingThread const&) = delete;
+            ActingProcessingThread& operator=(ActingProcessingThread const&) = delete;
+            ActingProcessingThread(ActingProcessingThread&&) = delete;
+            ActingProcessingThread& operator=(ActingProcessingThread&&) = delete;
+
+          private:
+            std::atomic<std::thread::id>& processingThreadId_;
+            std::thread::id previous_;
+        };
+    }
+
     ProcessingThread::ProcessingThread()
     {}
     ProcessingThread::~ProcessingThread()
@@ -24,10 +51,11 @@ namespace SecureShell
                 "ProcessingThread::start called while a previous thread is still joinable; call stop() first."
             );
         {
+            std::lock_guard cleanupLock{cleanupMutex_};
             std::lock_guard lock{taskMutex_};
             running_ = true;
+            intake_ = Intake::Open;
         }
-        shuttingDown_ = false;
         std::promise<void> awaitThreadStart{};
         thread_ = std::thread(
             [this, &awaitThreadStart, pollInterval]
@@ -41,10 +69,12 @@ namespace SecureShell
     }
     void ProcessingThread::stop()
     {
-        shuttingDown_ = true;
         bool interruptSocketWait = false;
         {
+            // Waits for cleanup tasks that run on a caller, the drain must not run alongside them.
+            std::lock_guard cleanupLock{cleanupMutex_};
             std::lock_guard lock{taskMutex_};
+            intake_ = Intake::Draining;
             running_ = false;
             wakeRequested_ = true;
             interruptSocketWait = waitingOnSocket_;
@@ -53,14 +83,36 @@ namespace SecureShell
         if (thread_.joinable())
             thread_.join();
 
-        // execute all pending tasks in FIFO order, matching the run loop:
-        std::lock_guard lock{taskMutex_};
+        // execute all pending tasks in FIFO order, matching the run loop, and outside the lock so pushes meanwhile are
+        // refused instead of waiting for the drain. Intake stays Draining until the end, so no cleanup task runs on a
+        // caller meanwhile.
+        std::unique_lock lock{taskMutex_};
+        processingThreadId_.store(std::this_thread::get_id());
         while (!tasks_.empty())
         {
-            tasks_.front()();
-            tasks_.pop_front();
+            {
+                auto task = std::move(tasks_.front());
+                tasks_.pop_front();
+                lock.unlock();
+                if (task.kind != TaskKind::Refused)
+                    runContained(task.run);
+            }
+            lock.lock();
+        }
+        // The id of the joined thread may be reused.
+        processingThreadId_.store({});
+        intake_ = Intake::Closed;
+    }
+    void ProcessingThread::runContained(std::function<void()> const& task)
+    {
+        try
+        {
+            task();
+        }
+        catch (...)
+        {
+            // Outside the loop no task ends anything, a promise task completes with its unfinished result.
         }
-        shuttingDown_ = false;
     }
     bool ProcessingThread::pushTask(std::function<void()> task)
     {
@@ -69,20 +121,66 @@ namespace SecureShell
             throw std::invalid_argument("Task must not be empty.");
         }
 
-        if (shuttingDown_)
+        bool interruptSocketWait = false;
         {
-            return false;
+            std::lock_guard lock{taskMutex_};
+            if (intake_ != Intake::Open)
+            {
+                return false;
+            }
+            tasks_.push_back({.run = std::move(task)});
+            wakeRequested_ = true;
+            interruptSocketWait = waitingOnSocket_;
         }
-
+        notifyWake(interruptSocketWait);
+        return true;
+    }
+    void ProcessingThread::queuePromiseTask(std::function<void()> task)
+    {
         bool interruptSocketWait = false;
         {
             std::lock_guard lock{taskMutex_};
-            tasks_.push_back(std::move(task));
+            if (intake_ == Intake::Closed)
+                return;
+            if (intake_ == Intake::Draining)
+            {
+                tasks_.push_back({.run = std::move(task), .kind = TaskKind::Refused});
+                return;
+            }
+            tasks_.push_back({.run = std::move(task)});
             wakeRequested_ = true;
             interruptSocketWait = waitingOnSocket_;
         }
         notifyWake(interruptSocketWait);
-        return true;
+    }
+    void ProcessingThread::pushCleanupTask(std::function<void()> task)
+    {
+        if (!task)
+        {
+            throw std::invalid_argument("Task must not be empty.");
+        }
+
+        // Held while the task runs here, so cleanup tasks run one at a time and never alongside a restarted thread.
+        std::lock_guard cleanupLock{cleanupMutex_};
+        bool queued = false;
+        bool interruptSocketWait = false;
+        {
+            std::lock_guard lock{taskMutex_};
+            if (intake_ != Intake::Closed)
+            {
+                tasks_.push_back({.run = std::move(task), .kind = TaskKind::Cleanup});
+                wakeRequested_ = true;
+                interruptSocketWait = waitingOnSocket_;
+                queued = true;
+            }
+        }
+        if (queued)
+        {
+            notifyWake(interruptSocketWait);
+            return;
+        }
+        const ActingProcessingThread acting{processingThreadId_};
+        runContained(task);
     }
     void ProcessingThread::wake()
     {
@@ -147,15 +245,14 @@ namespace SecureShell
             throw std::invalid_argument("Task must not be empty.");
         }
 
-        if (shuttingDown_)
-        {
-            return {false, PermanentTaskId{-1}};
-        }
-
         PermanentTaskId id{-1};
         bool interruptSocketWait = false;
         {
             std::lock_guard lock{taskMutex_};
+            if (intake_ != Intake::Open)
+            {
+                return {false, PermanentTaskId{-1}};
+            }
             ++permanentTaskIdCounter_;
             id = PermanentTaskId{permanentTaskIdCounter_};
             permanentTasks_.insert({id, PermanentTask{.run = std::move(task), .pollInterval = pollInterval}});
@@ -215,7 +312,8 @@ namespace SecureShell
                     }
                 );
                 lock.unlock();
-                return promise->get_future().wait_for(std::chrono::seconds{5}) == std::future_status::ready;
+                auto removed = promise->get_future();
+                return removed.wait_for(std::chrono::seconds{5}) == std::future_status::ready && removed.get();
             }
         }
 
@@ -287,7 +385,7 @@ namespace SecureShell
                             needs.awaitsInput = true;
                             needs.inputFallback = std::min(needs.inputFallback, lookAgain);
                         }
-                        return running_ && !shuttingDown_;
+                        return running_.load();
                     };
                     bool keepGoing = true;
                     for (auto task = rotatedBegin; keepGoing && task != permaTasksMoved.end(); ++task)
@@ -328,7 +426,7 @@ namespace SecureShell
                 }
 
                 {
-                    std::vector<std::function<void()>> tasks{};
+                    std::vector<QueuedTask> tasks{};
                     {
                         std::lock_guard lock(taskMutex_);
                         tasks.reserve(std::min(tasks_.size(), maximumTasksProcessableAtOnce));
@@ -346,14 +444,17 @@ namespace SecureShell
                     {
                         // Task is checked before adding, shouldnt possibly be empty:
 #ifndef NDEBUG
-                        if (!task)
+                        if (!task.run)
                         {
                             throw std::runtime_error("Task must not be empty.");
                         }
 #endif
+                        // Pushed while stop() drains, completed when the batch is gone.
+                        if (task.kind == TaskKind::Refused)
+                            continue;
                         try
                         {
-                            task();
+                            task.run();
                         }
                         catch (...)
                         {
@@ -371,8 +472,25 @@ namespace SecureShell
         }
         catch (...)
         {
-            std::lock_guard lock{taskMutex_};
-            running_ = false;
+            // Cleanup tasks pushed from now on run on their caller, after the queued ones ran here.
+            std::lock_guard cleanupLock{cleanupMutex_};
+            std::deque<QueuedTask> pending{};
+            {
+                std::lock_guard lock{taskMutex_};
+                running_ = false;
+                // stop() drains what is queued once it joined this thread, and closes intake afterwards.
+                if (intake_ == Intake::Draining)
+                    return;
+                intake_ = Intake::Closed;
+                pending = std::exchange(tasks_, {});
+            }
+            // Queued tasks would run only when stop() drains, which the threads waiting for them may never reach.
+            // Dropped promise tasks complete with their unfinished result, cleanup tasks still run.
+            for (auto& task : pending)
+            {
+                if (task.kind == TaskKind::Cleanup)
+                    runContained(task.run);
+            }
         }
     }
     void ProcessingThread::waitForWork(CycleNeeds const& needs)
@@ -431,12 +549,13 @@ namespace SecureShell
         }
         if (!withinProcessingThread() && running)
         {
-            return pushPromiseTask(
-                       []()
-                       {
-                           return true;
-                       }
-                   ).wait_for(maxWait) == std::future_status::ready;
+            auto cycled = pushPromiseTask(
+                []()
+                {
+                    return true;
+                }
+            );
+            return cycled.wait_for(maxWait) == std::future_status::ready && cycled.get();
         }
         return false;
     }
diff --git a/ssh/source/ssh/file_stream.cpp b/ssh/source/ssh/file_stream.cpp
index 050db1e8..50002723 100644
--- a/ssh/source/ssh/file_stream.cpp
+++ b/ssh/source/ssh/file_stream.cpp
@@ -562,22 +562,33 @@ namespace SecureShell
     std::future<std::expected<std::size_t, SftpError>>
     FileStream::readAll(std::function<bool(std::string_view data)> onReadFunc)
     {
+        using ReadResult = std::expected<std::size_t, SftpError>;
         struct ReadState : public std::enable_shared_from_this<ReadState>
         {
             FileStream& stream;
             BufferLease buffer;
             std::function<bool(std::string_view data)> callback;
-            std::promise<std::expected<std::size_t, SftpError>> promise;
+            // A read that is refused, never pushed or dropped leaves the unfinished error when the state is gone.
+            Detail::PromiseTaskState<ReadResult, UnfinishedTaskResult<ReadResult>> completion;
             std::size_t totalRead;
 
             ReadState(FileStream& stream, BufferLease buffer, std::function<bool(std::string_view data)> callback)
                 : stream{stream}
                 , buffer{std::move(buffer)}
                 , callback{std::move(callback)}
-                , promise{}
+                , completion{UnfinishedTaskResult<ReadResult>{}}
                 , totalRead{0}
             {}
 
+            void finish(ReadResult result)
+            {
+                auto finished = [&result]()
+                {
+                    return std::move(result);
+                };
+                completion.run(finished);
+            }
+
             void onRead(std::make_signed_t<std::size_t> amount)
             {
                 // The callback runs before the promise is set: the waiting caller may destroy what it captured.
@@ -585,14 +596,14 @@ namespace SecureShell
                 {
                     const auto error = stream.lastError();
                     callback({});
-                    promise.set_value(std::unexpected(error));
+                    finish(std::unexpected(error));
                     return;
                 }
 
                 if (amount == 0)
                 {
                     callback({});
-                    promise.set_value(totalRead);
+                    finish(totalRead);
                     return;
                 }
 
@@ -600,7 +611,7 @@ namespace SecureShell
 
                 if (!callback({buffer.data(), static_cast<std::size_t>(amount)}))
                 {
-                    promise.set_value(totalRead);
+                    finish(totalRead);
                     return;
                 }
 
@@ -614,7 +625,7 @@ namespace SecureShell
                     {
                         if (!state->stream.file_)
                         {
-                            state->promise.set_value(
+                            state->finish(
                                 std::unexpected(
                                     SftpError{
                                         .message = "File is null",
@@ -661,8 +672,9 @@ namespace SecureShell
         }
 
         auto state = std::make_shared<ReadState>(*this, std::move(lease), std::move(onReadFunc));
+        auto future = state->completion.future();
         state->doRead();
-        return state->promise.get_future();
+        return future;
     }
     std::size_t FileStream::writeLengthLimit() const
     {
@@ -681,7 +693,11 @@ namespace SecureShell
             [this, toWrite, onWriteComplete = std::move(onWriteComplete)]()
             {
                 if (!file_)
-                    return;
+                {
+                    return onWriteComplete(
+                        std::unexpected(SftpError{.message = "File is null", .wrapperError = WrapperErrors::FileNull})
+                    );
+                }
 
                 const auto written =
                     sftp_write(file_.get(), toWrite.data(), std::min(toWrite.size(), writeLengthLimit()));
@@ -753,16 +769,24 @@ namespace SecureShell
             );
         }
 
-        // Write in chunks for large data:
-        auto promise = std::make_shared<std::promise<std::expected<void, SftpError>>>();
+        // Write in chunks for large data. A chunk that is refused, never pushed or dropped leaves the unfinished error:
+        using WriteResult = std::expected<void, SftpError>;
+        auto state = std::make_shared<Detail::PromiseTaskState<WriteResult, UnfinishedTaskResult<WriteResult>>>(
+            UnfinishedTaskResult<WriteResult>{}
+        );
+        auto future = state->future();
         writePart(
             data,
-            [promise](std::expected<void, SftpError>&& result)
+            [state](WriteResult&& result)
             {
-                promise->set_value(std::move(result));
+                auto finished = [&result]()
+                {
+                    return std::move(result);
+                };
+                state->run(finished);
             }
         );
-        return promise->get_future();
+        return future;
     }
     sftp_file FileStream::release()
     {
diff --git a/ssh/source/ssh/session.cpp b/ssh/source/ssh/session.cpp
index faa9e0cb..9c85e0ab 100644
--- a/ssh/source/ssh/session.cpp
+++ b/ssh/source/ssh/session.cpp
@@ -173,10 +173,9 @@ namespace SecureShell
 
     std::future<std::expected<std::weak_ptr<Channel>, int>> Session::createPtyChannel(PtyCreationOptions options)
     {
-        auto promise = std::make_shared<std::promise<std::expected<std::weak_ptr<Channel>, int>>>();
-
-        processingThread_.pushTask(
-            [this, options = std::move(options), promise]() mutable
+        using Created = std::expected<std::weak_ptr<Channel>, int>;
+        return processingThread_.pushPromiseTask(
+            [this, options = std::move(options)]() mutable -> Created
             {
                 auto ptyChannel = std::make_unique<ssh::Channel>(session_);
                 auto& channel = *ptyChannel;
@@ -227,7 +226,7 @@ namespace SecureShell
 
                 if (result.result != SSH_OK)
                 {
-                    return promise->set_value(std::unexpected(session_.getErrorCode()));
+                    return std::unexpected(session_.getErrorCode());
                 }
 
                 auto sharedChannel =
@@ -236,31 +235,29 @@ namespace SecureShell
                     hiddenChannel_ = sharedChannel;
                 else
                     channels_.push_back(sharedChannel);
-                return promise->set_value(sharedChannel);
+                return sharedChannel;
+            },
+            []() -> Created
+            {
+                return std::unexpected(SSH_ERROR);
             }
         );
-
-        return promise->get_future();
     }
 
     std::future<std::expected<std::weak_ptr<SftpSession>, SftpError>> Session::createSftpSession()
     {
-        auto promise = std::make_shared<std::promise<std::expected<std::weak_ptr<SftpSession>, SftpError>>>();
-
-        processingThread_.pushTask(
-            [this, promise]() -> void
+        return processingThread_.pushPromiseTask(
+            [this]() -> std::expected<std::weak_ptr<SftpSession>, SftpError>
             {
                 auto sftp = sftp_new(session_.getCSession());
                 if (sftp == nullptr)
                 {
-                    return promise->set_value(
-                        std::unexpected(
-                            SftpError{
-                                .message = session_.getError(),
-                                .sshError = session_.getErrorCode(),
-                                .sftpError = 0,
-                            }
-                        )
+                    return std::unexpected(
+                        SftpError{
+                            .message = session_.getError(),
+                            .sshError = session_.getErrorCode(),
+                            .sftpError = 0,
+                        }
                     );
                 }
 
@@ -273,17 +270,15 @@ namespace SecureShell
                         .sftpError = sftp_get_error(sftp),
                     };
                     sftp_free(sftp);
-                    return promise->set_value(std::unexpected(error));
+                    return std::unexpected(error);
                 }
 
                 // Kept before the caller hears of it, who may shut the session down and walk the list right away.
                 auto sftpSession = std::make_shared<SftpSession>(this, processingThread_.createStrand(), sftp);
                 sftpSessions_.push_back(sftpSession);
-                promise->set_value(sftpSession);
+                return sftpSession;
             }
         );
-
-        return promise->get_future();
     }
 
     std::expected<std::unique_ptr<Session>, std::string> makeSession(
diff --git a/ssh/test/ssh/test_processing_thread.hpp b/ssh/test/ssh/test_processing_thread.hpp
index 373e66e2..95765383 100644
--- a/ssh/test/ssh/test_processing_thread.hpp
+++ b/ssh/test/ssh/test_processing_thread.hpp
@@ -15,10 +15,12 @@
 #include <algorithm>
 #include <array>
 #include <atomic>
+#include <expected>
 #include <functional>
 #include <future>
 #include <latch>
 #include <stdexcept>
+#include <string>
 #include <thread>
 #include <utility>
 #include <vector>
@@ -125,12 +127,199 @@ namespace SecureShell::Test
         ASSERT_EQ(ProcessingThread::maximumTasksProcessableAtOnce * 3, counter.load());
     }
 
-    TEST_F(ProcessingThreadTest, CanPushTaskAfterStop)
+    TEST_F(ProcessingThreadTest, CannotPushAfterStopUntilStartedAgain)
+    {
+        std::atomic_int ran = 0;
+        const auto countRun = [&ran]
+        {
+            ++ran;
+        };
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        processingThread.stop();
+
+        EXPECT_FALSE(processingThread.pushTask(countRun));
+        EXPECT_FALSE(processingThread
+                .pushPermanentTask(
+                    [](auto)
+                    {
+                        return true;
+                    }
+                )
+                .first);
+        processingThread.stop();
+        EXPECT_EQ(ran.load(), 0);
+
+        processingThread.start(std::chrono::milliseconds{1});
+        Awaiter awaiter{};
+        EXPECT_TRUE(processingThread.pushTask(
+            [&awaiter]
+            {
+                awaiter.arrive();
+            }
+        ));
+        EXPECT_TRUE(awaiter.waitFor());
+    }
+
+    TEST_F(ProcessingThreadTest, CannotPushWhileStopWaitsForTheLastTask)
+    {
+        Awaiter running{};
+        std::promise<void> release{};
+        const auto released = release.get_future().share();
+        std::atomic_int lateRan = 0;
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        processingThread.pushTask(
+            [&running, released]
+            {
+                running.arrive();
+                released.wait_for(5s);
+            }
+        );
+        ASSERT_TRUE(running.waitFor());
+        std::thread stopper{[&processingThread]
+            {
+                processingThread.stop();
+            }};
+        int accepted = 0;
+        const auto refused = waitUntil(
+            [&processingThread, &lateRan, &accepted]
+            {
+                const auto pushed = processingThread.pushTask(
+                    [&lateRan]
+                    {
+                        ++lateRan;
+                    }
+                );
+                if (pushed)
+                    ++accepted;
+                return !pushed;
+            }
+        );
+        const auto permanentPushed = processingThread
+                                         .pushPermanentTask(
+                                             [](auto)
+                                             {
+                                                 return true;
+                                             }
+                                         )
+                                         .first;
+        release.set_value();
+        stopper.join();
+
+        EXPECT_TRUE(refused);
+        EXPECT_FALSE(permanentPushed);
+        EXPECT_FALSE(processingThread.pushTask([] {}));
+        EXPECT_EQ(lateRan.load(), accepted);
+    }
+
+    TEST_F(ProcessingThreadTest, RefusedPromiseTaskCompletesWithTheUnfinishedResult)
+    {
+        std::atomic_int ran = 0;
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        processingThread.stop();
+
+        auto refusedBool = processingThread.pushPromiseTask(
+            [&ran]
+            {
+                ++ran;
+                return true;
+            }
+        );
+        auto refusedVoid = processingThread.pushPromiseTask(
+            [&ran]
+            {
+                ++ran;
+            }
+        );
+        auto refusedExpected = processingThread.pushPromiseTask(
+            [&ran]() -> std::expected<int, std::string>
+            {
+                ++ran;
+                return 1;
+            },
+            []() -> std::expected<int, std::string>
+            {
+                return std::unexpected("refused"s);
+            }
+        );
+
+        ASSERT_EQ(refusedBool.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(refusedBool.get());
+        ASSERT_EQ(refusedVoid.wait_for(0s), std::future_status::ready);
+        EXPECT_NO_THROW(refusedVoid.get());
+        ASSERT_EQ(refusedExpected.wait_for(0s), std::future_status::ready);
+        EXPECT_EQ(refusedExpected.get(), std::unexpected("refused"s));
+        EXPECT_FALSE(processingThread.awaitCycle(0ms));
+        EXPECT_EQ(ran.load(), 0);
+    }
+
+    TEST_F(ProcessingThreadTest, PromiseTaskOfAFinalizedStrandCompletesWithTheUnfinishedResult)
+    {
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        auto strand = processingThread.createStrand();
+        auto finalTask = strand->pushFinalPromiseTask(
+            []
+            {
+                return true;
+            }
+        );
+        ASSERT_EQ(finalTask.wait_for(5s), std::future_status::ready);
+        EXPECT_TRUE(finalTask.get());
+
+        auto afterFinal = strand->pushPromiseTask(
+            []
+            {
+                return true;
+            }
+        );
+        auto secondFinal = strand->pushFinalPromiseTask(
+            []() -> std::expected<int, std::string>
+            {
+                return 1;
+            },
+            []() -> std::expected<int, std::string>
+            {
+                return std::unexpected("finalized"s);
+            }
+        );
+        ASSERT_EQ(afterFinal.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(afterFinal.get());
+        ASSERT_EQ(secondFinal.wait_for(0s), std::future_status::ready);
+        EXPECT_EQ(secondFinal.get(), std::unexpected("finalized"s));
+    }
+
+    TEST_F(ProcessingThreadTest, PromiseTaskOfAStrandOnAStoppedThreadCompletesWithTheUnfinishedResult)
     {
         ProcessingThread processingThread;
         processingThread.start(std::chrono::milliseconds{1});
+        auto strand = processingThread.createStrand();
         processingThread.stop();
-        EXPECT_TRUE(processingThread.pushTask([] {}));
+
+        auto refused = strand->pushPromiseTask(
+            []
+            {
+                return true;
+            }
+        );
+        ASSERT_EQ(refused.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(refused.get());
+    }
+
+    TEST_F(ProcessingThreadTest, ThrowingPromiseTaskCompletesWithTheUnfinishedResult)
+    {
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        auto threw = processingThread.pushPromiseTask(
+            []() -> bool
+            {
+                throw std::runtime_error("boom");
+            }
+        );
+        ASSERT_EQ(threw.wait_for(5s), std::future_status::ready);
+        EXPECT_FALSE(threw.get());
     }
 
     TEST_F(ProcessingThreadTest, CannotPushTaskWhileStopping)
@@ -1026,6 +1215,10 @@ namespace SecureShell::Test
                                           [&slowTurns, &fastTurns]
                                           {
                                               return std::pair{slowTurns.load(), fastTurns.load()};
+                                          },
+                                          []
+                                          {
+                                              return std::pair{-1, -2};
                                           }
                                       )
                                       .get();
@@ -1068,6 +1261,10 @@ namespace SecureShell::Test
             []
             {
                 return 42;
+            },
+            []
+            {
+                return 0;
             }
         );
         ASSERT_EQ(done.wait_for(5s), std::future_status::ready);
@@ -1117,13 +1314,15 @@ namespace SecureShell::Test
         EXPECT_EQ(ran.load(), 1);
     }
 
-    TEST_F(ProcessingThreadTest, EveryAcceptedTaskRunsWhenStopRacesWithPushes)
+    TEST_F(ProcessingThreadTest, EveryAcceptedTaskRunsBeforeStopReturnsWhenPushesRace)
     {
         constexpr int pushers = 4;
         std::atomic_int accepted = 0;
         std::atomic_int ran = 0;
         std::atomic_bool keepPushing = true;
         bool pushing = false;
+        int ranWhenStopped = 0;
+        bool refusedAfterStop = false;
         {
             ProcessingThread processingThread;
             processingThread.start(1ms);
@@ -1156,12 +1355,16 @@ namespace SecureShell::Test
                 }
             );
             processingThread.stop();
+            ranWhenStopped = ran.load();
+            refusedAfterStop = !processingThread.pushTask([] {});
             keepPushing = false;
             for (auto& thread : threads)
                 thread.join();
-            // Tasks accepted after stop wait for the next start or for the destructor.
         }
         EXPECT_TRUE(pushing);
+        EXPECT_TRUE(refusedAfterStop);
+        // A push that passed the check before stop() but queued after its drain would run only in the destructor.
+        EXPECT_EQ(ranWhenStopped, accepted.load());
         EXPECT_EQ(ran.load(), accepted.load());
     }
 
@@ -1191,6 +1394,354 @@ namespace SecureShell::Test
         ));
     }
 
+    TEST_F(ProcessingThreadTest, PushesAfterATaskThrewAreRefusedAndQueuedTasksFail)
+    {
+        std::atomic_int ran = 0;
+        std::future<bool> queuedBehind{};
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        // Queued from within, the task waits for the next cycle the loop does not reach.
+        processingThread.pushTask(
+            [&processingThread, &queuedBehind, &ran]
+            {
+                queuedBehind = processingThread.pushPromiseTask(
+                    [&ran]
+                    {
+                        ++ran;
+                        return true;
+                    }
+                );
+                throw std::runtime_error("boom");
+            }
+        );
+        ASSERT_TRUE(waitUntil(
+            [&processingThread]
+            {
+                return !processingThread.isRunning();
+            }
+        ));
+        ASSERT_EQ(queuedBehind.wait_for(5s), std::future_status::ready);
+        EXPECT_FALSE(queuedBehind.get());
+
+        EXPECT_FALSE(processingThread.pushTask(
+            [&ran]
+            {
+                ++ran;
+            }
+        ));
+        EXPECT_FALSE(processingThread
+                .pushPermanentTask(
+                    [](auto)
+                    {
+                        return true;
+                    }
+                )
+                .first);
+        auto refused = processingThread.pushPromiseTask(
+            [&ran]
+            {
+                ++ran;
+                return true;
+            }
+        );
+        ASSERT_EQ(refused.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(refused.get());
+        processingThread.stop();
+        EXPECT_EQ(ran.load(), 0);
+    }
+
+    TEST_F(ProcessingThreadTest, CleanupTasksStillRunAfterATaskThrew)
+    {
+        std::atomic_bool queuedRanOnTheThread = false;
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        processingThread.pushTask(
+            [&processingThread, &queuedRanOnTheThread]
+            {
+                processingThread.pushCleanupTask(
+                    [&processingThread, &queuedRanOnTheThread]
+                    {
+                        queuedRanOnTheThread = processingThread.withinProcessingThread();
+                    }
+                );
+                throw std::runtime_error("boom");
+            }
+        );
+        ASSERT_TRUE(waitUntil(
+            [&processingThread]
+            {
+                return !processingThread.isRunning();
+            }
+        ));
+
+        // Nothing else runs it now, so it runs here before the push returns, acting as the processing thread.
+        auto later = processingThread.pushCleanupPromiseTask(
+            [&processingThread]
+            {
+                return processingThread.withinProcessingThread();
+            }
+        );
+        ASSERT_EQ(later.wait_for(0s), std::future_status::ready);
+        EXPECT_TRUE(later.get());
+        EXPECT_FALSE(processingThread.withinProcessingThread());
+        EXPECT_TRUE(queuedRanOnTheThread.load());
+    }
+
+    TEST_F(ProcessingThreadTest, CleanupTaskPushedWhileStopDrainsRunsInTheDrain)
+    {
+        Awaiter lastBatchRunning{};
+        std::promise<void> releaseLastBatch{};
+        const auto lastBatchReleased = releaseLastBatch.get_future().share();
+        Awaiter firstCleanupRunning{};
+        std::promise<void> releaseFirstCleanup{};
+        const auto firstCleanupReleased = releaseFirstCleanup.get_future().share();
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        // The last batch throws after stop() began, the cleanup task it queued holds up the rest of stop().
+        processingThread.pushTask(
+            [&processingThread, &lastBatchRunning, &firstCleanupRunning, lastBatchReleased, firstCleanupReleased]
+            {
+                lastBatchRunning.arrive();
+                lastBatchReleased.wait_for(5s);
+                processingThread.pushCleanupTask(
+                    [&firstCleanupRunning, firstCleanupReleased]
+                    {
+                        firstCleanupRunning.arrive();
+                        firstCleanupReleased.wait_for(5s);
+                    }
+                );
+                throw std::runtime_error("boom");
+            }
+        );
+        ASSERT_TRUE(lastBatchRunning.waitFor());
+        std::thread stopper{[&processingThread]
+            {
+                processingThread.stop();
+            }};
+        const auto stopperId = stopper.get_id();
+        ASSERT_TRUE(waitUntil(
+            [&processingThread]
+            {
+                return !processingThread.pushTask([] {});
+            }
+        ));
+        releaseLastBatch.set_value();
+        ASSERT_TRUE(firstCleanupRunning.waitFor());
+
+        // Pushed while stop() is still running, it must not run on its caller beside the rest of stop().
+        std::thread::id ranOn{};
+        bool within = false;
+        std::atomic_bool pushed = false;
+        std::thread pusher{[&]
+            {
+                processingThread.pushCleanupTask(
+                    [&]
+                    {
+                        ranOn = std::this_thread::get_id();
+                        within = processingThread.withinProcessingThread();
+                    }
+                );
+                pushed = true;
+            }};
+        const auto pushedAtOnce = waitUntil(
+            [&pushed]
+            {
+                return pushed.load();
+            },
+            1s
+        );
+        releaseFirstCleanup.set_value();
+        pusher.join();
+        stopper.join();
+
+        EXPECT_TRUE(pushedAtOnce);
+        EXPECT_EQ(ranOn, stopperId);
+        EXPECT_TRUE(within);
+        EXPECT_FALSE(processingThread.withinProcessingThread());
+    }
+
+    TEST_F(ProcessingThreadTest, ThrowingTaskInTheDrainOfStopKeepsTheOthersRunning)
+    {
+        std::atomic_int ran = 0;
+        // Never started, so stop() drains what was pushed.
+        ProcessingThread processingThread;
+        auto threw = processingThread.pushPromiseTask(
+            []() -> bool
+            {
+                throw std::runtime_error("boom");
+            }
+        );
+        processingThread.pushTask(
+            [&ran]
+            {
+                ++ran;
+            }
+        );
+
+        EXPECT_NO_THROW(processingThread.stop());
+        EXPECT_EQ(ran.load(), 1);
+        ASSERT_EQ(threw.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(threw.get());
+        // The drain finished, so a final task runs here at once.
+        auto strand = processingThread.createStrand();
+        auto finalTask = strand->pushFinalPromiseTask(
+            []
+            {
+                return true;
+            }
+        );
+        ASSERT_EQ(finalTask.wait_for(0s), std::future_status::ready);
+        EXPECT_TRUE(finalTask.get());
+    }
+
+    TEST_F(ProcessingThreadTest, ThrowingCleanupTaskAfterStopDoesNotThrowAtItsCaller)
+    {
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        processingThread.stop();
+
+        std::future<bool> threw{};
+        EXPECT_NO_THROW(
+            threw = processingThread.pushCleanupPromiseTask(
+                []() -> bool
+                {
+                    throw std::runtime_error("boom");
+                }
+            )
+        );
+        ASSERT_TRUE(threw.valid());
+        ASSERT_EQ(threw.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(threw.get());
+        EXPECT_FALSE(processingThread.withinProcessingThread());
+    }
+
+    TEST_F(ProcessingThreadTest, PromiseTaskPushedWhileStopDrainsCompletesAfterTheTasksBeforeIt)
+    {
+        Awaiter drainRunning{};
+        std::promise<void> release{};
+        const auto released = release.get_future().share();
+        std::atomic_bool laterRan = false;
+        // Never started, so stop() drains what was pushed.
+        ProcessingThread processingThread;
+        auto strand = processingThread.createStrand();
+        strand->pushTask(
+            [&drainRunning, released]
+            {
+                drainRunning.arrive();
+                released.wait_for(5s);
+            }
+        );
+        strand->pushTask(
+            [&laterRan]
+            {
+                laterRan = true;
+            }
+        );
+        std::thread stopper{[&processingThread]
+            {
+                processingThread.stop();
+            }};
+        ASSERT_TRUE(drainRunning.waitFor());
+
+        // A barrier like Operation::waitForStrandTasks: refused, but complete only once the drain got to it.
+        auto barrier = strand->pushPromiseTask([] {});
+        const auto readyEarly = barrier.wait_for(0s) == std::future_status::ready;
+        release.set_value();
+        const auto readyLater = barrier.wait_for(5s) == std::future_status::ready;
+        const auto laterRanFirst = laterRan.load();
+        stopper.join();
+
+        EXPECT_FALSE(readyEarly);
+        EXPECT_TRUE(readyLater);
+        EXPECT_TRUE(laterRanFirst);
+    }
+
+    TEST_F(ProcessingThreadTest, FinalTaskAfterStopRunsOnTheCallerAndMayDestroyItsStrand)
+    {
+        ProcessingThread processingThread;
+        processingThread.start(std::chrono::milliseconds{1});
+        auto strand = processingThread.createStrand();
+        auto* const finalizing = strand.get();
+        processingThread.stop();
+
+        auto closed = finalizing->pushFinalPromiseTask(
+            [&strand, &processingThread]
+            {
+                const auto within = processingThread.withinProcessingThread();
+                strand.reset();
+                return within;
+            }
+        );
+        ASSERT_EQ(closed.wait_for(0s), std::future_status::ready);
+        EXPECT_TRUE(closed.get());
+        EXPECT_EQ(strand, nullptr);
+        EXPECT_FALSE(processingThread.withinProcessingThread());
+    }
+
+    TEST_F(ProcessingThreadTest, NoTaskRunsWhileItsOwnerIsDestroyed)
+    {
+        // Members after the thread are destroyed before it, like the libssh session and channels of a Session.
+        struct Owner
+        {
+            ProcessingThread processingThread{};
+            std::vector<int> touched{};
+            std::function<void()> afterStop{};
+
+            ~Owner()
+            {
+                processingThread.stop();
+                afterStop();
+            }
+        };
+
+        std::atomic_bool stopped = false;
+        std::atomic_bool keepPushing = true;
+        std::atomic_int pushedAfterStop = 0;
+        std::atomic_int acceptedAfterStop = 0;
+        std::atomic_int ranAfterStop = 0;
+        auto owner = std::make_unique<Owner>();
+        auto* const raw = owner.get();
+        raw->processingThread.start(std::chrono::milliseconds{1});
+        std::thread pusher{[&]
+            {
+                while (keepPushing)
+                {
+                    const auto afterStop = stopped.load();
+                    const auto accepted = raw->processingThread.pushTask(
+                        [raw, &stopped, &ranAfterStop]
+                        {
+                            if (stopped)
+                                ++ranAfterStop;
+                            raw->touched.push_back(1);
+                        }
+                    );
+                    if (afterStop)
+                    {
+                        ++pushedAfterStop;
+                        if (accepted)
+                            ++acceptedAfterStop;
+                    }
+                }
+            }};
+        raw->afterStop = [&]
+        {
+            stopped = true;
+            waitUntil(
+                [&pushedAfterStop]
+                {
+                    return pushedAfterStop.load() >= 100;
+                }
+            );
+            keepPushing = false;
+            pusher.join();
+        };
+
+        owner.reset();
+        EXPECT_GE(pushedAfterStop.load(), 100);
+        EXPECT_EQ(acceptedAfterStop.load(), 0);
+        EXPECT_EQ(ranAfterStop.load(), 0);
+    }
+
     TEST_F(ProcessingThreadTest, StrandWakeRunsAnIdleTaskAgain)
     {
         std::atomic_int turns = 0;
@@ -1360,6 +1911,10 @@ namespace SecureShell::Test
             []
             {
                 return 42;
+            },
+            []
+            {
+                return 0;
             }
         );
         ASSERT_EQ(done.wait_for(5s), std::future_status::ready);
diff --git a/ssh/test/ssh/test_sftp.hpp b/ssh/test/ssh/test_sftp.hpp
index 7db830ec..df39abb6 100644
--- a/ssh/test/ssh/test_sftp.hpp
+++ b/ssh/test/ssh/test_sftp.hpp
@@ -12,7 +12,11 @@
 #include <algorithm>
 #include <array>
 #include <atomic>
+#include <expected>
+#include <future>
+#include <limits>
 #include <optional>
+#include <stdexcept>
 #include <string>
 
 using namespace std::chrono_literals;
@@ -55,6 +59,71 @@ namespace SecureShell::Test
             return std::make_pair(std::move(session).value(), sftpFuture.get().value().lock());
         }
 
+        /**
+         * @brief Ends the processing loop of @p sftp with a throwing task, then waits until promise tasks come back
+         *        with an error at once.
+         */
+        ::testing::AssertionResult endProcessingLoop(SftpSession& sftp)
+        {
+            sftp.perform(
+                []
+                {
+                    throw std::runtime_error("boom");
+                }
+            );
+            const auto deadline = std::chrono::steady_clock::now() + replyTimeout;
+            while (std::chrono::steady_clock::now() < deadline)
+            {
+                auto probe = sftp.performPromise(
+                    []() -> std::expected<void, SftpError>
+                    {
+                        return {};
+                    }
+                );
+                if (probe.wait_for(replyTimeout) == std::future_status::ready && !probe.get().has_value())
+                    return ::testing::AssertionSuccess();
+            }
+            return ::testing::AssertionFailure() << "the processing loop kept running tasks";
+        }
+
+        /**
+         * @brief Opens a remote file, null when that failed.
+         */
+        std::shared_ptr<FileStream> openStream(
+            SftpSession& sftp,
+            std::string_view path,
+            SftpSession::OpenType openType,
+            std::filesystem::perms permissions
+        )
+        {
+            auto opened = sftp.openFile(remote(path), openType, permissions);
+            if (opened.wait_for(replyTimeout) != std::future_status::ready)
+                return nullptr;
+            auto result = opened.get();
+            return result.has_value() ? result.value().lock() : nullptr;
+        }
+
+        /**
+         * @brief Opens a new remote file for writing.
+         */
+        std::shared_ptr<FileStream> openForWriting(SftpSession& sftp, std::string_view path)
+        {
+            return openStream(
+                sftp,
+                path,
+                SftpSession::OpenType::Write | SftpSession::OpenType::Create | SftpSession::OpenType::Truncate,
+                std::filesystem::perms{0644u}
+            );
+        }
+
+        /**
+         * @brief Opens a remote file for reading.
+         */
+        std::shared_ptr<FileStream> openForReading(SftpSession& sftp, std::string_view path)
+        {
+            return openStream(sftp, path, SftpSession::OpenType::Read, std::filesystem::perms::owner_read);
+        }
+
         std::string createAlphabetString(std::size_t length)
         {
             std::string result;
@@ -82,6 +151,135 @@ namespace SecureShell::Test
         EXPECT_TRUE(sftp.has_value());
     }
 
+    TEST_F(SftpTests, SftpSessionAndSessionCloseAfterATaskOnTheirThreadThrew)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [session, sftp] = createSftpSession(serverStartResult->port);
+        ASSERT_TRUE(openForReading(*sftp, "/home/test/large.txt"));
+
+        ASSERT_TRUE(endProcessingLoop(*sftp));
+        auto another = session->createSftpSession();
+        ASSERT_EQ(another.wait_for(0s), std::future_status::ready);
+        EXPECT_FALSE(another.get().has_value());
+
+        auto closing = std::async(
+            std::launch::async,
+            [&session, &sftp]
+            {
+                const auto closed = sftp->close();
+                session.reset();
+                return closed;
+            }
+        );
+        ASSERT_EQ(closing.wait_for(10s), std::future_status::ready);
+        EXPECT_TRUE(closing.get());
+    }
+
+    TEST_F(SftpTests, ReadAllAfterTheSftpSessionClosedFailsAtOnce)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [_, sftp] = createSftpSession(serverStartResult->port);
+        auto file = openForReading(*sftp, "/home/test/large.txt");
+        ASSERT_TRUE(file);
+        ASSERT_TRUE(sftp->close());
+
+        auto read = file->readAll(
+            [](std::string_view)
+            {
+                return true;
+            }
+        );
+        ASSERT_EQ(read.wait_for(0s), std::future_status::ready);
+        const auto result = read.get();
+        ASSERT_FALSE(result.has_value());
+        EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed);
+    }
+
+    TEST_F(SftpTests, ReadAllAfterATaskOnTheThreadThrewFailsAtOnce)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [_, sftp] = createSftpSession(serverStartResult->port);
+        auto file = openForReading(*sftp, "/home/test/large.txt");
+        ASSERT_TRUE(file);
+        ASSERT_TRUE(endProcessingLoop(*sftp));
+
+        auto read = file->readAll(
+            [](std::string_view)
+            {
+                return true;
+            }
+        );
+        ASSERT_EQ(read.wait_for(0s), std::future_status::ready);
+        const auto result = read.get();
+        ASSERT_FALSE(result.has_value());
+        EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed);
+    }
+
+    TEST_F(SftpTests, LargeWriteIsWrittenInChunks)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [_, sftp] = createSftpSession(serverStartResult->port);
+        auto file = openForWriting(*sftp, "/home/test/uploaded.bin");
+        ASSERT_TRUE(file);
+        const std::string content(file->writeLengthLimit() * 3u + 17u, 'w');
+
+        auto written = file->write(content);
+        ASSERT_EQ(written.wait_for(replyTimeout), std::future_status::ready);
+        EXPECT_TRUE(written.get().has_value());
+        file->close();
+
+        auto information = sftp->stat(remote("/home/test/uploaded.bin"));
+        ASSERT_EQ(information.wait_for(replyTimeout), std::future_status::ready);
+        auto stated = information.get();
+        ASSERT_TRUE(stated.has_value());
+        EXPECT_EQ(stated->size, content.size());
+    }
+
+    TEST_F(SftpTests, LargeWriteToAClosedFileFailsWithFileNull)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [_, sftp] = createSftpSession(serverStartResult->port);
+        auto file = openForWriting(*sftp, "/home/test/uploaded.bin");
+        ASSERT_TRUE(file);
+        file->close();
+
+        auto written = file->write(std::string(file->writeLengthLimit() * 2u, 'w'));
+        ASSERT_EQ(written.wait_for(replyTimeout), std::future_status::ready);
+        const auto result = written.get();
+        ASSERT_FALSE(result.has_value());
+        EXPECT_EQ(result.error().wrapperError, WrapperErrors::FileNull);
+    }
+
+    TEST_F(SftpTests, LargeWriteAfterTheSftpSessionClosedFailsAtOnce)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [_, sftp] = createSftpSession(serverStartResult->port);
+        auto file = openForWriting(*sftp, "/home/test/uploaded.bin");
+        ASSERT_TRUE(file);
+        ASSERT_TRUE(sftp->close());
+
+        auto written = file->write(std::string(file->writeLengthLimit() * 2u, 'w'));
+        ASSERT_EQ(written.wait_for(0s), std::future_status::ready);
+        const auto result = written.get();
+        ASSERT_FALSE(result.has_value());
+        EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed);
+    }
+
+    TEST_F(SftpTests, LargeWriteAfterATaskOnTheThreadThrewFailsAtOnce)
+    {
+        CREATE_SERVER_AND_JOINER(Sftp);
+        auto [_, sftp] = createSftpSession(serverStartResult->port);
+        auto file = openForWriting(*sftp, "/home/test/uploaded.bin");
+        ASSERT_TRUE(file);
+        ASSERT_TRUE(endProcessingLoop(*sftp));
+
+        auto written = file->write(std::string(file->writeLengthLimit() * 2u, 'w'));
+        ASSERT_EQ(written.wait_for(0s), std::future_status::ready);
+        const auto result = written.get();
+        ASSERT_FALSE(result.has_value());
+        EXPECT_EQ(result.error().wrapperError, WrapperErrors::TaskPushFailed);
+    }
+
     TEST_F(SftpTests, CanListDirectory)
     {
         CREATE_SERVER_AND_JOINER(Sftp);
@@ -1139,6 +1337,10 @@ namespace SecureShell::Test
                         for (auto* entry = static_cast<sftp_session>(sftp)->queue; entry != nullptr; entry = entry->next)
                             ++count;
                         return count;
+                    },
+                    []
+                    {
+                        return std::numeric_limits<std::size_t>::max();
                     }
                 )
                 .get();