diff --git a/src/ModularPipelines/Context/Downloader.cs b/src/ModularPipelines/Context/Downloader.cs
index 332d0209335..8195178bcbf 100644
--- a/src/ModularPipelines/Context/Downloader.cs
+++ b/src/ModularPipelines/Context/Downloader.cs
@@ -25,14 +25,14 @@ public Downloader(IModuleLoggerProvider moduleLoggerProvider, IHttpContext http,
     public async Task<string?> DownloadStringAsync(DownloadOptions options,
         CancellationToken cancellationToken = default)
     {
-        var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false);
+        using var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false);
 
         return await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
     }
 
     public async Task<File> DownloadFileAsync(DownloadFileOptions options, CancellationToken cancellationToken = default)
     {
-        var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false);
+        using var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false);
 
         var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
         await using (stream.ConfigureAwait(false))
@@ -44,10 +44,26 @@ public async Task<File> DownloadFileAsync(DownloadFileOptions options, Cancellat
                 throw new IOException($"{filePathToSave} already exists and overwrite is false");
             }
 
-            var newFile = _fileSystemProvider.Create(filePathToSave);
-            await using (newFile.ConfigureAwait(false))
+            var destinationDirectory = Path.GetDirectoryName(filePathToSave);
+            var temporaryPath = Path.Combine(
+                destinationDirectory ?? string.Empty,
+                _fileSystemProvider.GetRandomFileName());
+            try
             {
-                await stream.CopyToAsync(newFile, cancellationToken).ConfigureAwait(false);
+                var newFile = _fileSystemProvider.Create(temporaryPath);
+                await using (newFile.ConfigureAwait(false))
+                {
+                    await stream.CopyToAsync(newFile, cancellationToken).ConfigureAwait(false);
+                }
+
+                _fileSystemProvider.MoveFile(temporaryPath, filePathToSave, options.Overwrite);
+            }
+            finally
+            {
+                if (_fileSystemProvider.FileExists(temporaryPath))
+                {
+                    _fileSystemProvider.DeleteFile(temporaryPath);
+                }
             }
 
             _moduleLoggerProvider.GetLogger().LogInformation("File {Uri} downloaded to {SaveLocation}", options.DownloadUri, filePathToSave);
@@ -68,7 +84,15 @@ public async Task<HttpResponseMessage> DownloadResponseAsync(DownloadOptions opt
             HttpClient = options.HttpClient,
         }, cancellationToken).ConfigureAwait(false);
 
-        return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch
+        {
+            response.Dispose();
+            throw;
+        }
     }
 
     /// <summary>
diff --git a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs
index 9c544b37df5..7313398da89 100644
--- a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs
+++ b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs
@@ -8,38 +8,68 @@ public interface IFileSystemProvider
 {
     // File read operations
     Task<string> ReadAllTextAsync(string path, CancellationToken cancellationToken = default);
+
     IAsyncEnumerable<string> ReadLinesAsync(string path, CancellationToken cancellationToken = default);
+
     Task<byte[]> ReadAllBytesAsync(string path, CancellationToken cancellationToken = default);
 
     // File write operations
     Task WriteAllTextAsync(string path, string contents, CancellationToken cancellationToken = default);
+
     Task WriteAllBytesAsync(string path, byte[] contents, CancellationToken cancellationToken = default);
+
     Task WriteAllLinesAsync(string path, IEnumerable<string> contents, CancellationToken cancellationToken = default);
+
     Task AppendAllTextAsync(string path, string contents, CancellationToken cancellationToken = default);
+
     Task AppendAllLinesAsync(string path, IEnumerable<string> contents, CancellationToken cancellationToken = default);
 
     // File stream operations
     Stream OpenRead(string path);
+
     Stream Create(string path);
+
     Stream Open(string path, FileMode mode, FileAccess access);
 
     // File management operations
     void DeleteFile(string path);
+
     void CopyFile(string sourcePath, string destinationPath, bool overwrite);
+
     void MoveFile(string sourcePath, string destinationPath);
+
+    void MoveFile(string sourcePath, string destinationPath, bool overwrite)
+    {
+        if (overwrite && FileExists(destinationPath))
+        {
+            throw new NotSupportedException(
+                "This file system provider does not support atomic overwrite moves.");
+        }
+
+        MoveFile(sourcePath, destinationPath);
+    }
+
     bool FileExists(string path);
 
     // Directory operations
     void CreateDirectory(string path);
+
     void DeleteDirectory(string path, bool recursive);
+
     void MoveDirectory(string sourcePath, string destinationPath);
+
     bool DirectoryExists(string path);
+
     IEnumerable<string> EnumerateFiles(string path, string searchPattern, SearchOption searchOption);
+
     IEnumerable<string> EnumerateDirectories(string path, string searchPattern, SearchOption searchOption);
 
     // Path operations
     string GetTempPath();
+
     string GetRandomFileName();
+
     string Combine(params string[] paths);
+
     string GetRelativePath(string relativeTo, string path);
 }
diff --git a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs
index a245065efef..c7bd3a9ec75 100644
--- a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs
+++ b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs
@@ -9,11 +9,13 @@ namespace ModularPipelines.FileSystem;
 public sealed class SystemFileSystemProvider : IFileSystemProvider
 {
     /// <summary>
-    /// Singleton instance for use when no DI is available.
+    /// Gets singleton instance for use when no DI is available.
     /// </summary>
     public static SystemFileSystemProvider Instance { get; } = new();
 
-    private SystemFileSystemProvider() { }
+    private SystemFileSystemProvider()
+    {
+    }
 
     // File read operations
     public Task<string> ReadAllTextAsync(string path, CancellationToken cancellationToken = default)
@@ -61,6 +63,9 @@ public void CopyFile(string sourcePath, string destinationPath, bool overwrite)
     public void MoveFile(string sourcePath, string destinationPath)
         => System.IO.File.Move(sourcePath, destinationPath);
 
+    public void MoveFile(string sourcePath, string destinationPath, bool overwrite)
+        => System.IO.File.Move(sourcePath, destinationPath, overwrite);
+
     public bool FileExists(string path)
         => System.IO.File.Exists(path);
 
diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs
index 170d14a90de..7517ce645bc 100644
--- a/src/ModularPipelines/Http/Http.cs
+++ b/src/ModularPipelines/Http/Http.cs
@@ -28,36 +28,82 @@ public Http(
 
     public async Task<HttpResponseMessage> SendAsync(HttpOptions httpOptions, CancellationToken cancellationToken = default)
     {
-        // Priority: options property > pipeline default > no timeout
-        var effectiveTimeout = httpOptions.Timeout ?? _pipelineOptions.Value.DefaultHttpTimeout;
+        var httpClient = httpOptions.HttpClient ?? GetHttpClient(httpOptions.LoggingType);
+
+        // Priority: options property > pipeline default > HttpClient timeout
+        var effectiveTimeout = httpOptions.Timeout
+                               ?? _pipelineOptions.Value.DefaultHttpTimeout
+                               ?? GetFiniteTimeout(httpClient);
 
         // Create timeout token if timeout is specified
-        using var timeoutCts = effectiveTimeout.HasValue
+        var timeoutCts = effectiveTimeout.HasValue
             ? new CancellationTokenSource(effectiveTimeout.Value)
             : null;
 
-        // Link the timeout token with the passed cancellation token, or just use the original if no timeout
-        using var linkedCts = timeoutCts != null
+        // Keep caller cancellation alive through streamed body consumption even when no
+        // finite timeout applies.
+        var linkedCts = timeoutCts != null
             ? CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken)
-            : null;
+            : cancellationToken.CanBeCanceled
+                ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken)
+                : null;
 
         var effectiveCancellationToken = linkedCts?.Token ?? cancellationToken;
 
-        if (httpOptions.HttpClient != null)
+        try
         {
-            return await SendAndWrapLogging(httpOptions, effectiveCancellationToken).ConfigureAwait(false);
+            HttpResponseMessage WrapResponse(HttpResponseMessage response)
+            {
+                if (linkedCts is null)
+                {
+                    return response;
+                }
+
+                response.Content = new TimeoutHttpContent(response.Content, timeoutCts, linkedCts);
+                timeoutCts = null;
+                linkedCts = null;
+                return response;
+            }
+
+            if (httpOptions.HttpClient != null)
+            {
+                return await SendAndWrapLogging(
+                        httpClient,
+                        httpOptions,
+                        WrapResponse,
+                        effectiveCancellationToken)
+                    .ConfigureAwait(false);
+            }
+
+            var response = await httpClient.SendAsync(
+                    httpOptions.HttpRequestMessage,
+                    HttpCompletionOption.ResponseHeadersRead,
+                    effectiveCancellationToken)
+                .ConfigureAwait(false);
+
+            response = WrapResponse(response);
+            try
+            {
+                if (!httpOptions.ThrowOnNonSuccessStatusCode)
+                {
+                    return response;
+                }
+
+                return await response
+                    .EnsureSuccessStatusCodeWithContentAsync(effectiveCancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch
+            {
+                response.Dispose();
+                throw;
+            }
         }
-
-        var httpClient = GetHttpClient(httpOptions.LoggingType);
-
-        var response = await httpClient.SendAsync(httpOptions.HttpRequestMessage, effectiveCancellationToken).ConfigureAwait(false);
-
-        if (!httpOptions.ThrowOnNonSuccessStatusCode)
+        finally
         {
-            return response;
+            linkedCts?.Dispose();
+            timeoutCts?.Dispose();
         }
-
-        return await response.EnsureSuccessStatusCodeWithContentAsync(effectiveCancellationToken).ConfigureAwait(false);
     }
 
     private HttpClient GetHttpClient(HttpLoggingType loggingType)
@@ -66,34 +112,60 @@ private HttpClient GetHttpClient(HttpLoggingType loggingType)
         return _httpClientFactory.CreateClient(clientName);
     }
 
-    private async Task<HttpResponseMessage> SendAndWrapLogging(HttpOptions httpOptions, CancellationToken cancellationToken)
+    private async Task<HttpResponseMessage> SendAndWrapLogging(
+        HttpClient httpClient,
+        HttpOptions httpOptions,
+        Func<HttpResponseMessage, HttpResponseMessage> wrapResponse,
+        CancellationToken cancellationToken)
     {
         var logger = _moduleLoggerProvider.GetLogger();
         var loggingOptions = GetEffectiveLoggingOptions(httpOptions);
 
         if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Request))
         {
-            await _httpLogger.PrintRequest(httpOptions.HttpRequestMessage, logger, loggingOptions).ConfigureAwait(false);
+            await _httpLogger
+                .PrintRequest(
+                    httpOptions.HttpRequestMessage,
+                    logger,
+                    loggingOptions,
+                    cancellationToken)
+                .ConfigureAwait(false);
         }
 
         var stopWatch = Stopwatch.StartNew();
 
-        var response = await httpOptions.HttpClient!.SendAsync(httpOptions.HttpRequestMessage, cancellationToken).ConfigureAwait(false);
+        var response = await httpClient.SendAsync(
+                httpOptions.HttpRequestMessage,
+                HttpCompletionOption.ResponseHeadersRead,
+                cancellationToken)
+            .ConfigureAwait(false);
 
-        LogStatusCode(response.StatusCode, httpOptions, loggingOptions);
-        LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions);
-
-        if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response))
+        try
         {
-            await _httpLogger.PrintResponse(response, logger, loggingOptions).ConfigureAwait(false);
-        }
+            LogStatusCode(response.StatusCode, httpOptions, loggingOptions);
+            LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions);
 
-        if (!httpOptions.ThrowOnNonSuccessStatusCode)
+            if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response))
+            {
+                await _httpLogger
+                    .PrintResponse(response, logger, loggingOptions, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+
+            response = wrapResponse(response);
+
+            if (!httpOptions.ThrowOnNonSuccessStatusCode)
+            {
+                return response;
+            }
+
+            return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch
         {
-            return response;
+            response.Dispose();
+            throw;
         }
-
-        return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false);
     }
 
     private HttpLoggingOptions GetEffectiveLoggingOptions(HttpOptions httpOptions)
@@ -112,6 +184,13 @@ private HttpLoggingOptions GetEffectiveLoggingOptions(HttpOptions httpOptions)
         return HttpLoggingOptions.Default;
     }
 
+    private static TimeSpan? GetFiniteTimeout(HttpClient httpClient)
+    {
+        return httpClient.Timeout == System.Threading.Timeout.InfiniteTimeSpan
+            ? null
+            : httpClient.Timeout;
+    }
+
     private void LogDuration(TimeSpan duration, HttpOptions httpOptions, HttpLoggingOptions loggingOptions)
     {
         if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Duration) && loggingOptions.LogDuration)
diff --git a/src/ModularPipelines/Http/HttpBodySecretRedactor.cs b/src/ModularPipelines/Http/HttpBodySecretRedactor.cs
new file mode 100644
index 00000000000..403bdc7776d
--- /dev/null
+++ b/src/ModularPipelines/Http/HttpBodySecretRedactor.cs
@@ -0,0 +1,62 @@
+using Microsoft.Extensions.Options;
+using ModularPipelines.Constants;
+using ModularPipelines.Engine;
+using ModularPipelines.Options;
+
+namespace ModularPipelines.Http;
+
+internal static class HttpBodySecretRedactor
+{
+    public static string Redact(
+        string body,
+        bool isTruncated,
+        ISecretObfuscator secretObfuscator,
+        ISecretProvider secretProvider,
+        IOptions<SecretMaskingOptions> maskingOptions)
+    {
+        var boundarySafeBody = isTruncated
+            ? RedactPartialSecretAtBoundary(body, secretProvider.GetSnapshot().Secrets, maskingOptions.Value)
+            : body;
+        return secretObfuscator.Obfuscate(boundarySafeBody, null);
+    }
+
+    private static string RedactPartialSecretAtBoundary(
+        string body,
+        IReadOnlyList<string> secrets,
+        SecretMaskingOptions options)
+    {
+        if (body.Length == 0)
+        {
+            return body;
+        }
+
+        var comparison = options.CaseInsensitive
+            ? StringComparison.OrdinalIgnoreCase
+            : StringComparison.Ordinal;
+        var longestPartialMatch = 0;
+
+        foreach (var secret in secrets)
+        {
+            var maximumLength = Math.Min(body.Length, secret.Length - 1);
+            for (var length = maximumLength; length > longestPartialMatch; length--)
+            {
+                if (body.AsSpan(body.Length - length, length)
+                    .Equals(secret.AsSpan(0, length), comparison))
+                {
+                    longestPartialMatch = length;
+                    break;
+                }
+            }
+        }
+
+        if (longestPartialMatch == 0)
+        {
+            return body;
+        }
+
+        var mask = string.IsNullOrWhiteSpace(options.MaskValue)
+            ? LoggingConstants.SecretMask
+            : options.MaskValue;
+        return string.Concat(body.AsSpan(0, body.Length - longestPartialMatch), mask);
+    }
+}
diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs
new file mode 100644
index 00000000000..2ab2ef7a1ca
--- /dev/null
+++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs
@@ -0,0 +1,293 @@
+using System.Buffers;
+using System.Text;
+
+namespace ModularPipelines.Http;
+
+/// <summary>
+/// Reads bounded text previews while preserving the complete content stream for its consumer.
+/// </summary>
+internal static class HttpContentPreviewReader
+{
+    private const int MaximumPreviewBufferSize = 81920;
+
+    /// <summary>
+    /// Reads at most one byte beyond the configured preview limit.
+    /// </summary>
+    /// <param name="content">The HTTP content to preview.</param>
+    /// <param name="maxBytes">The maximum number of bytes to include in the preview.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The preview, truncation state, replayable content, and declared total length.</returns>
+    public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)>
+        ReadAsync(
+            HttpContent content,
+            int maxBytes,
+            CancellationToken cancellationToken)
+    {
+        if (maxBytes <= 0 || maxBytes == int.MaxValue)
+        {
+            var totalLength = content.Headers.ContentLength;
+            var bytes = await ReadAllBytesAsync(content, cancellationToken).ConfigureAwait(false);
+            var unboundedReplayContent = new ByteArrayContent(bytes);
+            CopyHeaders(content, unboundedReplayContent);
+            content.Dispose();
+            return (DecodeCompletePrefix(GetEncoding(unboundedReplayContent), bytes, bytes.Length),
+                false,
+                unboundedReplayContent,
+                totalLength ?? bytes.LongLength);
+        }
+
+        var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
+        using var cancellationRegistration = cancellationToken.Register(
+            static state => ((Stream) state!).Dispose(),
+            stream);
+        var probeLength = maxBytes + 1;
+        var readBuffer = ArrayPool<byte>.Shared.Rent(
+            Math.Min(probeLength, MaximumPreviewBufferSize));
+        using var buffer = new MemoryStream(GetInitialCapacity(content, probeLength));
+
+        try
+        {
+            while (buffer.Length < probeLength)
+            {
+                var bytesRemaining = probeLength - (int) buffer.Length;
+                var read = await stream.ReadAsync(
+                        readBuffer.AsMemory(
+                            0,
+                            Math.Min(MaximumPreviewBufferSize, bytesRemaining)),
+                        cancellationToken)
+                    .ConfigureAwait(false);
+                cancellationToken.ThrowIfCancellationRequested();
+                if (read == 0)
+                {
+                    break;
+                }
+
+                buffer.Write(readBuffer, 0, read);
+            }
+        }
+        catch (Exception exception)
+            when (cancellationToken.IsCancellationRequested
+                  && exception is ObjectDisposedException or IOException)
+        {
+            throw new OperationCanceledException(
+                "The HTTP content preview read was cancelled.",
+                exception,
+                cancellationToken);
+        }
+        finally
+        {
+            ArrayPool<byte>.Shared.Return(readBuffer);
+        }
+
+        var bufferedBytes = buffer.ToArray();
+        var bytesRead = bufferedBytes.Length;
+        var isTruncated = bytesRead > maxBytes;
+        var previewLength = Math.Min(bytesRead, maxBytes);
+        var encoding = GetEncoding(content);
+        var preview = DecodeCompletePrefix(encoding, bufferedBytes, previewLength);
+        var replayContent = CreateReplayContent(content, stream, bufferedBytes);
+
+        return (preview, isTruncated, replayContent, content.Headers.ContentLength);
+    }
+
+    /// <summary>
+    /// Reads request content into replayable storage so redirects and authentication retries can resend it.
+    /// </summary>
+    /// <param name="content">The HTTP request content to preview.</param>
+    /// <param name="maxBytes">The maximum number of bytes to include in the preview.</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The preview, truncation state, replayable content, and total length.</returns>
+    /// <remarks>This method takes ownership of and disposes <paramref name="content"/>.</remarks>
+    public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)>
+        ReadReplayableAsync(
+            HttpContent content,
+            int maxBytes,
+            CancellationToken cancellationToken)
+    {
+        var totalLength = content.Headers.ContentLength;
+        var bytes = await ReadAllBytesAsync(content, cancellationToken).ConfigureAwait(false);
+        var previewLength = maxBytes <= 0 || maxBytes == int.MaxValue
+            ? bytes.Length
+            : Math.Min(bytes.Length, maxBytes);
+        var isTruncated = previewLength < bytes.Length;
+        var preview = DecodeCompletePrefix(GetEncoding(content), bytes, previewLength);
+        var replayContent = new ByteArrayContent(bytes);
+        CopyHeaders(content, replayContent);
+        content.Dispose();
+
+        return (
+            preview,
+            isTruncated,
+            replayContent,
+            totalLength ?? bytes.LongLength);
+    }
+
+    private static async Task<byte[]> ReadAllBytesAsync(
+        HttpContent content,
+        CancellationToken cancellationToken)
+    {
+        var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
+        using var cancellationRegistration = cancellationToken.Register(
+            static state => ((Stream) state!).Dispose(),
+            stream);
+        using var buffer = new MemoryStream();
+
+        try
+        {
+            await stream.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false);
+            cancellationToken.ThrowIfCancellationRequested();
+            return buffer.ToArray();
+        }
+        catch (Exception exception)
+            when (cancellationToken.IsCancellationRequested
+                  && exception is ObjectDisposedException or IOException)
+        {
+            throw new OperationCanceledException(
+                "The HTTP content read was cancelled.",
+                exception,
+                cancellationToken);
+        }
+    }
+
+    private static int GetInitialCapacity(HttpContent content, int probeLength)
+    {
+        var declaredLength = content.Headers.ContentLength;
+        return declaredLength is >= 0
+            ? (int) Math.Min(Math.Min(declaredLength.Value, probeLength), MaximumPreviewBufferSize)
+            : Math.Min(probeLength, MaximumPreviewBufferSize);
+    }
+
+    private static Encoding GetEncoding(HttpContent content)
+    {
+        var charset = content.Headers.ContentType?.CharSet?.Trim('"');
+        if (string.IsNullOrWhiteSpace(charset))
+        {
+            return Encoding.UTF8;
+        }
+
+        try
+        {
+            return Encoding.GetEncoding(charset);
+        }
+        catch (ArgumentException)
+        {
+            return Encoding.UTF8;
+        }
+    }
+
+    private static string DecodeCompletePrefix(Encoding encoding, byte[] buffer, int length)
+    {
+        var decoder = encoding.GetDecoder();
+        var characters = new char[encoding.GetMaxCharCount(length)];
+        decoder.Convert(
+            buffer.AsSpan(0, length),
+            characters,
+            flush: false,
+            out _,
+            out var charactersUsed,
+            out _);
+        return new string(characters, 0, charactersUsed);
+    }
+
+    private static StreamContent CreateReplayContent(
+        HttpContent originalContent,
+        Stream remainingStream,
+        byte[] prefix)
+    {
+        var replayContent = new StreamContent(
+            new PrefixReplayStream(prefix, remainingStream, originalContent));
+        CopyHeaders(originalContent, replayContent);
+
+        return replayContent;
+    }
+
+    private static void CopyHeaders(HttpContent source, HttpContent destination)
+    {
+        foreach (var header in source.Headers)
+        {
+            destination.Headers.TryAddWithoutValidation(header.Key, header.Value);
+        }
+    }
+
+    private sealed class PrefixReplayStream(
+        byte[] prefix,
+        Stream remainingStream,
+        HttpContent ownedContent) : Stream
+    {
+        private int _prefixPosition;
+
+        public override bool CanRead => true;
+
+        public override bool CanSeek => false;
+
+        public override bool CanWrite => false;
+
+        public override long Length => throw new NotSupportedException();
+
+        public override long Position
+        {
+            get => throw new NotSupportedException();
+            set => throw new NotSupportedException();
+        }
+
+        public override void Flush()
+        {
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            var prefixBytesRead = ReadPrefix(buffer.AsSpan(offset, count));
+            return prefixBytesRead > 0
+                ? prefixBytesRead
+                : remainingStream.Read(buffer, offset, count);
+        }
+
+        public override Task<int> ReadAsync(
+            byte[] buffer,
+            int offset,
+            int count,
+            CancellationToken cancellationToken)
+        {
+            return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
+        }
+
+        public override async ValueTask<int> ReadAsync(
+            Memory<byte> buffer,
+            CancellationToken cancellationToken = default)
+        {
+            var prefixBytesRead = ReadPrefix(buffer.Span);
+            return prefixBytesRead > 0
+                ? prefixBytesRead
+                : await remainingStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+        }
+
+        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+        public override void SetLength(long value) => throw new NotSupportedException();
+
+        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                ownedContent.Dispose();
+            }
+
+            base.Dispose(disposing);
+        }
+
+        private int ReadPrefix(Span<byte> destination)
+        {
+            var bytesToCopy = Math.Min(destination.Length, prefix.Length - _prefixPosition);
+            if (bytesToCopy <= 0)
+            {
+                return 0;
+            }
+
+            prefix.AsSpan(_prefixPosition, bytesToCopy).CopyTo(destination);
+            _prefixPosition += bytesToCopy;
+            return bytesToCopy;
+        }
+    }
+}
diff --git a/src/ModularPipelines/Http/HttpLogger.cs b/src/ModularPipelines/Http/HttpLogger.cs
index 4f08d1c9178..5ffe4d04674 100644
--- a/src/ModularPipelines/Http/HttpLogger.cs
+++ b/src/ModularPipelines/Http/HttpLogger.cs
@@ -28,7 +28,20 @@ public HttpLogger(IHttpRequestFormatter requestFormatter, IHttpResponseFormatter
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger)
     {
-        return PrintRequest(request, logger, HttpLoggingOptions.Default);
+        return PrintRequest(
+            request,
+            logger,
+            HttpLoggingOptions.Default,
+            CancellationToken.None);
+    }
+
+    /// <inheritdoc/>
+    public Task PrintRequest(
+        HttpRequestMessage request,
+        IModuleLogger logger,
+        CancellationToken cancellationToken)
+    {
+        return PrintRequest(request, logger, HttpLoggingOptions.Default, cancellationToken);
     }
 
     /// <summary>
@@ -38,14 +51,26 @@ public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger)
     /// <param name="logger">The current module logger.</param>
     /// <param name="options">Options controlling what parts of the request to log.</param>
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    public async Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options)
+    public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options)
+    {
+        return PrintRequest(request, logger, options, CancellationToken.None);
+    }
+
+    /// <inheritdoc/>
+    public async Task PrintRequest(
+        HttpRequestMessage request,
+        IModuleLogger logger,
+        HttpLoggingOptions options,
+        CancellationToken cancellationToken)
     {
         if (!options.LogRequest)
         {
             return;
         }
 
-        var formattedRequest = await _requestFormatter.FormatAsync(request, options).ConfigureAwait(false);
+        var formattedRequest = await _requestFormatter
+            .FormatAsync(request, options, cancellationToken)
+            .ConfigureAwait(false);
         logger.LogInformation("HTTP Request:\n{Request}", formattedRequest);
     }
 
@@ -57,7 +82,20 @@ public async Task PrintRequest(HttpRequestMessage request, IModuleLogger logger,
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger)
     {
-        return PrintResponse(response, logger, HttpLoggingOptions.Default);
+        return PrintResponse(
+            response,
+            logger,
+            HttpLoggingOptions.Default,
+            CancellationToken.None);
+    }
+
+    /// <inheritdoc/>
+    public Task PrintResponse(
+        HttpResponseMessage response,
+        IModuleLogger logger,
+        CancellationToken cancellationToken)
+    {
+        return PrintResponse(response, logger, HttpLoggingOptions.Default, cancellationToken);
     }
 
     /// <summary>
@@ -67,20 +105,32 @@ public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger)
     /// <param name="logger">The current module logger.</param>
     /// <param name="options">Options controlling what parts of the response to log.</param>
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
-    public async Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options)
+    public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options)
+    {
+        return PrintResponse(response, logger, options, CancellationToken.None);
+    }
+
+    /// <inheritdoc/>
+    public async Task PrintResponse(
+        HttpResponseMessage response,
+        IModuleLogger logger,
+        HttpLoggingOptions options,
+        CancellationToken cancellationToken)
     {
         if (!options.LogResponse)
         {
             return;
         }
 
-        var formattedResponse = await _responseFormatter.FormatAsync(response, options).ConfigureAwait(false);
+        var formattedResponse = await _responseFormatter
+            .FormatAsync(response, options, cancellationToken)
+            .ConfigureAwait(false);
         logger.LogInformation("HTTP Response:\n{Response}", formattedResponse);
     }
 
     public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger)
     {
-        var statusCode = httpStatusCode == null ? null as int? : (int)httpStatusCode;
+        var statusCode = httpStatusCode == null ? null as int? : (int) httpStatusCode;
         var icon = statusCode is >= 200 and < 300 ? "+" : "x";
 
         logger.LogInformation("{Icon} HTTP Status: {StatusCode} {HttpStatusCode}", icon, statusCode, httpStatusCode);
diff --git a/src/ModularPipelines/Http/HttpRequestFormatter.cs b/src/ModularPipelines/Http/HttpRequestFormatter.cs
index 6907bb2d97b..7512223da82 100644
--- a/src/ModularPipelines/Http/HttpRequestFormatter.cs
+++ b/src/ModularPipelines/Http/HttpRequestFormatter.cs
@@ -1,5 +1,6 @@
 using System.Net.Http.Headers;
 using System.Text;
+using Microsoft.Extensions.Options;
 using ModularPipelines.Constants;
 using ModularPipelines.Engine;
 using ModularPipelines.Options;
@@ -31,10 +32,17 @@ namespace ModularPipelines.Http;
 internal class HttpRequestFormatter : IHttpRequestFormatter
 {
     private readonly ISecretObfuscator _secretObfuscator;
+    private readonly ISecretProvider _secretProvider;
+    private readonly IOptions<SecretMaskingOptions> _secretMaskingOptions;
 
-    public HttpRequestFormatter(ISecretObfuscator secretObfuscator)
+    public HttpRequestFormatter(
+        ISecretObfuscator secretObfuscator,
+        ISecretProvider secretProvider,
+        IOptions<SecretMaskingOptions> secretMaskingOptions)
     {
         _secretObfuscator = secretObfuscator;
+        _secretProvider = secretProvider;
+        _secretMaskingOptions = secretMaskingOptions;
     }
 
     private static readonly HashSet<string> TextMediaTypes = new(StringComparer.OrdinalIgnoreCase)
@@ -70,7 +78,7 @@ public async Task<string> FormatAsync(HttpRequestMessage request, HttpLoggingOpt
 
         if (options.LogRequestBody)
         {
-            await AppendBodyAsync(sb, request.Content, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false);
+            await AppendBodyAsync(sb, request, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false);
         }
 
         return sb.ToString();
@@ -127,10 +135,15 @@ private static bool IsSensitiveHeader(string headerName, IReadOnlyList<string> s
         return false;
     }
 
-    private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int maxBodySize, CancellationToken cancellationToken)
+    private async Task AppendBodyAsync(
+        StringBuilder sb,
+        HttpRequestMessage request,
+        int maxBodySize,
+        CancellationToken cancellationToken)
     {
         sb.AppendLine("Body");
 
+        var content = request.Content;
         if (content == null)
         {
             sb.AppendLine("\t(null)");
@@ -146,29 +159,43 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m
             return;
         }
 
-        var body = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
+        var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader
+            .ReadReplayableAsync(content, maxBodySize, cancellationToken)
+            .ConfigureAwait(false);
+        request.Content = replayContent;
 
-        if (string.IsNullOrWhiteSpace(body))
+        if (string.IsNullOrWhiteSpace(body) && !isTruncated)
         {
             sb.AppendLine("\t(empty)");
             return;
         }
 
         // Obfuscate sensitive values in the body
-        var obfuscatedBody = _secretObfuscator.Obfuscate(body, null);
+        var obfuscatedBody = HttpBodySecretRedactor.Redact(
+            body,
+            isTruncated,
+            _secretObfuscator,
+            _secretProvider,
+            _secretMaskingOptions);
 
-        // Truncate if body is too large
-        if (maxBodySize > 0 && obfuscatedBody.Length > maxBodySize)
+        sb.AppendLine($"\t{obfuscatedBody}");
+        if (isTruncated)
         {
-            sb.AppendLine($"\t{obfuscatedBody[..maxBodySize]}");
-            sb.AppendLine($"\t... [truncated, total size: {obfuscatedBody.Length:N0} characters]");
-        }
-        else
-        {
-            sb.AppendLine($"\t{obfuscatedBody}");
+            AppendTruncationMessage(sb, maxBodySize, totalLength);
         }
     }
 
+    private static void AppendTruncationMessage(
+        StringBuilder sb,
+        int maxBodySize,
+        long? totalLength)
+    {
+        var size = totalLength.HasValue
+            ? $"total size: {totalLength.Value:N0} bytes"
+            : $"after {maxBodySize:N0} bytes";
+        sb.AppendLine($"\t... [truncated, {size}]");
+    }
+
     private static bool IsBinaryContent(HttpContent content)
     {
         var contentType = content.Headers.ContentType?.MediaType;
diff --git a/src/ModularPipelines/Http/HttpResponseExtensions.cs b/src/ModularPipelines/Http/HttpResponseExtensions.cs
index d4cfff64de2..c4123c4543b 100644
--- a/src/ModularPipelines/Http/HttpResponseExtensions.cs
+++ b/src/ModularPipelines/Http/HttpResponseExtensions.cs
@@ -30,28 +30,20 @@ public static async Task<HttpResponseMessage> EnsureSuccessStatusCodeWithContent
         {
             responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
         }
-        catch (ObjectDisposedException)
-        {
-            // Response content stream was already disposed - continue with null content
-        }
-        catch (InvalidOperationException)
-        {
-            // Content stream is in an invalid state - continue with null content
-        }
-        catch (HttpRequestException)
-        {
-            // Network error while reading content - continue with null content
-        }
         catch (OperationCanceledException)
         {
-            // Reading was cancelled - continue with null content
+            throw;
         }
         catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException))
         {
-            // Other unexpected errors reading content - continue with null content
+            cancellationToken.ThrowIfCancellationRequested();
+
+            // Errors reading content do not hide the primary HTTP status failure.
             // The primary error information is in the status code and reason phrase
         }
 
+        cancellationToken.ThrowIfCancellationRequested();
+
         throw new HttpResponseException(
             response.StatusCode,
             response.ReasonPhrase,
diff --git a/src/ModularPipelines/Http/HttpResponseFormatter.cs b/src/ModularPipelines/Http/HttpResponseFormatter.cs
index 886a17d963e..3c49af94aad 100644
--- a/src/ModularPipelines/Http/HttpResponseFormatter.cs
+++ b/src/ModularPipelines/Http/HttpResponseFormatter.cs
@@ -1,5 +1,6 @@
 using System.Net.Http.Headers;
 using System.Text;
+using Microsoft.Extensions.Options;
 using ModularPipelines.Constants;
 using ModularPipelines.Engine;
 using ModularPipelines.Options;
@@ -31,10 +32,17 @@ namespace ModularPipelines.Http;
 internal class HttpResponseFormatter : IHttpResponseFormatter
 {
     private readonly ISecretObfuscator _secretObfuscator;
+    private readonly ISecretProvider _secretProvider;
+    private readonly IOptions<SecretMaskingOptions> _secretMaskingOptions;
 
-    public HttpResponseFormatter(ISecretObfuscator secretObfuscator)
+    public HttpResponseFormatter(
+        ISecretObfuscator secretObfuscator,
+        ISecretProvider secretProvider,
+        IOptions<SecretMaskingOptions> secretMaskingOptions)
     {
         _secretObfuscator = secretObfuscator;
+        _secretProvider = secretProvider;
+        _secretMaskingOptions = secretMaskingOptions;
     }
 
     private static readonly HashSet<string> TextMediaTypes = new(StringComparer.OrdinalIgnoreCase)
@@ -70,7 +78,7 @@ public async Task<string> FormatAsync(HttpResponseMessage response, HttpLoggingO
 
         if (options.LogResponseBody)
         {
-            await AppendBodyAsync(sb, response.Content, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false);
+            await AppendBodyAsync(sb, response, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false);
         }
 
         return sb.ToString();
@@ -127,10 +135,15 @@ private static bool IsSensitiveHeader(string headerName, IReadOnlyList<string> s
         return false;
     }
 
-    private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int maxBodySize, CancellationToken cancellationToken)
+    private async Task AppendBodyAsync(
+        StringBuilder sb,
+        HttpResponseMessage response,
+        int maxBodySize,
+        CancellationToken cancellationToken)
     {
         sb.AppendLine("Body");
 
+        var content = response.Content;
         if (content == null)
         {
             sb.AppendLine("\t(null)");
@@ -146,29 +159,43 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m
             return;
         }
 
-        var body = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
+        var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader
+            .ReadAsync(content, maxBodySize, cancellationToken)
+            .ConfigureAwait(false);
+        response.Content = replayContent;
 
-        if (string.IsNullOrWhiteSpace(body))
+        if (string.IsNullOrWhiteSpace(body) && !isTruncated)
         {
             sb.AppendLine("\t(empty)");
             return;
         }
 
         // Obfuscate sensitive values in the body
-        var obfuscatedBody = _secretObfuscator.Obfuscate(body, null);
+        var obfuscatedBody = HttpBodySecretRedactor.Redact(
+            body,
+            isTruncated,
+            _secretObfuscator,
+            _secretProvider,
+            _secretMaskingOptions);
 
-        // Truncate if body is too large
-        if (maxBodySize > 0 && obfuscatedBody.Length > maxBodySize)
+        sb.AppendLine($"\t{obfuscatedBody}");
+        if (isTruncated)
         {
-            sb.AppendLine($"\t{obfuscatedBody[..maxBodySize]}");
-            sb.AppendLine($"\t... [truncated, total size: {obfuscatedBody.Length:N0} characters]");
-        }
-        else
-        {
-            sb.AppendLine($"\t{obfuscatedBody}");
+            AppendTruncationMessage(sb, maxBodySize, totalLength);
         }
     }
 
+    private static void AppendTruncationMessage(
+        StringBuilder sb,
+        int maxBodySize,
+        long? totalLength)
+    {
+        var size = totalLength.HasValue
+            ? $"total size: {totalLength.Value:N0} bytes"
+            : $"after {maxBodySize:N0} bytes";
+        sb.AppendLine($"\t... [truncated, {size}]");
+    }
+
     private static bool IsBinaryContent(HttpContent content)
     {
         var contentType = content.Headers.ContentType?.MediaType;
diff --git a/src/ModularPipelines/Http/IHttpLogger.cs b/src/ModularPipelines/Http/IHttpLogger.cs
index 25699c97a10..bd075570f92 100644
--- a/src/ModularPipelines/Http/IHttpLogger.cs
+++ b/src/ModularPipelines/Http/IHttpLogger.cs
@@ -17,6 +17,21 @@ public interface IHttpLogger
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     Task PrintRequest(HttpRequestMessage request, IModuleLogger logger);
 
+    /// <summary>
+    /// Prints the HTTP request.
+    /// </summary>
+    /// <param name="request">The HTTP request to print.</param>
+    /// <param name="logger">The current module logger.</param>
+    /// <param name="cancellationToken">A token to cancel request formatting.</param>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    Task PrintRequest(
+        HttpRequestMessage request,
+        IModuleLogger logger,
+        CancellationToken cancellationToken)
+    {
+        return PrintRequest(request, logger).WaitAsync(cancellationToken);
+    }
+
     /// <summary>
     /// Prints the HTTP request with logging options.
     /// </summary>
@@ -26,6 +41,23 @@ public interface IHttpLogger
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options);
 
+    /// <summary>
+    /// Prints the HTTP request with logging options.
+    /// </summary>
+    /// <param name="request">The HTTP request to print.</param>
+    /// <param name="logger">The current module logger.</param>
+    /// <param name="options">Options controlling what parts of the request to log.</param>
+    /// <param name="cancellationToken">A token to cancel request formatting.</param>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    Task PrintRequest(
+        HttpRequestMessage request,
+        IModuleLogger logger,
+        HttpLoggingOptions options,
+        CancellationToken cancellationToken)
+    {
+        return PrintRequest(request, logger, options).WaitAsync(cancellationToken);
+    }
+
     /// <summary>
     /// Prints the HTTP response.
     /// </summary>
@@ -34,6 +66,21 @@ public interface IHttpLogger
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     Task PrintResponse(HttpResponseMessage response, IModuleLogger logger);
 
+    /// <summary>
+    /// Prints the HTTP response.
+    /// </summary>
+    /// <param name="response">The HTTP response to print.</param>
+    /// <param name="logger">The current module logger.</param>
+    /// <param name="cancellationToken">A token to cancel response formatting.</param>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    Task PrintResponse(
+        HttpResponseMessage response,
+        IModuleLogger logger,
+        CancellationToken cancellationToken)
+    {
+        return PrintResponse(response, logger).WaitAsync(cancellationToken);
+    }
+
     /// <summary>
     /// Prints the HTTP response with logging options.
     /// </summary>
@@ -43,6 +90,23 @@ public interface IHttpLogger
     /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
     Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options);
 
+    /// <summary>
+    /// Prints the HTTP response with logging options.
+    /// </summary>
+    /// <param name="response">The HTTP response to print.</param>
+    /// <param name="logger">The current module logger.</param>
+    /// <param name="options">Options controlling what parts of the response to log.</param>
+    /// <param name="cancellationToken">A token to cancel response formatting.</param>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    Task PrintResponse(
+        HttpResponseMessage response,
+        IModuleLogger logger,
+        HttpLoggingOptions options,
+        CancellationToken cancellationToken)
+    {
+        return PrintResponse(response, logger, options).WaitAsync(cancellationToken);
+    }
+
     /// <summary>
     /// Prints the HTTP status code.
     /// </summary>
diff --git a/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs b/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs
index 21aa4271df1..8c3603e90fb 100644
--- a/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs
+++ b/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs
@@ -17,10 +17,12 @@ public RequestLoggingHttpHandler(IModuleLoggerProvider loggerProvider, IHttpLogg
     protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
     {
         var logger = _loggerProvider.GetLogger();
-        await _httpLogger.PrintRequest(request, logger).ConfigureAwait(false);
+        await _httpLogger
+            .PrintRequest(request, logger, cancellationToken)
+            .ConfigureAwait(false);
 
         var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false);
 
         return response;
     }
-}
\ No newline at end of file
+}
diff --git a/src/ModularPipelines/Http/ResilienceHttpHandler.cs b/src/ModularPipelines/Http/ResilienceHttpHandler.cs
index 9774c8c6c45..49561e2c2fa 100644
--- a/src/ModularPipelines/Http/ResilienceHttpHandler.cs
+++ b/src/ModularPipelines/Http/ResilienceHttpHandler.cs
@@ -105,14 +105,21 @@ private Task OnRetry(DelegateResult<HttpResponseMessage> outcome, TimeSpan delay
                 outcome.Exception.GetType().Name,
                 outcome.Exception.Message,
                 retryAttempt,
-                (int)delay.TotalMilliseconds);
+                (int) delay.TotalMilliseconds);
         }
         else if (outcome.Result != null)
         {
-            logger.LogWarning("HTTP request returned {StatusCode}. Retry attempt {RetryAttempt} after {Delay}ms",
-                (int)outcome.Result.StatusCode,
-                retryAttempt,
-                (int)delay.TotalMilliseconds);
+            try
+            {
+                logger.LogWarning("HTTP request returned {StatusCode}. Retry attempt {RetryAttempt} after {Delay}ms",
+                    (int) outcome.Result.StatusCode,
+                    retryAttempt,
+                    (int) delay.TotalMilliseconds);
+            }
+            finally
+            {
+                outcome.Result.Dispose();
+            }
         }
 
         return Task.CompletedTask;
diff --git a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs
index d1d9fce4ebe..df3a9c34790 100644
--- a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs
+++ b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs
@@ -18,14 +18,18 @@ protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage
     {
         var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false);
 
-        // Buffer the response content so it can be read multiple times.
-        // Without this, reading the body for logging consumes the stream,
-        // making it unreadable by subsequent code. See issue #1610.
-        await response.Content.LoadIntoBufferAsync().ConfigureAwait(false);
-
-        var logger = _loggerProvider.GetLogger();
-        await _httpLogger.PrintResponse(response, logger).ConfigureAwait(false);
-
-        return response;
+        try
+        {
+            var logger = _loggerProvider.GetLogger();
+            await _httpLogger
+                .PrintResponse(response, logger, cancellationToken)
+                .ConfigureAwait(false);
+            return response;
+        }
+        catch
+        {
+            response.Dispose();
+            throw;
+        }
     }
 }
diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs
new file mode 100644
index 00000000000..1c7dbcf5ed2
--- /dev/null
+++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs
@@ -0,0 +1,300 @@
+using System.Net;
+
+namespace ModularPipelines.Http;
+
+internal sealed class TimeoutHttpContent : HttpContent
+{
+    private readonly HttpContent _innerContent;
+    private readonly CancellationTokenSource? _timeoutCancellationTokenSource;
+    private readonly CancellationTokenSource _linkedCancellationTokenSource;
+
+    public TimeoutHttpContent(
+        HttpContent innerContent,
+        CancellationTokenSource? timeoutCancellationTokenSource,
+        CancellationTokenSource linkedCancellationTokenSource)
+    {
+        _innerContent = innerContent;
+        _timeoutCancellationTokenSource = timeoutCancellationTokenSource;
+        _linkedCancellationTokenSource = linkedCancellationTokenSource;
+
+        foreach (var header in innerContent.Headers)
+        {
+            Headers.TryAddWithoutValidation(header.Key, header.Value);
+        }
+    }
+
+    protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context)
+    {
+        return CopyInnerContentToAsync(stream, CancellationToken.None);
+    }
+
+    protected override async Task SerializeToStreamAsync(
+        Stream stream,
+        TransportContext? context,
+        CancellationToken cancellationToken)
+    {
+        await CopyInnerContentToAsync(stream, cancellationToken).ConfigureAwait(false);
+    }
+
+    protected override async Task<Stream> CreateContentReadStreamAsync()
+    {
+        _linkedCancellationTokenSource.Token.ThrowIfCancellationRequested();
+        var stream = await _innerContent
+            .ReadAsStreamAsync(_linkedCancellationTokenSource.Token)
+            .ConfigureAwait(false);
+        return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token);
+    }
+
+    protected override Stream CreateContentReadStream(CancellationToken cancellationToken)
+    {
+        using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken);
+        var effectiveCancellationToken =
+            readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token;
+        effectiveCancellationToken.ThrowIfCancellationRequested();
+        var stream = _innerContent.ReadAsStream(effectiveCancellationToken);
+        return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token);
+    }
+
+    protected override async Task<Stream> CreateContentReadStreamAsync(
+        CancellationToken cancellationToken)
+    {
+        using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken);
+        var effectiveCancellationToken =
+            readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token;
+        effectiveCancellationToken.ThrowIfCancellationRequested();
+        var stream = await _innerContent
+            .ReadAsStreamAsync(effectiveCancellationToken)
+            .ConfigureAwait(false);
+        return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token);
+    }
+
+    protected override bool TryComputeLength(out long length)
+    {
+        if (_innerContent.Headers.ContentLength is { } contentLength)
+        {
+            length = contentLength;
+            return true;
+        }
+
+        length = 0;
+        return false;
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _innerContent.Dispose();
+            _linkedCancellationTokenSource.Dispose();
+            _timeoutCancellationTokenSource?.Dispose();
+        }
+
+        base.Dispose(disposing);
+    }
+
+    private async Task CopyInnerContentToAsync(
+        Stream destination,
+        CancellationToken cancellationToken)
+    {
+        using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken);
+        var effectiveCancellationToken =
+            readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token;
+        effectiveCancellationToken.ThrowIfCancellationRequested();
+        var source = await _innerContent
+            .ReadAsStreamAsync(effectiveCancellationToken)
+            .ConfigureAwait(false);
+        using var cancellationRegistration = effectiveCancellationToken.Register(
+            static state => ((Stream) state!).Dispose(),
+            source);
+
+        try
+        {
+            await source.CopyToAsync(destination, effectiveCancellationToken).ConfigureAwait(false);
+            effectiveCancellationToken.ThrowIfCancellationRequested();
+        }
+        catch (Exception exception)
+            when (effectiveCancellationToken.IsCancellationRequested
+                  && exception is ObjectDisposedException or IOException)
+        {
+            throw new OperationCanceledException(
+                "The HTTP response body copy was cancelled.",
+                exception,
+                effectiveCancellationToken);
+        }
+    }
+
+    private CancellationTokenSource? CreateReadCancellationTokenSource(
+        CancellationToken cancellationToken)
+    {
+        return cancellationToken.CanBeCanceled &&
+               cancellationToken != _linkedCancellationTokenSource.Token
+            ? CancellationTokenSource.CreateLinkedTokenSource(
+                cancellationToken,
+                _linkedCancellationTokenSource.Token)
+            : null;
+    }
+
+    private sealed class TimeoutReadStream(
+        Stream innerStream,
+        CancellationToken timeoutCancellationToken) : Stream
+    {
+        private readonly CancellationTokenRegistration _timeoutRegistration =
+            timeoutCancellationToken.Register(
+                static state => ((Stream) state!).Dispose(),
+                innerStream);
+
+        public override bool CanRead => innerStream.CanRead;
+
+        public override bool CanSeek => innerStream.CanSeek;
+
+        public override bool CanWrite => innerStream.CanWrite;
+
+        public override long Length => innerStream.Length;
+
+        public override long Position
+        {
+            get => innerStream.Position;
+            set => innerStream.Position = value;
+        }
+
+        public override void Flush() => innerStream.Flush();
+
+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            return innerStream.FlushAsync(cancellationToken);
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            timeoutCancellationToken.ThrowIfCancellationRequested();
+            try
+            {
+                var bytesRead = innerStream.Read(buffer, offset, count);
+                timeoutCancellationToken.ThrowIfCancellationRequested();
+                return bytesRead;
+            }
+            catch (Exception exception)
+                when (timeoutCancellationToken.IsCancellationRequested &&
+                      exception is ObjectDisposedException or IOException)
+            {
+                throw CreateTimeoutException(exception);
+            }
+        }
+
+        public override int Read(Span<byte> buffer)
+        {
+            timeoutCancellationToken.ThrowIfCancellationRequested();
+            try
+            {
+                var bytesRead = innerStream.Read(buffer);
+                timeoutCancellationToken.ThrowIfCancellationRequested();
+                return bytesRead;
+            }
+            catch (Exception exception)
+                when (timeoutCancellationToken.IsCancellationRequested &&
+                      exception is ObjectDisposedException or IOException)
+            {
+                throw CreateTimeoutException(exception);
+            }
+        }
+
+        public override Task<int> ReadAsync(
+            byte[] buffer,
+            int offset,
+            int count,
+            CancellationToken cancellationToken)
+        {
+            return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
+        }
+
+        public override async ValueTask<int> ReadAsync(
+            Memory<byte> buffer,
+            CancellationToken cancellationToken = default)
+        {
+            using var readCancellationTokenSource =
+                CreateReadCancellationTokenSource(cancellationToken);
+            var effectiveCancellationToken =
+                readCancellationTokenSource?.Token ?? timeoutCancellationToken;
+            effectiveCancellationToken.ThrowIfCancellationRequested();
+            using var cancellationRegistration = readCancellationTokenSource?.Token.Register(
+                static state => ((Stream) state!).Dispose(),
+                innerStream);
+            try
+            {
+                var bytesRead = await innerStream.ReadAsync(
+                        buffer,
+                        effectiveCancellationToken)
+                    .ConfigureAwait(false);
+                effectiveCancellationToken.ThrowIfCancellationRequested();
+                return bytesRead;
+            }
+            catch (Exception exception)
+                when (effectiveCancellationToken.IsCancellationRequested &&
+                      exception is ObjectDisposedException or IOException)
+            {
+                throw CreateReadCancellationException(
+                    exception,
+                    effectiveCancellationToken);
+            }
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            return innerStream.Seek(offset, origin);
+        }
+
+        public override void SetLength(long value) => innerStream.SetLength(value);
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            innerStream.Write(buffer, offset, count);
+        }
+
+        public override async ValueTask DisposeAsync()
+        {
+            _timeoutRegistration.Dispose();
+            await innerStream.DisposeAsync().ConfigureAwait(false);
+            GC.SuppressFinalize(this);
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                _timeoutRegistration.Dispose();
+                innerStream.Dispose();
+            }
+
+            base.Dispose(disposing);
+        }
+
+        private CancellationTokenSource? CreateReadCancellationTokenSource(
+            CancellationToken cancellationToken)
+        {
+            return cancellationToken.CanBeCanceled &&
+                   cancellationToken != timeoutCancellationToken
+                ? CancellationTokenSource.CreateLinkedTokenSource(
+                    cancellationToken,
+                    timeoutCancellationToken)
+                : null;
+        }
+
+        private OperationCanceledException CreateTimeoutException(Exception exception)
+        {
+            return new OperationCanceledException(
+                "The HTTP response body read timed out.",
+                exception,
+                timeoutCancellationToken);
+        }
+
+        private static OperationCanceledException CreateReadCancellationException(
+            Exception exception,
+            CancellationToken cancellationToken)
+        {
+            return new OperationCanceledException(
+                "The HTTP response body read was cancelled.",
+                exception,
+                cancellationToken);
+        }
+    }
+}
diff --git a/src/ModularPipelines/Options/DownloadOptions.cs b/src/ModularPipelines/Options/DownloadOptions.cs
index 2b169509714..601eda114df 100644
--- a/src/ModularPipelines/Options/DownloadOptions.cs
+++ b/src/ModularPipelines/Options/DownloadOptions.cs
@@ -21,5 +21,6 @@ public record DownloadOptions(Uri DownloadUri)
     /// <summary>
     /// Gets the type of HTTP logging to perform during the download.
     /// </summary>
-    public HttpLoggingType LoggingType { get; init; } = HttpLoggingType.Request | HttpLoggingType.Response | HttpLoggingType.StatusCode | HttpLoggingType.Duration;
-}
\ No newline at end of file
+    public HttpLoggingType LoggingType { get; init; } =
+        HttpLoggingType.Request | HttpLoggingType.StatusCode | HttpLoggingType.Duration;
+}
diff --git a/src/ModularPipelines/Options/HttpLoggingOptions.cs b/src/ModularPipelines/Options/HttpLoggingOptions.cs
index 7f364cef51a..916a4e55cc5 100644
--- a/src/ModularPipelines/Options/HttpLoggingOptions.cs
+++ b/src/ModularPipelines/Options/HttpLoggingOptions.cs
@@ -67,7 +67,7 @@ public record HttpLoggingOptions
     public bool LogDuration { get; init; } = true;
 
     /// <summary>
-    /// Gets the maximum body size in characters to log. Default is 4096.
+    /// Gets the maximum body size in bytes to read and log. Default is 4096.
     /// Bodies larger than this will be truncated with a message indicating the full size.
     /// Set to 0 or negative to disable truncation.
     /// </summary>
diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs
new file mode 100644
index 00000000000..3e0aed2146b
--- /dev/null
+++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs
@@ -0,0 +1,286 @@
+using System.Net.Http.Headers;
+using System.Text;
+using ModularPipelines.Constants;
+using ModularPipelines.Engine;
+using ModularPipelines.Http;
+using ModularPipelines.Options;
+using Moq;
+
+namespace ModularPipelines.UnitTests.Context;
+
+public class HttpFormatterTests
+{
+    [Test]
+    public async Task ResponseFormatter_BoundsPreviewReadAndPreservesBody()
+    {
+        const string body = "abcdefghijklmnopqrstuvwxyz";
+        var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body));
+        using var response = new HttpResponseMessage
+        {
+            Content = CreateTextContent(stream),
+        };
+        var formatter = CreateResponseFormatter();
+
+        var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 5,
+        });
+        var bytesReadForPreview = stream.BytesRead;
+        var replayedBody = await response.Content.ReadAsStringAsync();
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(bytesReadForPreview).IsLessThanOrEqualTo(6);
+            await Assert.That(formatted).Contains("\tabcde");
+            await Assert.That(formatted).DoesNotContain("abcdef");
+            await Assert.That(replayedBody).IsEqualTo(body);
+        }
+    }
+
+    [Test]
+    public async Task ResponseFormatter_DoesNotAllocateEntireLargePreviewLimit()
+    {
+        var stream = new CountingReadStream([]);
+        using var response = new HttpResponseMessage
+        {
+            Content = CreateTextContent(stream),
+        };
+        var formatter = CreateResponseFormatter();
+
+        await formatter.FormatAsync(response, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 64 * 1024 * 1024,
+        });
+
+        await Assert.That(stream.MaximumRequestedReadSize).IsLessThanOrEqualTo(81920);
+    }
+
+    [Test]
+    public async Task RequestFormatter_PreservesBodyForRepeatedSends()
+    {
+        const string body = "abcdefghijklmnopqrstuvwxyz";
+        var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body));
+        var content = CreateTextContent(stream);
+        using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test")
+        {
+            Content = content,
+        };
+        var formatter = CreateRequestFormatter();
+
+        var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 5,
+        });
+        using var firstSend = new MemoryStream();
+        using var secondSend = new MemoryStream();
+        await request.Content.CopyToAsync(firstSend);
+        await request.Content.CopyToAsync(secondSend);
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(formatted).Contains("\tabcde");
+            await Assert.That(formatted).DoesNotContain("abcdef");
+            await Assert.That(Encoding.UTF8.GetString(firstSend.ToArray())).IsEqualTo(body);
+            await Assert.That(Encoding.UTF8.GetString(secondSend.ToArray())).IsEqualTo(body);
+            await Assert.That(stream.IsDisposed).IsTrue();
+            await Assert.That(content.DisposeCount).IsEqualTo(1);
+        }
+    }
+
+    [Test]
+    public async Task ResponseFormatter_UnboundedPreviewPreservesBody()
+    {
+        const string body = "abcdefghijklmnopqrstuvwxyz";
+        using var response = new HttpResponseMessage
+        {
+            Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes(body))),
+        };
+        var formatter = CreateResponseFormatter();
+
+        var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 0,
+        });
+        var replayedBody = await response.Content.ReadAsStringAsync();
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(formatted).Contains(body);
+            await Assert.That(replayedBody).IsEqualTo(body);
+        }
+    }
+
+    [Test]
+    public async Task RequestFormatter_UnboundedPreviewPreservesBody()
+    {
+        const string body = "abcdefghijklmnopqrstuvwxyz";
+        using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test")
+        {
+            Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes(body))),
+        };
+        var formatter = CreateRequestFormatter();
+
+        var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 0,
+        });
+        var replayedBody = await request.Content.ReadAsStringAsync();
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(formatted).Contains(body);
+            await Assert.That(replayedBody).IsEqualTo(body);
+        }
+    }
+
+    [Test]
+    public async Task ResponseFormatter_RedactsSecretCrossingPreviewBoundary()
+    {
+        const string secret = "secret-value";
+        using var response = new HttpResponseMessage
+        {
+            Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))),
+        };
+        var formatter = CreateResponseFormatter(secret);
+
+        var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 13,
+        });
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}");
+            await Assert.That(formatted).DoesNotContain("prefix-secret");
+        }
+    }
+
+    [Test]
+    public async Task RequestFormatter_RedactsSecretCrossingPreviewBoundary()
+    {
+        const string secret = "secret-value";
+        using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test")
+        {
+            Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))),
+        };
+        var formatter = CreateRequestFormatter(secret);
+
+        var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 13,
+        });
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}");
+            await Assert.That(formatted).DoesNotContain("prefix-secret");
+        }
+    }
+
+    [Test]
+    public async Task ResponseFormatter_RedactsSecretWhenPreviewSplitsMultibyteCharacter()
+    {
+        const string secret = "abcédef";
+        using var response = new HttpResponseMessage
+        {
+            Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))),
+        };
+        var formatter = CreateResponseFormatter(secret);
+
+        var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions
+        {
+            MaxBodySizeToLog = 11,
+        });
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}");
+            await Assert.That(formatted).DoesNotContain("prefix-abc");
+        }
+    }
+
+    private static CountingStreamContent CreateTextContent(Stream stream)
+    {
+        var content = new CountingStreamContent(stream);
+        content.Headers.ContentType = new MediaTypeHeaderValue("text/plain")
+        {
+            CharSet = "utf-8",
+        };
+        return content;
+    }
+
+    private sealed class CountingStreamContent(Stream stream) : StreamContent(stream)
+    {
+        public int DisposeCount { get; private set; }
+
+        protected override void Dispose(bool disposing)
+        {
+            DisposeCount++;
+            base.Dispose(disposing);
+        }
+    }
+
+    private static ISecretObfuscator CreateObfuscator(params string[] secrets)
+    {
+        var obfuscator = new Mock<ISecretObfuscator>();
+        obfuscator
+            .Setup(x => x.Obfuscate(It.IsAny<string?>(), It.IsAny<object?>()))
+            .Returns((string? input, object? _) => secrets.Aggregate(
+                input ?? string.Empty,
+                (value, secret) => value.Replace(
+                    secret,
+                    LoggingConstants.SecretMask,
+                    StringComparison.Ordinal)));
+        return obfuscator.Object;
+    }
+
+    private static HttpRequestFormatter CreateRequestFormatter(params string[] secrets)
+    {
+        return new HttpRequestFormatter(
+            CreateObfuscator(secrets),
+            CreateSecretProvider(secrets),
+            Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()));
+    }
+
+    private static HttpResponseFormatter CreateResponseFormatter(params string[] secrets)
+    {
+        return new HttpResponseFormatter(
+            CreateObfuscator(secrets),
+            CreateSecretProvider(secrets),
+            Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()));
+    }
+
+    private static ISecretProvider CreateSecretProvider(string[] secrets)
+    {
+        var secretProvider = new Mock<ISecretProvider>();
+        secretProvider
+            .Setup(x => x.GetSnapshot())
+            .Returns(new SecretSnapshot(0, secrets));
+        return secretProvider.Object;
+    }
+
+    private sealed class CountingReadStream(byte[] contents) : MemoryStream(contents)
+    {
+        public int BytesRead { get; private set; }
+
+        public int MaximumRequestedReadSize { get; private set; }
+
+        public bool IsDisposed { get; private set; }
+
+        public override async ValueTask<int> ReadAsync(
+            Memory<byte> buffer,
+            CancellationToken cancellationToken = default)
+        {
+            MaximumRequestedReadSize = Math.Max(MaximumRequestedReadSize, buffer.Length);
+            var read = await base.ReadAsync(buffer, cancellationToken);
+            BytesRead += read;
+            return read;
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            IsDisposed = true;
+            base.Dispose(disposing);
+        }
+    }
+}
diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs
index 2873eeb4093..8bbbfeafb67 100644
--- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs
+++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs
@@ -1,11 +1,17 @@
+using System.Net;
+using System.Net.Http.Headers;
 using System.Reflection;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using ModularPipelines.Context.Domains.Network;
+using ModularPipelines.Engine;
+using ModularPipelines.Extensions;
 using ModularPipelines.Http;
+using ModularPipelines.Logging;
 using ModularPipelines.Options;
 using ModularPipelines.TestHelpers;
 using ModularPipelines.UnitTests.Helpers;
+using Moq;
 using NReco.Logging.File;
 using File = System.IO.File;
 using LogLevel = Microsoft.Extensions.Logging.LogLevel;
@@ -14,6 +20,665 @@ namespace ModularPipelines.UnitTests.Context;
 
 public class HttpTests : TestBase
 {
+    [Test]
+    public async Task SendAsync_ReturnsAfterHeadersWithoutBufferingResponseBody()
+    {
+        var content = new BlockingHttpContent();
+        var handler = new ImmediateResponseHandler(content);
+        using var httpClient = new HttpClient(handler);
+        var result = await GetService<IHttpContext>((_, _) => { });
+        var sendTask = result.T.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/large-file"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+        });
+
+        await handler.RequestReceived.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        HttpResponseMessage? response = null;
+        try
+        {
+            response = await sendTask.WaitAsync(TimeSpan.FromSeconds(1));
+        }
+        finally
+        {
+            content.Release();
+            response ??= await sendTask;
+            response.Dispose();
+        }
+    }
+
+    [Test]
+    [Arguments(true)]
+    [Arguments(false)]
+    public async Task SendAsync_ConfiguredTimeoutCancelsStreamedBody(bool useRequestTimeout)
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream();
+        using var httpClient = new HttpClient(new ImmediateResponseHandler(new StreamContent(contentStream)));
+        var result = await GetService<IHttpContext>((builder, _) =>
+            builder.ConfigurePipelineOptions(options => options with
+            {
+                DefaultHttpTimeout = useRequestTimeout ? null : timeout,
+            }));
+        using var response = await result.T.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-body"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+            Timeout = useRequestTimeout ? timeout : null,
+        });
+        var stream = await response.Content.ReadAsStreamAsync();
+        var readTask = stream.ReadAsync(new byte[1]).AsTask();
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(
+                async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    [Arguments(true)]
+    [Arguments(false)]
+    public async Task SendAsync_HttpClientTimeoutCancelsStreamedBody(bool useCustomClient)
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream();
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)))
+        {
+            Timeout = timeout,
+        };
+        var httpClientFactory = new Mock<IHttpClientFactory>();
+        httpClientFactory
+            .Setup(x => x.CreateClient(It.IsAny<string>()))
+            .Returns(httpClient);
+        var http = new ModularPipelines.Http.Http(
+            httpClientFactory.Object,
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/client-timeout"))
+        {
+            HttpClient = useCustomClient ? httpClient : null,
+            LoggingType = HttpLoggingType.None,
+        });
+        var stream = response.Content.ReadAsStream();
+        var readTask = stream.ReadAsync(new byte[1]).AsTask();
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(
+                async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    public async Task SendAsync_InfiniteTimeoutRetainsCallerCancellationForStreamedBody()
+    {
+        var contentStream = new BlockingReadStream();
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)))
+        {
+            Timeout = Timeout.InfiniteTimeSpan,
+        };
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var cancellationTokenSource = new CancellationTokenSource();
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/caller-cancellation"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+        }, cancellationTokenSource.Token);
+        var stream = response.Content.ReadAsStream();
+        var readTask = stream.ReadAsync(new byte[1]).AsTask();
+
+        cancellationTokenSource.Cancel();
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(
+                async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    [Arguments(true)]
+    [Arguments(false)]
+    public async Task SendAsync_NormalizesAsyncStreamFailureAfterTimeout(
+        bool throwIOExceptionWhenDisposed)
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(
+            throwIOExceptionWhenDisposed,
+            ignoreAsyncCancellation: true);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/async-timeout"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+            Timeout = timeout,
+        });
+        var stream = response.Content.ReadAsStream();
+        var readTask = stream.ReadAsync(new byte[1]).AsTask();
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(
+                async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    [Arguments(true)]
+    [Arguments(false)]
+    public async Task SendAsync_NormalizesAsyncStreamFailureAfterPerReadCancellation(
+        bool throwIOExceptionWhenDisposed)
+    {
+        var contentStream = new BlockingReadStream(
+            throwIOExceptionWhenDisposed,
+            ignoreAsyncCancellation: true);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)))
+        {
+            Timeout = Timeout.InfiniteTimeSpan,
+        };
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/read-cancellation"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+            Timeout = TimeSpan.FromMinutes(1),
+        });
+        var stream = response.Content.ReadAsStream();
+        using var cancellationTokenSource = new CancellationTokenSource();
+        var readTask = stream
+            .ReadAsync(new byte[1], cancellationTokenSource.Token)
+            .AsTask();
+
+        await contentStream.ReadStarted.WaitAsync(TimeSpan.FromSeconds(1));
+        cancellationTokenSource.Cancel();
+
+        try
+        {
+            var exception = await Assert.ThrowsAsync<OperationCanceledException>(
+                async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+            await Assert.That(exception!.CancellationToken.IsCancellationRequested).IsTrue();
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    [Arguments(false)]
+    [Arguments(true)]
+    public async Task SendAsync_DoesNotTreatCancellationAsSuccessfulEndOfStream(
+        bool useBufferedCopy)
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(
+            ignoreAsyncCancellation: true,
+            returnEofWhenDisposed: true);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/cancelled-eof"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+            Timeout = timeout,
+        });
+
+        Task readTask;
+        if (useBufferedCopy)
+        {
+            readTask = response.Content.ReadAsStringAsync();
+        }
+        else
+        {
+            var stream = await response.Content.ReadAsStreamAsync();
+            readTask = stream.ReadAsync(new byte[1]).AsTask();
+        }
+
+        await Assert.ThrowsAsync<OperationCanceledException>(
+            async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+    }
+
+    [Test]
+    public async Task SendAsync_CancelsLegacyResponseLogger()
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            new LegacyBodyLogger(logRequest: false),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+
+        await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+            await http.SendAsync(new HttpOptions(
+                    new HttpRequestMessage(HttpMethod.Get, "https://example.test/legacy-logger"))
+                {
+                    HttpClient = httpClient,
+                    LoggingType = HttpLoggingType.Response,
+                    Timeout = timeout,
+                })
+                .WaitAsync(TimeSpan.FromSeconds(1)));
+    }
+
+    [Test]
+    public async Task SendAsync_CancelsLegacyRequestLogger()
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StringContent("response")));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            new LegacyBodyLogger(logRequest: true),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var request = new HttpRequestMessage(
+            HttpMethod.Post,
+            "https://example.test/legacy-request-logger")
+        {
+            Content = new StreamContent(contentStream),
+        };
+
+        await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+            await http.SendAsync(new HttpOptions(request)
+                {
+                    HttpClient = httpClient,
+                    LoggingType = HttpLoggingType.Request,
+                    Timeout = timeout,
+                })
+                .WaitAsync(TimeSpan.FromSeconds(1)));
+    }
+
+    [Test]
+    [Arguments(true, true, false)]
+    [Arguments(true, false, false)]
+    [Arguments(false, true, false)]
+    [Arguments(false, false, false)]
+    [Arguments(true, false, true)]
+    [Arguments(false, false, true)]
+    public async Task SendAsync_ConfiguredTimeoutInterruptsSynchronousBodyRead(
+        bool useCopyTo,
+        bool throwIOExceptionWhenDisposed,
+        bool returnEofWhenDisposed)
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(
+            throwIOExceptionWhenDisposed,
+            returnEofWhenDisposed: returnEofWhenDisposed);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)));
+        var result = await GetService<IHttpContext>((_, _) => { });
+        using var response = await result.T.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/synchronous-read"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+            Timeout = timeout,
+        });
+        var stream = await response.Content.ReadAsStreamAsync();
+        var readTask = Task.Run(() =>
+        {
+            if (useCopyTo)
+            {
+                stream.CopyTo(Stream.Null);
+            }
+            else
+            {
+                stream.ReadExactly(new byte[1]);
+            }
+        });
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(
+                async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    public async Task SendAsync_CustomClientLogsRawContentBeforeApplyingBodyTimeout()
+    {
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(new MemoryStream([1, 2, 3]))))
+        {
+            Timeout = TimeSpan.FromSeconds(5),
+        };
+        Type? loggedContentType = null;
+        var httpLogger = new Mock<IHttpLogger>();
+        httpLogger
+            .Setup(x => x.PrintResponse(
+                It.IsAny<HttpResponseMessage>(),
+                It.IsAny<IModuleLogger>(),
+                It.IsAny<HttpLoggingOptions>(),
+                It.IsAny<CancellationToken>()))
+            .Callback<HttpResponseMessage, IModuleLogger, HttpLoggingOptions, CancellationToken>(
+                (response, _, _, _) => loggedContentType = response.Content.GetType())
+            .Returns(Task.CompletedTask);
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            httpLogger.Object,
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/binary"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.Response,
+        });
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(loggedContentType).IsEqualTo(typeof(StreamContent));
+            await Assert.That(response.Content).IsTypeOf<TimeoutHttpContent>();
+        }
+    }
+
+    [Test]
+    [Arguments(4096, false)]
+    [Arguments(0, false)]
+    [Arguments(0, true)]
+    public async Task SendAsync_CustomClientTimeoutInterruptsNonCooperativeResponseLogging(
+        int maxBodySizeToLog,
+        bool returnEofWhenDisposed)
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(
+            ignoreAsyncCancellation: true,
+            returnEofWhenDisposed: returnEofWhenDisposed);
+        var content = new StreamContent(contentStream);
+        content.Headers.ContentType = new MediaTypeHeaderValue("text/plain");
+        using var httpClient = new HttpClient(new ImmediateResponseHandler(content));
+        var moduleLoggerProvider = new Mock<IModuleLoggerProvider>();
+        moduleLoggerProvider
+            .Setup(x => x.GetLogger())
+            .Returns(Mock.Of<IModuleLogger>());
+        var httpLogger = new HttpLogger(
+            Mock.Of<IHttpRequestFormatter>(),
+            new HttpResponseFormatter(
+                Mock.Of<ISecretObfuscator>(),
+                Mock.Of<ISecretProvider>(),
+                Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            moduleLoggerProvider.Object,
+            httpLogger,
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+
+        await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+            await http.SendAsync(new HttpOptions(
+                    new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-custom-log-body"))
+                {
+                    HttpClient = httpClient,
+                    LoggingType = HttpLoggingType.Response,
+                    LogSettings = new HttpLoggingOptions
+                    {
+                        MaxBodySizeToLog = maxBodySizeToLog,
+                    },
+                    Timeout = timeout,
+                })
+                .WaitAsync(TimeSpan.FromSeconds(1)));
+    }
+
+    [Test]
+    public async Task SendAsync_ConfiguredTimeoutInterruptsBufferedBodyCopy()
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true);
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(new StreamContent(contentStream)));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/buffered-timeout"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.None,
+            Timeout = timeout,
+        });
+
+        await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+            await response.Content.ReadAsStringAsync().WaitAsync(TimeSpan.FromSeconds(1)));
+    }
+
+    [Test]
+    [Arguments(true, true)]
+    [Arguments(true, false)]
+    [Arguments(false, true)]
+    [Arguments(false, false)]
+    public async Task SendAsync_PreservesCancellationWhileReadingErrorContent(
+        bool useCustomClient,
+        bool useConfiguredTimeout)
+    {
+        var contentStream = new BlockingReadStream();
+        using var httpClient = new HttpClient(
+            new ImmediateResponseHandler(
+                new StreamContent(contentStream),
+                HttpStatusCode.InternalServerError))
+        {
+            Timeout = Timeout.InfiniteTimeSpan,
+        };
+        var httpClientFactory = new Mock<IHttpClientFactory>();
+        httpClientFactory
+            .Setup(x => x.CreateClient(It.IsAny<string>()))
+            .Returns(httpClient);
+        var http = new ModularPipelines.Http.Http(
+            httpClientFactory.Object,
+            Mock.Of<IModuleLoggerProvider>(),
+            Mock.Of<IHttpLogger>(),
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var cancellationTokenSource = new CancellationTokenSource();
+        cancellationTokenSource.CancelAfter(TimeSpan.FromMilliseconds(100));
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+                await http.SendAsync(new HttpOptions(
+                        new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-error-body"))
+                    {
+                        HttpClient = useCustomClient ? httpClient : null,
+                        LoggingType = HttpLoggingType.None,
+                        ThrowOnNonSuccessStatusCode = true,
+                        Timeout = useConfiguredTimeout ? TimeSpan.FromMilliseconds(100) : null,
+                    },
+                    useConfiguredTimeout ? CancellationToken.None : cancellationTokenSource.Token)
+                    .WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    public async Task SendAsync_CustomClientKeepsTimeoutOutsideLoggedReplayContent()
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var content = new StreamContent(
+            new MemoryStream("response body"u8.ToArray()));
+        content.Headers.ContentType = new MediaTypeHeaderValue("text/plain");
+        using var httpClient = new HttpClient(new ImmediateResponseHandler(content));
+        var moduleLoggerProvider = new Mock<IModuleLoggerProvider>();
+        moduleLoggerProvider
+            .Setup(x => x.GetLogger())
+            .Returns(Mock.Of<IModuleLogger>());
+        var httpLogger = new HttpLogger(
+            Mock.Of<IHttpRequestFormatter>(),
+            new HttpResponseFormatter(
+                Mock.Of<ISecretObfuscator>(),
+                Mock.Of<ISecretProvider>(),
+                Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())));
+        var http = new ModularPipelines.Http.Http(
+            Mock.Of<IHttpClientFactory>(),
+            moduleLoggerProvider.Object,
+            httpLogger,
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+        using var response = await http.SendAsync(new HttpOptions(
+            new HttpRequestMessage(HttpMethod.Get, "https://example.test/logged-timeout"))
+        {
+            HttpClient = httpClient,
+            LoggingType = HttpLoggingType.Response,
+            Timeout = timeout,
+        });
+
+        await Task.Delay(timeout + TimeSpan.FromMilliseconds(50));
+
+        await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+        {
+            var stream = await response.Content.ReadAsStreamAsync();
+            await stream.ReadExactlyAsync(new byte[1]);
+        });
+    }
+
+    [Test]
+    public async Task SendAsync_ConfiguredTimeoutCancelsFactoryResponseLogging()
+    {
+        var timeout = TimeSpan.FromMilliseconds(100);
+        var contentStream = new BlockingReadStream();
+        var content = new StreamContent(contentStream);
+        content.Headers.ContentType = new MediaTypeHeaderValue("text/plain");
+
+        var moduleLoggerProvider = new Mock<IModuleLoggerProvider>();
+        moduleLoggerProvider
+            .Setup(x => x.GetLogger())
+            .Returns(Mock.Of<IModuleLogger>());
+        var httpLogger = new HttpLogger(
+            Mock.Of<IHttpRequestFormatter>(),
+            new HttpResponseFormatter(
+                Mock.Of<ISecretObfuscator>(),
+                Mock.Of<ISecretProvider>(),
+                Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())));
+        var responseLoggingHandler = new ResponseLoggingHttpHandler(
+            moduleLoggerProvider.Object,
+            httpLogger)
+        {
+            InnerHandler = new ImmediateResponseHandler(content),
+        };
+        using var httpClient = new HttpClient(responseLoggingHandler);
+        var httpClientFactory = new Mock<IHttpClientFactory>();
+        httpClientFactory
+            .Setup(x => x.CreateClient(It.IsAny<string>()))
+            .Returns(httpClient);
+        var http = new ModularPipelines.Http.Http(
+            httpClientFactory.Object,
+            moduleLoggerProvider.Object,
+            httpLogger,
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions()));
+
+        try
+        {
+            await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+                await http.SendAsync(new HttpOptions(
+                        new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-log-body"))
+                    {
+                        LoggingType = HttpLoggingType.Response,
+                        Timeout = timeout,
+                    })
+                    .WaitAsync(TimeSpan.FromSeconds(1)));
+        }
+        finally
+        {
+            contentStream.Release();
+        }
+    }
+
+    [Test]
+    public async Task ResilienceHandler_DisposesRetryableResponseBeforeNextAttempt()
+    {
+        var retryContent = new TrackingStringContent("retry");
+        var innerHandler = new SequenceResponseHandler(
+            new HttpResponseMessage(HttpStatusCode.ServiceUnavailable)
+            {
+                Content = retryContent,
+            },
+            new HttpResponseMessage(HttpStatusCode.OK));
+        var moduleLoggerProvider = new Mock<IModuleLoggerProvider>();
+        moduleLoggerProvider
+            .Setup(x => x.GetLogger())
+            .Returns(Mock.Of<IModuleLogger>());
+        using var handler = new ResilienceHttpHandler(
+            moduleLoggerProvider.Object,
+            Microsoft.Extensions.Options.Options.Create(new PipelineOptions
+            {
+                DefaultHttpResilienceOptions = new HttpResilienceOptions
+                {
+                    MaxRetryAttempts = 1,
+                    InitialDelay = TimeSpan.Zero,
+                    JitterFactor = 0,
+                },
+            }))
+        {
+            InnerHandler = innerHandler,
+        };
+        using var invoker = new HttpMessageInvoker(handler);
+        using var request = new HttpRequestMessage(HttpMethod.Get, "https://example.test/retry");
+        using var response = await invoker.SendAsync(request, CancellationToken.None);
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(response.StatusCode).IsEqualTo(HttpStatusCode.OK);
+            await Assert.That(innerHandler.CallCount).IsEqualTo(2);
+            await Assert.That(retryContent.IsDisposed).IsTrue();
+        }
+    }
+
     [Test]
     public async Task PublicApi_DoesNotExposeRawHttpClients()
     {
@@ -155,4 +820,227 @@ await http.SendAsync(new HttpOptions(new HttpRequestMessage(HttpMethod.Get, serv
             await Assert.That(indexOfDuration).IsLessThan(indexOfResponse);
         }
     }
+
+    private sealed class ImmediateResponseHandler(
+        HttpContent content,
+        HttpStatusCode statusCode = HttpStatusCode.OK) : HttpMessageHandler
+    {
+        public TaskCompletionSource RequestReceived { get; } =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        protected override Task<HttpResponseMessage> SendAsync(
+            HttpRequestMessage request,
+            CancellationToken cancellationToken)
+        {
+            RequestReceived.TrySetResult();
+            return Task.FromResult(new HttpResponseMessage(statusCode)
+            {
+                Content = content,
+                RequestMessage = request,
+            });
+        }
+    }
+
+    private sealed class BlockingHttpContent : HttpContent
+    {
+        private readonly TaskCompletionSource _release =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public void Release() => _release.TrySetResult();
+
+        protected override async Task SerializeToStreamAsync(
+            Stream stream,
+            TransportContext? context)
+        {
+            await _release.Task;
+            await stream.WriteAsync("response body"u8.ToArray());
+        }
+
+        protected override bool TryComputeLength(out long length)
+        {
+            length = 0;
+            return false;
+        }
+    }
+
+    private sealed class BlockingReadStream(
+        bool throwIOExceptionWhenDisposed = false,
+        bool ignoreAsyncCancellation = false,
+        bool returnEofWhenDisposed = false) : Stream
+    {
+        private readonly TaskCompletionSource _release =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource _readStarted =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly ManualResetEventSlim _synchronousRelease = new();
+        private bool _isDisposed;
+
+        public override bool CanRead => true;
+
+        public override bool CanSeek => false;
+
+        public override bool CanWrite => false;
+
+        public override long Length => throw new NotSupportedException();
+
+        public override long Position
+        {
+            get => throw new NotSupportedException();
+            set => throw new NotSupportedException();
+        }
+
+        public Task ReadStarted => _readStarted.Task;
+
+        public void Release()
+        {
+            _release.TrySetResult();
+            _synchronousRelease.Set();
+        }
+
+        public override void Flush()
+        {
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            return WaitForSynchronousRead();
+        }
+
+        public override int Read(Span<byte> buffer)
+        {
+            return WaitForSynchronousRead();
+        }
+
+        public override async ValueTask<int> ReadAsync(
+            Memory<byte> buffer,
+            CancellationToken cancellationToken = default)
+        {
+            _readStarted.TrySetResult();
+            if (ignoreAsyncCancellation)
+            {
+                await _release.Task;
+                if (!returnEofWhenDisposed)
+                {
+                    ThrowIfDisposed();
+                }
+            }
+            else
+            {
+                await _release.Task.WaitAsync(cancellationToken);
+            }
+
+            return 0;
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                _isDisposed = true;
+                Release();
+            }
+
+            base.Dispose(disposing);
+        }
+
+        private int WaitForSynchronousRead()
+        {
+            _synchronousRelease.Wait();
+            ThrowIfDisposed();
+            return 0;
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            throw new NotSupportedException();
+        }
+
+        private void ThrowIfDisposed()
+        {
+            if (_isDisposed)
+            {
+                throw throwIOExceptionWhenDisposed
+                    ? new IOException("The stream was interrupted.")
+                    : new ObjectDisposedException(nameof(BlockingReadStream));
+            }
+        }
+    }
+
+    private sealed class LegacyBodyLogger(bool logRequest) : IHttpLogger
+    {
+        public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger)
+        {
+            return logRequest
+                ? request.Content!.ReadAsStringAsync()
+                : Task.CompletedTask;
+        }
+
+        public Task PrintRequest(
+            HttpRequestMessage request,
+            IModuleLogger logger,
+            HttpLoggingOptions options)
+        {
+            return logRequest
+                ? request.Content!.ReadAsStringAsync()
+                : Task.CompletedTask;
+        }
+
+        public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger)
+        {
+            return logRequest
+                ? Task.CompletedTask
+                : response.Content.ReadAsStringAsync();
+        }
+
+        public Task PrintResponse(
+            HttpResponseMessage response,
+            IModuleLogger logger,
+            HttpLoggingOptions options)
+        {
+            return logRequest
+                ? Task.CompletedTask
+                : response.Content.ReadAsStringAsync();
+        }
+
+        public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger)
+        {
+        }
+
+        public void PrintDuration(TimeSpan duration, IModuleLogger logger)
+        {
+        }
+    }
+
+    private sealed class SequenceResponseHandler(params HttpResponseMessage[] responses) : HttpMessageHandler
+    {
+        public int CallCount { get; private set; }
+
+        protected override Task<HttpResponseMessage> SendAsync(
+            HttpRequestMessage request,
+            CancellationToken cancellationToken)
+        {
+            return Task.FromResult(responses[CallCount++]);
+        }
+    }
+
+    private sealed class TrackingStringContent(string content) : StringContent(content)
+    {
+        public bool IsDisposed { get; private set; }
+
+        protected override void Dispose(bool disposing)
+        {
+            IsDisposed = true;
+            base.Dispose(disposing);
+        }
+    }
 }
diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs
index 8cf8da07acc..a2ef3832e44 100644
--- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs
+++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs
@@ -1,13 +1,26 @@
+using System.Net;
 using ModularPipelines.Context;
 using ModularPipelines.Context.Domains.Files;
 using ModularPipelines.Context.Domains.Network;
+using ModularPipelines.Exceptions;
+using ModularPipelines.FileSystem;
+using ModularPipelines.Logging;
 using ModularPipelines.Options;
 using ModularPipelines.TestHelpers;
+using Moq;
 
 namespace ModularPipelines.UnitTests.Helpers;
 
 public class DownloaderTests : TestBase
 {
+    [Test]
+    public async Task DownloadOptions_DisableResponseBodyLoggingByDefault()
+    {
+        var options = new DownloadOptions(new Uri("https://example.test/file"));
+
+        await Assert.That(options.LoggingType.HasFlag(ModularPipelines.Http.HttpLoggingType.Response)).IsFalse();
+    }
+
     [Test]
     public async Task Can_Download()
     {
@@ -19,4 +32,219 @@ public async Task Can_Download()
 
         await Assert.That(checksum.Md5(file)).IsEqualTo("AEDF5D7C23744269F358814E602AFE89");
     }
+
+    [Test]
+    public async Task DownloadStringAsync_DisposesResponse()
+    {
+        var content = new TrackingStringContent("download");
+        var downloader = CreateDownloader(content);
+
+        var result = await downloader.DownloadStringAsync(
+            new DownloadOptions(new Uri("https://example.test/download")));
+
+        using (Assert.Multiple())
+        {
+            await Assert.That(result).IsEqualTo("download");
+            await Assert.That(content.IsDisposed).IsTrue();
+        }
+    }
+
+    [Test]
+    public async Task DownloadFileAsync_DisposesResponse()
+    {
+        var content = new TrackingStringContent("download");
+        var fileSystemProvider = new Mock<IFileSystemProvider>();
+        fileSystemProvider
+            .Setup(x => x.Create(It.IsAny<string>()))
+            .Returns(() => new MemoryStream());
+        fileSystemProvider
+            .Setup(x => x.GetRandomFileName())
+            .Returns("random.tmp");
+        var downloader = CreateDownloader(content, fileSystemProvider.Object);
+
+        await downloader.DownloadFileAsync(new DownloadFileOptions(
+            new Uri("https://example.test/download"))
+        {
+            SavePath = "download.bin",
+        });
+
+        await Assert.That(content.IsDisposed).IsTrue();
+    }
+
+    [Test]
+    public async Task DownloadFileAsync_UsesBoundedSiblingTemporaryName()
+    {
+        var destination = Path.Combine(
+            "downloads",
+            new string('x', 240) + ".bin");
+        var expectedTemporaryPath = Path.Combine("downloads", "random.tmp");
+        var fileSystemProvider = new Mock<IFileSystemProvider>();
+        fileSystemProvider
+            .Setup(x => x.GetRandomFileName())
+            .Returns("random.tmp");
+        fileSystemProvider
+            .Setup(x => x.Create(expectedTemporaryPath))
+            .Returns(() => new MemoryStream());
+        var downloader = CreateDownloader(
+            new StringContent("download"),
+            fileSystemProvider.Object);
+
+        await downloader.DownloadFileAsync(new DownloadFileOptions(
+            new Uri("https://example.test/download"))
+        {
+            SavePath = destination,
+        });
+
+        fileSystemProvider.Verify(x => x.Create(expectedTemporaryPath), Times.Once());
+        fileSystemProvider.Verify(
+            x => x.MoveFile(expectedTemporaryPath, destination, true),
+            Times.Once());
+    }
+
+    [Test]
+    public async Task LegacyProviderFallbackDoesNotDeleteExistingDestination()
+    {
+        const string destination = "download.bin";
+        var fileSystemProvider = new Mock<IFileSystemProvider>
+        {
+            CallBase = true,
+        };
+        fileSystemProvider
+            .Setup(x => x.FileExists(destination))
+            .Returns(true);
+
+        var exception = Assert.Throws<NotSupportedException>(() =>
+            fileSystemProvider.Object.MoveFile("temporary.download", destination, overwrite: true));
+
+        await Assert.That(exception.Message).Contains("does not support atomic overwrite moves");
+        fileSystemProvider.Verify(x => x.DeleteFile(destination), Times.Never());
+        fileSystemProvider.Verify(
+            x => x.MoveFile(It.IsAny<string>(), It.IsAny<string>()),
+            Times.Never());
+    }
+
+    [Test]
+    public async Task DownloadFileAsync_PreservesExistingFileWhenStreamingFails()
+    {
+        var directory = Path.Combine(Path.GetTempPath(), $"modular-pipelines-download-{Guid.NewGuid():N}");
+        var destination = Path.Combine(directory, "download.bin");
+        Directory.CreateDirectory(directory);
+        await System.IO.File.WriteAllTextAsync(destination, "original");
+
+        try
+        {
+            var content = new StreamContent(new FailingReadStream("partial download"u8.ToArray()));
+            var downloader = CreateDownloader(content);
+
+            await Assert.ThrowsAsync<IOException>(() => downloader.DownloadFileAsync(new DownloadFileOptions(
+                new Uri("https://example.test/download"))
+            {
+                SavePath = destination,
+            }));
+
+            using (Assert.Multiple())
+            {
+                await Assert.That(await System.IO.File.ReadAllTextAsync(destination)).IsEqualTo("original");
+                await Assert.That(Directory.GetFiles(directory).Length).IsEqualTo(1);
+            }
+        }
+        finally
+        {
+            Directory.Delete(directory, recursive: true);
+        }
+    }
+
+    [Test]
+    public async Task DownloadFileAsync_ReplacesExistingFileAfterStreamingSucceeds()
+    {
+        var directory = Path.Combine(Path.GetTempPath(), $"modular-pipelines-download-{Guid.NewGuid():N}");
+        var destination = Path.Combine(directory, "download.bin");
+        Directory.CreateDirectory(directory);
+        await System.IO.File.WriteAllTextAsync(destination, "original");
+
+        try
+        {
+            var downloader = CreateDownloader(new StringContent("replacement"));
+
+            await downloader.DownloadFileAsync(new DownloadFileOptions(
+                new Uri("https://example.test/download"))
+            {
+                SavePath = destination,
+            });
+
+            using (Assert.Multiple())
+            {
+                await Assert.That(await System.IO.File.ReadAllTextAsync(destination)).IsEqualTo("replacement");
+                await Assert.That(Directory.GetFiles(directory).Length).IsEqualTo(1);
+            }
+        }
+        finally
+        {
+            Directory.Delete(directory, recursive: true);
+        }
+    }
+
+    [Test]
+    public async Task DownloadResponseAsync_DisposesResponseWhenStatusValidationFails()
+    {
+        var content = new TrackingStringContent("failure");
+        var downloader = CreateDownloader(content, statusCode: HttpStatusCode.BadRequest);
+
+        await Assert.ThrowsAsync<HttpResponseException>(() =>
+            downloader.DownloadResponseAsync(
+                new DownloadOptions(new Uri("https://example.test/download"))));
+
+        await Assert.That(content.IsDisposed).IsTrue();
+    }
+
+    private static Downloader CreateDownloader(
+        HttpContent content,
+        IFileSystemProvider? fileSystemProvider = null,
+        HttpStatusCode statusCode = HttpStatusCode.OK)
+    {
+        var http = new Mock<IHttpContext>();
+        http.Setup(x => x.SendAsync(
+                It.IsAny<HttpOptions>(),
+                It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new HttpResponseMessage(statusCode)
+            {
+                Content = content,
+            });
+        var moduleLoggerProvider = new Mock<IModuleLoggerProvider>();
+        moduleLoggerProvider
+            .Setup(x => x.GetLogger())
+            .Returns(Mock.Of<IModuleLogger>());
+
+        return new Downloader(
+            moduleLoggerProvider.Object,
+            http.Object,
+            fileSystemProvider ?? SystemFileSystemProvider.Instance);
+    }
+
+    private sealed class TrackingStringContent(string content) : StringContent(content)
+    {
+        public bool IsDisposed { get; private set; }
+
+        protected override void Dispose(bool disposing)
+        {
+            IsDisposed = true;
+            base.Dispose(disposing);
+        }
+    }
+
+    private sealed class FailingReadStream(byte[] content) : MemoryStream(content)
+    {
+        private bool _hasRead;
+
+        public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            if (_hasRead)
+            {
+                throw new IOException("Simulated streaming failure");
+            }
+
+            _hasRead = true;
+            return base.ReadAsync(buffer, cancellationToken);
+        }
+    }
 }
