diff --git a/docs/code_snippets/03_03_file_uploads.dart b/docs/code_snippets/03_03_file_uploads.dart
index 9e443045..977a2d11 100644
--- a/docs/code_snippets/03_03_file_uploads.dart
+++ b/docs/code_snippets/03_03_file_uploads.dart
@@ -20,17 +20,23 @@ Future<void> howToUploadAFileOrImageStep1() async {
     custom: {'width': 600, 'height': 400},
   );
 
-  // Upload the attachment
-  final result = await attachmentUploader.upload(
-    streamAttachment,
-    // Optionally track upload progress
-    onProgress: (progress) {
-      // Handle progress updates
-    },
-  );
-
-  // Map the result to an Attachment model to send with an activity
-  final uploadedAttachment = result.getOrThrow();
+  // Start the upload. The task comes back straight away, already running, and
+  // `task.cancel()` calls it off.
+  final task = attachmentUploader.upload(streamAttachment);
+
+  // Optionally track upload progress. `fraction` is null while the file's
+  // length is unknown, which is when an indeterminate bar is the right thing
+  // to show.
+  task.state.listen((state) {
+    if (state case UploadInProgress(progress: UploadProgress(:final fraction?))) {
+      print('${(fraction * 100).round()}% sent');
+    }
+  });
+
+  // Map the result to an Attachment model to send with an activity. A failed
+  // upload carries its error, so `getOrThrow` opts into throwing it; `fold`
+  // handles it instead.
+  final uploadedAttachment = (await task.result).getOrThrow();
   final attachmentReadyToBeSent = Attachment(
     imageUrl: uploadedAttachment.remoteUrl,
     assetUrl: uploadedAttachment.remoteUrl,
diff --git a/melos.yaml b/melos.yaml
index abe240de..040035c0 100644
--- a/melos.yaml
+++ b/melos.yaml
@@ -25,6 +25,7 @@ command:
       auto_route: ^11.0.0
       collection: ^1.18.0
       chewie: ^1.11.3
+      clock: ^1.1.2
       dio: ^5.9.0
       equatable: ^2.0.5
       flutter_state_notifier: ^1.0.0
@@ -53,7 +54,7 @@ command:
       stream_core: 
         git:
           url: https://github.com/GetStream/stream-core-flutter.git
-          ref: f83b5d4d706a79fc429de2d27aead4394b83c1fb
+          ref: 505432f213af8911d54ab3c855d6ad6fbe492fc9
           path: packages/stream_core
       video_player: ^2.10.0
       uuid: ^4.5.1
@@ -63,6 +64,7 @@ command:
     dev_dependencies:
       auto_route_generator: ^10.0.0
       build_runner: ^2.4.15
+      fake_async: ^1.3.3
       flutter_launcher_icons: ^0.14.4
       freezed: ^3.0.0
       injectable_generator: ^3.0.0
diff --git a/packages/stream_feeds/CHANGELOG.md b/packages/stream_feeds/CHANGELOG.md
index e8273f1a..e96ab482 100644
--- a/packages/stream_feeds/CHANGELOG.md
+++ b/packages/stream_feeds/CHANGELOG.md
@@ -7,6 +7,10 @@
 - `PollResponseData.votingVisibility` is now required, so anything constructing one directly must supply it
 - `ActivityCommentList.state` returns `ActivityCommentListState` rather than `StateNotifier<ActivityCommentListState>`, matching the other state classes
 - Removed the call, recording, streaming and chat types that were never part of the Feeds API
+- Every failure the SDK reports for work it attempted now arrives as a `StreamException` subclass — `StreamApiException`, `StreamNetworkException`, `StreamAuthenticationException` or `StreamClientException` — replacing `ClientException` and `HttpClientException`, which are removed. `StreamApiError` remains, as the server's error payload and the type of `ConnectionErrorEvent.error`, but is no longer what the SDK throws or returns. `StreamFeedsException` aliases the base type, so one `on` clause catches all four
+- `connect` throws a `StateError` when a connection is already established or in progress, and a `StreamFeedsException` carrying the cause when it fails
+- `StreamAttachmentUploader.upload`, reached through `StreamFeedsClient.attachmentUploader`, returns an `AttachmentUploadTask` rather than a `Future<Result<UploadedAttachment>>`, and takes no `onProgress`: progress arrives on the task's `state`. `uploadBatch` returns an `AttachmentUploadBatch` rather than a `Stream<Result<UploadedAttachment>>`
+- `Feed.addActivity`, `Feed.addComment` and `Activity.addCommentsBatch` throw an `ArgumentError` when two attachments in one request share an id, rather than reporting it through the returned `Result`
 
 ### ✨ Features
 
@@ -29,15 +33,16 @@
 
 ### 🐛 Bug Fixes
 
+- Fixed a batch never running again after its first: an add that arrived once a batch had run joined that settled one instead of starting its own, so feed capabilities were fetched once per client and every feed discovered afterwards was answered with the first batch's result
 - Fixed `connect` failing when called straight after `disconnect`
 - Fixed a connection that could not authenticate hanging until it timed out, rather than failing with the reason
 - Fixed the `X-Stream-Client` header: the SDK identifier was sent twice, the version was hardcoded, and the OS was left out
 
 ### 🔄 Changed
 
+- Attachment uploads for a batch of requests now share one concurrency limit instead of one each, so `Activity.addCommentsBatch` no longer starts several uploads per comment at once; a failure also calls off the uploads still in flight rather than letting them finish work that is about to be discarded
 - `disconnect` now only closes the connection, leaving the client reusable with its existing subscriptions intact; releasing it is `dispose`
 - An expired token now recovers on its own: the connection comes back with one the `TokenProvider` issued afterwards, without the app doing anything
-- `connect` throws a `ClientException` when a connection is already established or in progress, and the one it throws on failure carries the underlying cause
 - Renamed the types below. The old names still compile, with a deprecation warning, and `dart fix --apply` migrates them:
 
 | Old name | New name |
diff --git a/packages/stream_feeds/lib/src/cdn/cdn_api.dart b/packages/stream_feeds/lib/src/cdn/cdn_api.dart
index 3953b005..3eabf014 100644
--- a/packages/stream_feeds/lib/src/cdn/cdn_api.dart
+++ b/packages/stream_feeds/lib/src/cdn/cdn_api.dart
@@ -45,5 +45,5 @@ abstract interface class CdnApi {
 
 class _ResultCallAdapter<T> extends CallAdapter<Future<T>, Future<Result<T>>> {
   @override
-  Future<Result<T>> adapt(Future<T> Function() call) => runSafely(call);
+  Future<Result<T>> adapt(Future<T> Function() call) => runApiSafely(call);
 }
diff --git a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart
index cb01daf8..4d8f213e 100644
--- a/packages/stream_feeds/lib/src/client/feeds_client_impl.dart
+++ b/packages/stream_feeds/lib/src/client/feeds_client_impl.dart
@@ -255,15 +255,18 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
 
   Future<void> _authenticateUser(
     WsRequestSender send,
-    StreamApiError? previousError,
+    StreamApiException? previousError,
   ) async {
-    if (previousError?.isTokenExpiredError ?? false) {
+    if (previousError?.isTokenExpired ?? false) {
       _tokenManager.expireToken();
 
       // A guest cannot refresh: another exchange answers with a different guest. The session ends
       // here, and the app starts another by building a new client.
       if (_tokenManager.usesStaticProvider) {
-        throw ClientException(message: 'The token was refused and the provider has no other to give');
+        throw StreamAuthenticationException(
+          message: 'The token was refused and the provider has no other to give',
+          cause: previousError,
+        );
       }
     }
 
@@ -297,11 +300,11 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
     }
 
     if (connectionState.value case Connecting() || Authenticating()) {
-      throw ClientException(message: 'Connection already in progress for ${user.id}');
+      throw StateError('Connection already in progress for ${user.id}');
     }
 
     if (connectionState.value case Connected()) {
-      throw ClientException(message: 'Connection already available for ${user.id}');
+      throw StateError('Connection already available for ${user.id}');
     }
 
     _logger.d(() => 'connect ${user.id} (${user.type.name}), webSocket: $connectWebSocket');
@@ -332,14 +335,16 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
   Future<void> _exchangeForGuestIdentity() async {
     final result = await _guestRepository.createGuest(user);
 
-    // Reported like every other connect failure, with the cause attached.
-    final response = result.getOrElse(
-      (error, stackTrace) => throw ClientException(
+    // Reported like every other connect failure, already classified by the API call seam.
+    final response = result.getOrElse((error, stackTrace) {
+      var exception = StreamException.tryFrom(error);
+      exception ??= StreamClientException(
         message: 'Failed to create a guest user',
-        error: error,
-        stackTrace: stackTrace,
-      ),
-    );
+        cause: error,
+      );
+
+      Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current);
+    });
 
     final tokenProvider = TokenProvider.static(response.token);
     _logger.d(() => 'guest created, server assigned ${response.user.id}');
@@ -370,7 +375,16 @@ class StreamFeedsClientImpl with Disposable implements StreamFeedsClient {
 
     if (state case Disconnected(:final source)) {
       _logger.w(() => 'connect ${user.id} failed: ${source.closeReason}', error: source.cause);
-      throw ClientException(message: source.closeReason, error: source.cause);
+
+      var exception = StreamException.tryFrom(source.cause);
+      exception ??= StreamNetworkException(message: source.closeReason, cause: source.cause);
+
+      final stackTrace = switch (source) {
+        ServerInitiated(:final stackTrace) || AuthenticationFailed(:final stackTrace) => stackTrace,
+        UserInitiated() || SystemInitiated() || UnHealthyConnection() || ConnectTimeout() => null,
+      };
+
+      Error.throwWithStackTrace(exception, stackTrace ?? StackTrace.current);
     }
 
     _logger.d(() => 'connected ${user.id}');
diff --git a/packages/stream_feeds/lib/src/feeds_client.dart b/packages/stream_feeds/lib/src/feeds_client.dart
index 947570de..df06417c 100644
--- a/packages/stream_feeds/lib/src/feeds_client.dart
+++ b/packages/stream_feeds/lib/src/feeds_client.dart
@@ -49,6 +49,12 @@ import 'state/user_list.dart';
 
 export 'client/moderation_client.dart';
 
+/// The root of the failures this SDK reports.
+///
+/// An alias of [StreamException], so code written against either name catches
+/// the same failures.
+typedef StreamFeedsException = StreamException;
+
 /// {@template stream_feeds_client}
 /// Stream Feeds client for building scalable newsfeeds and activity streams.
 ///
@@ -258,8 +264,9 @@ abstract interface class StreamFeedsClient {
 
   /// Establishes a connection to the Stream service.
   ///
-  /// Call this before anything else on the client. Throws a [ClientException] if the connection
-  /// fails, or if one is already established or in progress, and a [StateError] after [dispose].
+  /// Call this before anything else on the client. Throws a [StreamFeedsException] if the
+  /// connection fails, and a [StateError] when one is already established or in progress, or
+  /// after [dispose].
   ///
   /// Pass [connectWebSocket] as `false` if the client only needs to make requests. In that case:
   ///
diff --git a/packages/stream_feeds/lib/src/generated/api/api/default_api.dart b/packages/stream_feeds/lib/src/generated/api/api/default_api.dart
index eb735350..f4345545 100644
--- a/packages/stream_feeds/lib/src/generated/api/api/default_api.dart
+++ b/packages/stream_feeds/lib/src/generated/api/api/default_api.dart
@@ -801,5 +801,5 @@ abstract interface class DefaultApi {
 
 class _ResultCallAdapter<T> extends CallAdapter<Future<T>, Future<Result<T>>> {
   @override
-  Future<Result<T>> adapt(Future<T> Function() call) => runSafely(call);
+  Future<Result<T>> adapt(Future<T> Function() call) => runApiSafely(call);
 }
diff --git a/packages/stream_feeds/lib/src/repository/capabilities_repository.dart b/packages/stream_feeds/lib/src/repository/capabilities_repository.dart
index d60c3919..29177697 100644
--- a/packages/stream_feeds/lib/src/repository/capabilities_repository.dart
+++ b/packages/stream_feeds/lib/src/repository/capabilities_repository.dart
@@ -77,25 +77,15 @@ class CapabilitiesRepository {
 }
 
 extension on Result<Map<String, List<FeedOwnCapability>>> {
-  bool shouldRetry() {
-    switch (this) {
-      case api.Success():
-        return false;
-
-      case final api.Failure failure:
-        final error = failure.error;
-        if (error is! StreamDioException) {
-          return false;
-        }
-        final exception = error.exception;
-        if (exception is! HttpClientException) {
-          return false;
-        }
-        final statusCode = exception.statusCode;
-        if (statusCode == null) {
-          return false;
-        }
-        return statusCode < 100 || statusCode >= 500;
-    }
-  }
+  bool shouldRetry() => switch (this) {
+    api.Success() => false,
+    api.Failure(:final error) => switch (error) {
+      StreamNetworkException(isCancelled: true) => false,
+      StreamNetworkException() => true,
+      // A rate limit is not retried here: this waits a fixed moment, which is
+      // not the wait a rate limit asks for.
+      StreamApiException(:final statusCode) => statusCode < 100 || statusCode >= 500,
+      _ => false,
+    },
+  };
 }
diff --git a/packages/stream_feeds/lib/src/state/activity.dart b/packages/stream_feeds/lib/src/state/activity.dart
index 295f1e2b..784fbf6a 100644
--- a/packages/stream_feeds/lib/src/state/activity.dart
+++ b/packages/stream_feeds/lib/src/state/activity.dart
@@ -186,6 +186,16 @@ class Activity with Disposable {
 
   /// Adds a comment to this activity.
   ///
+  /// Attachments in [ActivityAddCommentRequest.attachmentUploads] are uploaded
+  /// first, and the comment is added once they are all in. To follow those
+  /// uploads or call them off, upload through
+  /// `StreamFeedsClient.attachmentUploader` instead and pass the results as
+  /// [ActivityAddCommentRequest.attachments].
+  ///
+  /// Throws an [ArgumentError] if two of those attachments share an id. Ids
+  /// default to a fresh UUID, so this only happens when one is given
+  /// explicitly, or the same attachment is listed twice.
+  ///
   /// Returns a [Result] containing the created [CommentData] or an error.
   Future<Result<CommentData>> addComment({
     required ActivityAddCommentRequest request,
@@ -203,6 +213,16 @@ class Activity with Disposable {
 
   /// Adds multiple comments to this activity in a batch.
   ///
+  /// Every request's attachments are uploaded as one batch, so a failure in one
+  /// can call off the uploads of the others. To follow those uploads or call
+  /// them off, upload through `StreamFeedsClient.attachmentUploader` instead
+  /// and pass the results as [ActivityAddCommentRequest.attachments].
+  ///
+  /// Throws an [ArgumentError] if two attachments share an id, across the whole
+  /// batch rather than within one request — one batch cannot tell them apart.
+  /// Ids default to a fresh UUID, so this only happens when one is given
+  /// explicitly, or the same attachment is listed twice.
+  ///
   /// Returns a [Result] containing a list of created [CommentData] or an error.
   Future<Result<List<CommentData>>> addCommentsBatch(
     List<ActivityAddCommentRequest> requests,
diff --git a/packages/stream_feeds/lib/src/state/feed.dart b/packages/stream_feeds/lib/src/state/feed.dart
index 1b5ac20a..7ba8f109 100644
--- a/packages/stream_feeds/lib/src/state/feed.dart
+++ b/packages/stream_feeds/lib/src/state/feed.dart
@@ -186,6 +186,16 @@ class Feed with Disposable {
   ///
   /// The [request] contains the activity data to add.
   ///
+  /// Attachments in [FeedAddActivityRequest.attachmentUploads] are uploaded
+  /// first, and the activity is added once they are all in. To follow those
+  /// uploads or call them off, upload through
+  /// `StreamFeedsClient.attachmentUploader` instead and pass the results as
+  /// [FeedAddActivityRequest.attachments].
+  ///
+  /// Throws an [ArgumentError] if two of those attachments share an id. Ids
+  /// default to a fresh UUID, so this only happens when one is given
+  /// explicitly, or the same attachment is listed twice.
+  ///
   /// Returns a [Result] containing the added [ActivityData] if successful, or an error if the
   /// operation fails.
   Future<Result<ActivityData>> addActivity({
diff --git a/packages/stream_feeds/lib/src/state/feed_state.dart b/packages/stream_feeds/lib/src/state/feed_state.dart
index ed0c1bbd..eecee6ae 100644
--- a/packages/stream_feeds/lib/src/state/feed_state.dart
+++ b/packages/stream_feeds/lib/src/state/feed_state.dart
@@ -1,5 +1,6 @@
 import 'dart:math';
 
+import 'package:clock/clock.dart';
 import 'package:collection/collection.dart';
 import 'package:freezed_annotation/freezed_annotation.dart';
 import 'package:state_notifier/state_notifier.dart';
@@ -671,7 +672,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unread: 0,
       readActivities: readActivities,
-      lastReadAt: DateTime.timestamp(),
+      lastReadAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
@@ -691,7 +692,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unseen: 0,
       seenActivities: seenActivities,
-      lastSeenAt: DateTime.timestamp(),
+      lastSeenAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
@@ -715,7 +716,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unread: updatedUnreadCount,
       readActivities: updatedReadActivities,
-      lastReadAt: DateTime.timestamp(),
+      lastReadAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
@@ -739,7 +740,7 @@ extension on FeedState {
     final updatedNotificationStatus = notificationStatus?.copyWith(
       unseen: updatedUnseenCount,
       seenActivities: updatedSeenActivities,
-      lastSeenAt: DateTime.timestamp(),
+      lastSeenAt: clock.now().toUtc(),
     );
 
     return copyWith(notificationStatus: updatedNotificationStatus);
diff --git a/packages/stream_feeds/lib/src/utils/batcher.dart b/packages/stream_feeds/lib/src/utils/batcher.dart
index b8c855aa..9c6360c0 100644
--- a/packages/stream_feeds/lib/src/utils/batcher.dart
+++ b/packages/stream_feeds/lib/src/utils/batcher.dart
@@ -1,5 +1,7 @@
 import 'dart:async';
 
+import 'package:clock/clock.dart';
+
 class Batcher<T, R> {
   Batcher({
     required this.action,
@@ -23,8 +25,10 @@ class Batcher<T, R> {
     }
 
     _itemsToProcess.add(item);
-    _nextActionCompleter ??= _planBatchFetch();
-    return _nextActionCompleter!.future;
+
+    // `_planBatchFetch` owns this field, and clears it when it runs on the spot.
+    final completer = _nextActionCompleter ?? _planBatchFetch();
+    return completer.future;
   }
 
   void dispose() {
@@ -32,11 +36,11 @@ class Batcher<T, R> {
   }
 
   Completer<R> _planBatchFetch() {
-    final timeSinceLastRun = DateTime.now().difference(_lastRun);
+    final timeSinceLastRun = clock.now().difference(_lastRun);
 
     final newActionCompleter = Completer<R>();
     _nextActionCompleter = newActionCompleter;
-    _lastRun = DateTime.now();
+    _lastRun = clock.now();
 
     if (timeSinceLastRun >= interval) {
       _runBatch();
diff --git a/packages/stream_feeds/lib/src/utils/uploader.dart b/packages/stream_feeds/lib/src/utils/uploader.dart
index 93408cc1..014eea92 100644
--- a/packages/stream_feeds/lib/src/utils/uploader.dart
+++ b/packages/stream_feeds/lib/src/utils/uploader.dart
@@ -26,118 +26,108 @@ extension HasAttachmentsExtension on StreamAttachmentUploader {
   /// Returns a [Result] containing the updated request or an error.
   Future<Result<T>> processRequest<T extends HasAttachments<T>>(
     T request, {
-    OnBatchUploadProgress? onProgress,
     int maxConcurrent = 5,
     bool eagerError = true,
   }) async {
-    final attachmentsToUpload = request.attachmentUploads;
-    // If there are no attachments to upload, return the original request.
-    if (attachmentsToUpload == null || attachmentsToUpload.isEmpty) {
-      return Result.success(request);
-    }
-
-    final uploadResult = await _uploadAll(
-      attachmentsToUpload,
-      onProgress: onProgress,
+    final processed = await processRequestsBatch(
+      [request],
       maxConcurrent: maxConcurrent,
       eagerError: eagerError,
     );
 
-    return uploadResult.map((attachments) {
-      final uploadedAttachments = <String, api.Attachment>{
-        for (final uploaded in attachments)
-          uploaded.id: api.Attachment(
-            type: uploaded.type,
-            custom: {...?uploaded.custom},
-            assetUrl: uploaded.remoteUrl,
-            imageUrl: uploaded.remoteUrl,
-            thumbUrl: uploaded.thumbnailUrl,
-          ),
-      };
-
-      // Merge uploaded attachments with existing ones, avoiding duplicates
-      final current = request.attachments ?? [];
-      final updatedAttachments = current.merge(
-        uploadedAttachments.values,
-        key: (it) => (it.type, it.assetUrl, it.imageUrl),
-      );
-
-      // Remove processed uploads from the upload queue using ID-based filtering
-      final uploadedIds = uploadedAttachments.keys.toSet();
-      final updatedAttachmentUploads = attachmentsToUpload.where(
-        (upload) => !uploadedIds.contains(upload.id),
-      );
-
-      return request.withAttachments(
-        attachments: updatedAttachments.takeIf((it) => it.isNotEmpty),
-        attachmentUploads: updatedAttachmentUploads.toList(),
-      );
-    });
+    return processed.map((requests) => requests.single);
   }
 
   /// Processes multiple requests with attachment uploads in parallel.
   ///
-  /// Processes each request individually using [processRequest] and returns
-  /// a list of updated requests with all attachments ready for API submission.
+  /// Uploads every request's attachments as one batch, so [maxConcurrent]
+  /// bounds the uploads across all of them rather than within each one, and
+  /// merges each request's own attachments back into it.
+  ///
+  /// When [eagerError] is true the first upload to fail becomes the result's
+  /// and the rest are called off; when false every attachment is attempted and
+  /// the ones that did not make it stay queued on their own request.
   ///
-  /// Returns a [Result] containing the list of updated requests or an error.
+  /// Throws an [ArgumentError] if two requests share an attachment id, which
+  /// one batch could not tell apart.
   Future<Result<List<T>>> processRequestsBatch<T extends HasAttachments<T>>(
     List<T> requests, {
-    OnBatchUploadProgress? onProgress,
     int maxConcurrent = 5,
     bool eagerError = true,
-  }) {
-    return runSafely(() async {
-      final batch = requests.map(
-        (request) => processRequest(
-          request,
-          onProgress: onProgress,
-          maxConcurrent: maxConcurrent,
-          eagerError: eagerError,
-        ),
-      );
-
-      final processed = await batch.wait;
-
-      final successfulRequests = <T?>[];
-      for (final result in processed) {
-        // If eagerError is enabled, throw on first failure
-        if (result.exceptionOrNull() case final error? when eagerError) {
-          final stackTrace = result.stackTraceOrNull();
-          Error.throwWithStackTrace(error, stackTrace ?? StackTrace.current);
-        }
-
-        successfulRequests.add(result.getOrNull());
-      }
-
-      return successfulRequests.nonNulls.toList();
-    });
-  }
+  }) async {
+    final attachmentsToUpload = [
+      for (final request in requests) ...?request.attachmentUploads,
+    ];
 
-  // Uploads multiple attachments in parallel with progress tracking.
-  //
-  // Processes [attachments] in batches with configurable concurrency and progress
-  // reporting. Returns a [Result] containing the list of uploaded attachments.
-  //
-  // Returns a [Result] containing a list of [UploadedAttachment] or an error.
-  Future<Result<List<UploadedAttachment>>> _uploadAll(
-    Iterable<StreamAttachment> attachments, {
-    OnBatchUploadProgress? onProgress,
-    int maxConcurrent = 5,
-    bool eagerError = true,
-  }) {
-    return runSafely(() async {
-      final batch = uploadBatch(
-        attachments,
-        onProgress: onProgress,
-        maxConcurrent: maxConcurrent,
-        eagerError: eagerError,
-      );
-
-      final batchResult = await batch.toList();
-      final uploadedAttachments = batchResult.map((it) => it.getOrNull());
-
-      return uploadedAttachments.nonNulls.toList();
-    });
+    if (attachmentsToUpload.isEmpty) return Result.success(requests);
+
+    final batch = uploadBatch(
+      attachmentsToUpload,
+      maxConcurrent: maxConcurrent,
+      eagerError: eagerError,
+    );
+
+    return switch (await batch.result) {
+      BatchUploadCompleted(:final items) => Result.success(_distribute(requests, items)),
+      BatchUploadStoppedOnError(:final error, :final stackTrace) => Result.failure(error, stackTrace),
+      BatchUploadCancelled() => const Result.failure(
+        StreamNetworkException(message: 'The attachment uploads were cancelled', isCancelled: true),
+      ),
+    };
   }
 }
+
+// Each request with its share of the batch folded in.
+//
+// The map is built once and looked up per request, so this stays linear in the
+// number of attachments however many requests share the batch.
+List<T> _distribute<T extends HasAttachments<T>>(
+  List<T> requests,
+  List<BatchUploadItemResult> items,
+) {
+  final succeeded = items.map((it) => it.result.getOrNull()).nonNulls;
+  final uploaded = {for (final it in succeeded) it.id: _toApiAttachment(it)};
+
+  return [for (final request in requests) _withUploaded(request, uploaded)];
+}
+
+// An uploaded attachment as the api model a request carries.
+//
+// Both urls get the same one whatever the attachment is, and `type` is what
+// tells them apart. The Swift SDK maps it the same way; Android instead fills
+// `imageUrl` for images only.
+api.Attachment _toApiAttachment(
+  UploadedAttachment attachment,
+) => api.Attachment(
+  type: attachment.type,
+  custom: {...?attachment.custom},
+  assetUrl: attachment.remoteUrl,
+  imageUrl: attachment.remoteUrl,
+  thumbUrl: attachment.thumbnailUrl,
+);
+
+// One request's share of [uploaded] merged in beside the attachments it already
+// had and taken off its upload queue. Whatever did not make it stays queued for
+// a later attempt.
+T _withUploaded<T extends HasAttachments<T>>(
+  T request,
+  Map<String, api.Attachment> uploaded,
+) {
+  final queued = request.attachmentUploads;
+  if (queued == null || queued.isEmpty) return request;
+
+  final merged = queued.map((it) => uploaded[it.id]).nonNulls;
+  final stillQueued = queued.where((it) => !uploaded.containsKey(it.id));
+
+  // Merge uploaded attachments with existing ones, avoiding duplicates
+  final current = request.attachments ?? [];
+  final updated = current.merge(
+    merged,
+    key: (it) => (it.type, it.assetUrl, it.imageUrl),
+  );
+
+  return request.withAttachments(
+    attachments: updated.takeIf((it) => it.isNotEmpty),
+    attachmentUploads: stillQueued.toList(),
+  );
+}
diff --git a/packages/stream_feeds/pubspec.yaml b/packages/stream_feeds/pubspec.yaml
index 8f7723bb..d160bb77 100644
--- a/packages/stream_feeds/pubspec.yaml
+++ b/packages/stream_feeds/pubspec.yaml
@@ -19,6 +19,7 @@ environment:
   sdk: ^3.12.0
 
 dependencies:
+  clock: ^1.1.2
   collection: ^1.18.0
   dio: ^5.9.0
   equatable: ^2.0.5
@@ -41,12 +42,13 @@ dependencies:
     # ignore: invalid_dependency
     git:
       url: https://github.com/GetStream/stream-core-flutter.git
-      ref: f83b5d4d706a79fc429de2d27aead4394b83c1fb
+      ref: 505432f213af8911d54ab3c855d6ad6fbe492fc9
       path: packages/stream_core
   uuid: ^4.5.1
 
 dev_dependencies:
   build_runner: ^2.4.15
+  fake_async: ^1.3.3
   freezed: ^3.0.0
   json_serializable: ^6.9.5
   retrofit_generator: ^10.2.6
diff --git a/packages/stream_feeds/test/client/feeds_client_test.dart b/packages/stream_feeds/test/client/feeds_client_test.dart
index 94432036..a22466fd 100644
--- a/packages/stream_feeds/test/client/feeds_client_test.dart
+++ b/packages/stream_feeds/test/client/feeds_client_test.dart
@@ -48,7 +48,7 @@ void main() {
         );
 
         // Attempt connection - should fail
-        await expectLater(tester.client.connect(), throwsA(isA<ClientException>()));
+        await expectLater(tester.client.connect(), throwsA(isA<StreamException>()));
 
         // Verify state transitions expectation
         await connectionStateExpectation;
@@ -121,7 +121,7 @@ void main() {
       body: (tester) async {
         await expectLater(
           tester.client.connect(),
-          throwsA(isA<ClientException>().having((it) => it.apiError?.code, 'apiError.code', 40)),
+          throwsA(isA<StreamApiException>().having((it) => it.code, 'code', 40)),
         );
       },
     );
@@ -229,9 +229,17 @@ void main() {
       },
       body: (tester) async {
         // Ignored, this would sit in `Authenticating` until the connect timeout swept it up.
+        // Nothing authenticated the credentials and nothing refused them, so the failure
+        // reported is the send's own rather than a verdict on the token.
         await expectLater(
           tester.client.connect(),
-          throwsA(isA<ClientException>()),
+          throwsA(
+            isA<StreamClientException>().having(
+              (it) => it.cause,
+              'cause',
+              isA<Exception>().having((it) => '$it', 'message', contains('mid-handshake')),
+            ),
+          ),
         );
 
         expect(
@@ -256,8 +264,8 @@ void main() {
         await expectLater(
           tester.client.connect(),
           throwsA(
-            isA<ClientException>().having(
-              (it) => it.underlyingError,
+            isA<StreamAuthenticationException>().having(
+              (it) => it.cause,
               'cause',
               isA<Exception>().having((it) => '$it', 'message', contains('token endpoint is down')),
             ),
@@ -330,7 +338,7 @@ void main() {
         // Told they asked for something they already have, rather than silently doing nothing.
         expect(
           () => tester.client.connect(),
-          throwsA(isA<ClientException>().having((it) => it.message, 'message', contains('already available'))),
+          throwsA(isA<StateError>().having((it) => it.message, 'message', contains('already available'))),
         );
 
         // The connection it already had is left alone.
@@ -350,7 +358,7 @@ void main() {
 
         expect(
           () => tester.client.connect(),
-          throwsA(isA<ClientException>().having((it) => it.message, 'message', contains('already in progress'))),
+          throwsA(isA<StateError>().having((it) => it.message, 'message', contains('already in progress'))),
         );
 
         // The attempt already under way is the one that completes.
@@ -451,7 +459,7 @@ void main() {
         await tester.client.disconnect();
 
         // Reported, rather than left waiting on a connection no longer coming.
-        await expectLater(connecting, throwsA(isA<ClientException>()));
+        await expectLater(connecting, throwsA(isA<StreamNetworkException>()));
         expect(tester.client.connectionState.value, isA<Disconnected>());
       },
     );
@@ -1398,9 +1406,9 @@ void main() {
         await expectLater(
           tester.client.connect(),
           throwsA(
-            isA<ClientException>()
+            isA<StreamClientException>()
                 .having((it) => it.message, 'message', 'Failed to create a guest user')
-                .having((it) => it.underlyingError, 'cause', isException),
+                .having((it) => it.cause, 'cause', isException),
           ),
         );
 
@@ -1498,7 +1506,7 @@ void main() {
           error: Exception('Failed to create guest'),
         );
 
-        await expectLater(tester.client.connect(), throwsA(isA<ClientException>()));
+        await expectLater(tester.client.connect(), throwsA(isA<StreamClientException>()));
 
         addTearDown(tester.client.dispose);
       },
diff --git a/packages/stream_feeds/test/state/feed_test.dart b/packages/stream_feeds/test/state/feed_test.dart
index af3ebeae..d7358b46 100644
--- a/packages/stream_feeds/test/state/feed_test.dart
+++ b/packages/stream_feeds/test/state/feed_test.dart
@@ -1,5 +1,8 @@
 // ignore_for_file: avoid_redundant_argument_values
 
+import 'dart:async';
+import 'dart:io';
+
 import 'package:stream_feeds/stream_feeds.dart';
 import 'package:stream_feeds_test/stream_feeds_test.dart';
 
@@ -664,6 +667,49 @@ void main() {
       },
     );
 
+    // The activity below names a feed this client has not cached, which is what
+    // sends the handler to fetch its capabilities.
+    OwnBatchRequest capabilitiesFor(String feed) => OwnBatchRequest(feeds: [feed]);
+
+    // The wait `CapabilitiesRepository` allows itself before its one retry.
+    const retryBackoff = Duration(milliseconds: 500);
+
+    ActivityAddedEvent activityInUncachedFeed() => ActivityAddedEvent(
+      type: EventTypes.activityAdded,
+      createdAt: DateTime.timestamp(),
+      custom: const {},
+      fid: feedId.rawValue,
+      activity: createDefaultActivityResponse(
+        id: 'new-activity',
+        userId: 'user-1',
+        currentFeed: createDefaultFeedResponse(id: 'other', groupId: 'user'),
+      ),
+    );
+
+    feedTest(
+      'ActivityAddedEvent - retries a capabilities fetch that failed on the network',
+      user: const User(id: 'user-1'),
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        registerFallbackValue(const OwnBatchRequest(feeds: []));
+
+        tester.mockApiFailure(
+          (api) => api.ownBatch(ownBatchRequest: capabilitiesFor('other')),
+          error: const StreamNetworkException(message: 'Connection failed'),
+        );
+
+        await tester.emitEvent(activityInUncachedFeed());
+
+        // A blip is worth asking again for; the loop allows exactly one retry.
+        await Future<void>.delayed(retryBackoff + const Duration(milliseconds: 200));
+        tester.verifyApiCalled(
+          (api) => api.ownBatch(ownBatchRequest: capabilitiesFor('other')),
+          times: 2,
+        );
+      },
+    );
+
     feedTest(
       'ActivityAddedEvent - should add activity to feed',
       user: const User(id: 'user-1'),
@@ -4918,4 +4964,296 @@ void main() {
       ),
     );
   });
+
+  group('Attachment uploads', () {
+    const feedId = FeedId(group: 'user', id: 'john');
+
+    const uploadResponse = FileUploadResponse(
+      duration: '0ms',
+      file: 'https://cdn/a',
+      thumbUrl: 'https://cdn/a-thumb',
+    );
+
+    const refused = StreamApiException(
+      message: 'too large',
+      statusCode: 413,
+      code: StreamErrorCode.payloadTooBig,
+    );
+
+    // What the file upload above merges into, once posted.
+    const uploadedAttachment = Attachment(
+      custom: {},
+      type: 'file',
+      assetUrl: 'https://cdn/a',
+      imageUrl: 'https://cdn/a',
+      thumbUrl: 'https://cdn/a-thumb',
+    );
+
+    const mergedRequest = AddActivityRequest(
+      type: 'post',
+      feeds: [],
+      attachments: [uploadedAttachment],
+    );
+
+    const mergedMixedRequest = AddActivityRequest(
+      type: 'post',
+      feeds: [],
+      attachments: [
+        uploadedAttachment,
+        Attachment(
+          custom: {},
+          type: 'image',
+          assetUrl: 'https://cdn/img',
+          imageUrl: 'https://cdn/img',
+        ),
+      ],
+    );
+
+    const mergedCommentRequest = AddCommentRequest(
+      comment: 'Look at this',
+      objectId: 'activity-1',
+      objectType: 'activity',
+      attachments: [uploadedAttachment],
+    );
+
+    const mergedCommentsBatchRequest = AddCommentsBatchRequest(
+      comments: [
+        AddCommentRequest(
+          comment: 'with attachment',
+          objectId: 'activity-1',
+          objectType: 'activity',
+          attachments: [uploadedAttachment],
+        ),
+        AddCommentRequest(
+          comment: 'plain',
+          objectId: 'activity-1',
+          objectType: 'activity',
+        ),
+      ],
+    );
+
+    StreamAttachment uploadAttachment(String id, {AttachmentType type = AttachmentType.file}) {
+      // A real file: the CDN client reads the upload's bytes from its path.
+      final directory = Directory.systemTemp.createTempSync('feeds_upload_test');
+      addTearDown(() => directory.deleteSync(recursive: true));
+
+      File('${directory.path}/$id.bin').writeAsBytesSync(const [1, 2, 3]);
+
+      return StreamAttachment(id: id, type: type, file: AttachmentFile('${directory.path}/$id.bin'));
+    }
+
+    FeedAddActivityRequest requestWithUpload() {
+      return FeedAddActivityRequest(
+        type: 'post',
+        attachmentUploads: [uploadAttachment('a')],
+      );
+    }
+
+    setUpAll(() {
+      registerFallbackValue(<MultipartFile>[]);
+      registerFallbackValue(const AddActivityRequest(type: 'post', feeds: []));
+      registerFallbackValue(const AddCommentRequest(comment: 'fallback', objectId: 'x', objectType: 'activity'));
+    });
+
+    Future<Result<FileUploadResponse>> uploadCall(CdnApi cdn) => cdn.uploadFile(
+      file: any(named: 'file'),
+      onUploadProgress: any(named: 'onUploadProgress'),
+      cancelToken: any(named: 'cancelToken'),
+    );
+
+    Future<Result<ImageUploadResponse>> imageUploadCall(CdnApi cdn) => cdn.uploadImage(
+      file: any(named: 'file'),
+      onUploadProgress: any(named: 'onUploadProgress'),
+      cancelToken: any(named: 'cancelToken'),
+    );
+
+    feedTest(
+      'addActivity() - uploads the attachments and posts the merged request',
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        tester.mockCdn(uploadCall, result: uploadResponse);
+        tester.mockApi(
+          (api) => api.addActivity(addActivityRequest: mergedRequest),
+          result: AddActivityResponse(
+            duration: '0ms',
+            activity: createDefaultActivityResponse(
+              id: 'activity-1',
+              feeds: [feedId.rawValue],
+            ),
+          ),
+        );
+
+        final result = await tester.feed.addActivity(request: requestWithUpload());
+
+        expect(result.isSuccess, isTrue);
+      },
+      // The uploaded file arrived merged into the posted request, its upload
+      // queue spent.
+      verify: (tester) => tester.verifyApi(
+        (api) => api.addActivity(addActivityRequest: mergedRequest),
+      ),
+    );
+
+    feedTest(
+      "addActivity() - fails with the upload's own error, posting nothing",
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        tester.mockCdnFailure(uploadCall, error: refused);
+
+        final result = await tester.feed.addActivity(request: requestWithUpload());
+
+        // The upload's failure surfaces classified, and nothing was posted
+        // for an activity whose attachments never made it.
+        expect(result.exceptionOrNull(), same(refused));
+        tester.verifyNeverCalled(
+          (api) => api.addActivity(addActivityRequest: any(named: 'addActivityRequest')),
+        );
+      },
+    );
+
+    feedTest(
+      'addActivity() - routes each attachment through its own endpoint and merges them all',
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        tester.mockCdn(uploadCall, result: uploadResponse);
+        tester.mockCdn(
+          imageUploadCall,
+          result: const ImageUploadResponse(duration: '0ms', file: 'https://cdn/img'),
+        );
+        tester.mockApi(
+          (api) => api.addActivity(addActivityRequest: mergedMixedRequest),
+          result: AddActivityResponse(
+            duration: '0ms',
+            activity: createDefaultActivityResponse(id: 'activity-1', feeds: [feedId.rawValue]),
+          ),
+        );
+
+        final result = await tester.feed.addActivity(
+          request: FeedAddActivityRequest(
+            type: 'post',
+            attachmentUploads: [
+              uploadAttachment('a'),
+              uploadAttachment('b', type: AttachmentType.image),
+            ],
+          ),
+        );
+
+        expect(result.isSuccess, isTrue);
+      },
+      // The file went through the file endpoint, the image through the image
+      // one, and both arrived merged into the posted request.
+      verify: (tester) => tester.verifyApi(
+        (api) => api.addActivity(addActivityRequest: mergedMixedRequest),
+      ),
+    );
+
+    feedTest(
+      'addComment() - uploads the attachments and posts the merged comment',
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        tester.mockCdn(uploadCall, result: uploadResponse);
+        tester.mockApi(
+          (api) => api.addComment(addCommentRequest: mergedCommentRequest),
+          result: createDefaultAddCommentResponse(objectId: 'activity-1', text: 'Look at this'),
+        );
+
+        final result = await tester.feed.addComment(
+          request: ActivityAddCommentRequest(
+            activityId: 'activity-1',
+            comment: 'Look at this',
+            attachmentUploads: [uploadAttachment('a')],
+          ),
+        );
+
+        expect(result.isSuccess, isTrue);
+      },
+      // The comment went out with the uploaded attachment merged in.
+      verify: (tester) => tester.verifyApi(
+        (api) => api.addComment(addCommentRequest: mergedCommentRequest),
+      ),
+    );
+
+    feedTest(
+      'addActivity() - refuses two attachments that share an id, before uploading anything',
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        // A misuse rather than a failure, so it escapes the `Result` the method
+        // otherwise reports through — a caller that only folds cannot see it.
+        await expectLater(
+          tester.feed.addActivity(
+            request: FeedAddActivityRequest(
+              type: 'post',
+              attachmentUploads: [uploadAttachment('same'), uploadAttachment('same')],
+            ),
+          ),
+          throwsArgumentError,
+        );
+
+        tester.verifyNeverCalled(
+          (api) => api.addActivity(addActivityRequest: any(named: 'addActivityRequest')),
+        );
+      },
+    );
+
+    feedTest(
+      "addComment() - fails with the upload's own error, posting nothing",
+      build: (client) => client.feedFromId(feedId),
+      setUp: (tester) => tester.getOrCreate(),
+      body: (tester) async {
+        tester.mockCdnFailure(uploadCall, error: refused);
+
+        final result = await tester.feed.addComment(
+          request: ActivityAddCommentRequest(
+            activityId: 'activity-1',
+            comment: 'Look at this',
+            attachmentUploads: [uploadAttachment('a')],
+          ),
+        );
+
+        expect(result.exceptionOrNull(), same(refused));
+        tester.verifyNeverCalled(
+          (api) => api.addComment(addCommentRequest: any(named: 'addCommentRequest')),
+        );
+      },
+    );
+
+    activityTest(
+      "addCommentsBatch() - uploads each request's attachments before posting the batch",
+      build: (client) => client.activity(activityId: 'activity-1', fid: feedId),
+      body: (tester) async {
+        tester.mockCdn(uploadCall, result: uploadResponse);
+        tester.mockApi(
+          (api) => api.addCommentsBatch(addCommentsBatchRequest: mergedCommentsBatchRequest),
+          result: AddCommentsBatchResponse(
+            duration: '0ms',
+            comments: [
+              createDefaultCommentResponse(id: 'comment-1', objectId: 'activity-1', text: 'with attachment'),
+              createDefaultCommentResponse(id: 'comment-2', objectId: 'activity-1', text: 'plain'),
+            ],
+          ),
+        );
+
+        final result = await tester.activity.addCommentsBatch([
+          ActivityAddCommentRequest(
+            activityId: 'activity-1',
+            comment: 'with attachment',
+            attachmentUploads: [uploadAttachment('a')],
+          ),
+          const ActivityAddCommentRequest(activityId: 'activity-1', comment: 'plain'),
+        ]);
+
+        expect(result.isSuccess, isTrue);
+      },
+      // Each comment keeps its own attachments: the upload landed on the
+      // request that carried it, and only on that one.
+      verify: (tester) => tester.verifyApi(
+        (api) => api.addCommentsBatch(addCommentsBatchRequest: mergedCommentsBatchRequest),
+      ),
+    );
+  });
 }
diff --git a/packages/stream_feeds/test/utils/batcher_test.dart b/packages/stream_feeds/test/utils/batcher_test.dart
new file mode 100644
index 00000000..6c07ad2d
--- /dev/null
+++ b/packages/stream_feeds/test/utils/batcher_test.dart
@@ -0,0 +1,134 @@
+import 'package:fake_async/fake_async.dart';
+import 'package:stream_feeds/src/utils/batcher.dart';
+import 'package:stream_feeds_test/stream_feeds_test.dart';
+
+void main() {
+  group('Batcher', () {
+    ({Batcher<String, int> batcher, List<List<String>> batches}) build({
+      Duration interval = const Duration(seconds: 2),
+    }) {
+      final batches = <List<String>>[];
+      final batcher = Batcher<String, int>(
+        interval: interval,
+        action: (items) async {
+          batches.add(items);
+          return items.length;
+        },
+      );
+
+      return (batcher: batcher, batches: batches);
+    }
+
+    test('asks straight away, having nothing to wait behind', () {
+      fakeAsync((async) {
+        final (:batcher, :batches) = build();
+
+        batcher.add('a').ignore();
+        async.flushMicrotasks();
+
+        expect(batches, [
+          ['a'],
+        ]);
+      });
+    });
+
+    test('starts a new batch once the one before it has run', () {
+      fakeAsync((async) {
+        final (:batcher, :batches) = build();
+
+        batcher.add('a').ignore();
+        async.flushMicrotasks();
+
+        // The second add has to plan a batch of its own: joining the settled
+        // one would answer it with a result gathered before it asked.
+        batcher.add('b').ignore();
+        async.elapse(const Duration(seconds: 3));
+
+        expect(batches, [
+          ['a'],
+          ['b'],
+        ]);
+      });
+    });
+
+    test('answers each add with the batch that carried it', () {
+      fakeAsync((async) {
+        final batcher = Batcher<String, String>(
+          action: (items) async => items.join(','),
+        );
+
+        String? first;
+        String? second;
+
+        batcher.add('a').then((it) => first = it).ignore();
+        async.flushMicrotasks();
+
+        batcher.add('b').then((it) => second = it).ignore();
+        async.elapse(const Duration(seconds: 3));
+
+        expect(first, 'a');
+        // Joining the settled batch would answer this with 'a' — a result
+        // gathered before it asked.
+        expect(second, 'b');
+      });
+    });
+
+    test('answers everyone in a batch with the whole batch', () {
+      fakeAsync((async) {
+        final batcher = Batcher<String, String>(
+          action: (items) async => items.join(','),
+        );
+        final answers = <String, String>{};
+
+        batcher.add('a').ignore();
+        async.flushMicrotasks();
+
+        // Both land inside one window, so both are told what the batch as a
+        // whole came back with, not just their own part of it.
+        batcher.add('b').then((it) => answers['b'] = it).ignore();
+        batcher.add('c').then((it) => answers['c'] = it).ignore();
+        async.elapse(const Duration(seconds: 3));
+
+        expect(answers, {'b': 'b,c', 'c': 'b,c'});
+      });
+    });
+
+    test('collects everything that arrives inside one window', () {
+      fakeAsync((async) {
+        final (:batcher, :batches) = build();
+
+        batcher.add('a').ignore();
+        async.flushMicrotasks();
+
+        batcher.add('b').ignore();
+        batcher.add('c').ignore();
+        async.elapse(const Duration(seconds: 3));
+
+        expect(batches, [
+          ['a'],
+          ['b', 'c'],
+        ], reason: 'the two behind the window went out together');
+      });
+    });
+
+    test('measures the window against the clock it is given', () {
+      fakeAsync((async) {
+        final (:batcher, :batches) = build();
+
+        batcher.add('a').ignore();
+        async.flushMicrotasks();
+
+        // Reading the wall clock instead would hold this add inside the window
+        // however long a test waited.
+        async.elapse(const Duration(seconds: 3));
+        batcher.add('b').ignore();
+        async.flushMicrotasks();
+
+        expect(batches, [
+          ['a'],
+          ['b'],
+        ], reason: 'the window had passed, so it asked on the spot');
+      });
+    });
+  });
+}
diff --git a/packages/stream_feeds/test/utils/uploader_test.dart b/packages/stream_feeds/test/utils/uploader_test.dart
new file mode 100644
index 00000000..55a4924e
--- /dev/null
+++ b/packages/stream_feeds/test/utils/uploader_test.dart
@@ -0,0 +1,342 @@
+import 'dart:async';
+
+import 'package:stream_core/stream_core.dart';
+import 'package:stream_feeds/src/generated/api/models.dart' as api;
+import 'package:stream_feeds/src/utils/uploader.dart';
+import 'package:stream_feeds_test/stream_feeds_test.dart';
+
+class _TestRequest implements HasAttachments<_TestRequest> {
+  const _TestRequest({this.attachments, this.attachmentUploads});
+
+  @override
+  final List<api.Attachment>? attachments;
+
+  @override
+  final List<StreamAttachment>? attachmentUploads;
+
+  @override
+  _TestRequest withAttachments({
+    List<api.Attachment>? attachments,
+    List<StreamAttachment>? attachmentUploads,
+  }) => _TestRequest(attachments: attachments, attachmentUploads: attachmentUploads);
+}
+
+/// A CDN whose outcome per file is scripted by [outcomes].
+class _FakeCdn implements CdnClient {
+  _FakeCdn(this.outcomes);
+
+  final Map<AttachmentFile, Result<UploadedFile>> outcomes;
+
+  Future<Result<UploadedFile>> _upload(AttachmentFile file) async => outcomes[file]!;
+
+  @override
+  Future<Result<UploadedFile>> uploadFile(
+    AttachmentFile file, {
+    ProgressCallback? onProgress,
+    CancelToken? cancelToken,
+  }) => _upload(file);
+
+  @override
+  Future<Result<UploadedFile>> uploadImage(
+    AttachmentFile image, {
+    ProgressCallback? onProgress,
+    CancelToken? cancelToken,
+  }) => _upload(image);
+
+  @override
+  Future<Result<void>> deleteFile(String url, {CancelToken? cancelToken}) async => const Result.success(null);
+
+  @override
+  Future<Result<void>> deleteImage(String url, {CancelToken? cancelToken}) async => const Result.success(null);
+}
+
+/// A CDN that parks every upload until the test answers it, so how many are on
+/// the wire at once is observable.
+class _GatedCdn implements CdnClient {
+  final _pending = <AttachmentFile, Completer<Result<UploadedFile>>>{};
+
+  /// How many uploads are on the wire right now.
+  int get inFlight => _pending.values.where((it) => !it.isCompleted).length;
+
+  /// Answers the upload of [file] with [outcome].
+  void answer(AttachmentFile file, Result<UploadedFile> outcome) => _pending[file]!.complete(outcome);
+
+  Future<Result<UploadedFile>> _upload(AttachmentFile file) =>
+      (_pending[file] ??= Completer<Result<UploadedFile>>()).future;
+
+  @override
+  Future<Result<UploadedFile>> uploadFile(
+    AttachmentFile file, {
+    ProgressCallback? onProgress,
+    CancelToken? cancelToken,
+  }) => _upload(file);
+
+  @override
+  Future<Result<UploadedFile>> uploadImage(
+    AttachmentFile image, {
+    ProgressCallback? onProgress,
+    CancelToken? cancelToken,
+  }) => _upload(image);
+
+  @override
+  Future<Result<void>> deleteFile(String url, {CancelToken? cancelToken}) async => const Result.success(null);
+
+  @override
+  Future<Result<void>> deleteImage(String url, {CancelToken? cancelToken}) async => const Result.success(null);
+}
+
+StreamAttachment _attachment(String id, AttachmentFile file) => StreamAttachment(
+  id: id,
+  type: AttachmentType.file,
+  file: file,
+);
+
+const _refused = StreamApiException(message: 'too large', statusCode: 413, code: StreamErrorCode.payloadTooBig);
+
+void main() {
+  final fileA = AttachmentFile.fromData(Uint8List(0));
+  final fileB = AttachmentFile.fromData(Uint8List(0));
+
+  const uploadedA = UploadedFile(fileUrl: 'https://cdn/a', thumbUrl: 'https://cdn/a-thumb');
+  const uploadedB = UploadedFile(fileUrl: 'https://cdn/b');
+
+  group('processRequest', () {
+    test('returns the request untouched when there is nothing to upload', () async {
+      final uploader = StreamAttachmentUploader(cdn: _FakeCdn({}));
+      const request = _TestRequest(attachmentUploads: []);
+
+      final result = await uploader.processRequest(request);
+
+      expect(result.getOrNull(), same(request));
+    });
+
+    test('merges every uploaded attachment and empties the upload queue', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}),
+      );
+      final request = _TestRequest(
+        attachmentUploads: [_attachment('a', fileA), _attachment('b', fileB)],
+      );
+
+      final processed = await uploader.processRequest(request);
+
+      final updated = processed.getOrNull()!;
+      expect(
+        updated.attachments?.map((it) => it.assetUrl),
+        unorderedEquals(['https://cdn/a', 'https://cdn/b']),
+      );
+      expect(
+        updated.attachments?.map((it) => it.thumbUrl),
+        unorderedEquals(['https://cdn/a-thumb', null]),
+      );
+      expect(updated.attachmentUploads, isEmpty);
+    });
+
+    test('keeps existing attachments and does not duplicate an already merged upload', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA)}),
+      );
+      const existing = api.Attachment(custom: {}, assetUrl: 'https://cdn/existing');
+      const alreadyMerged = api.Attachment(
+        custom: {},
+        type: 'file',
+        assetUrl: 'https://cdn/a',
+        imageUrl: 'https://cdn/a',
+      );
+      final request = _TestRequest(
+        attachments: const [existing, alreadyMerged],
+        attachmentUploads: [_attachment('a', fileA)],
+      );
+
+      final processed = await uploader.processRequest(request);
+
+      expect(
+        processed.getOrNull()!.attachments?.map((it) => it.assetUrl),
+        unorderedEquals(['https://cdn/existing', 'https://cdn/a']),
+      );
+    });
+
+    test("fails as one when an upload fails, carrying the upload's own error", () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}),
+      );
+      final request = _TestRequest(
+        attachmentUploads: [_attachment('a', fileA), _attachment('b', fileB)],
+      );
+
+      final processed = await uploader.processRequest(request);
+
+      expect(processed.exceptionOrNull(), same(_refused));
+    });
+
+    test('without eagerError, keeps the failed upload queued for a later attempt', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}),
+      );
+      final request = _TestRequest(
+        attachmentUploads: [_attachment('a', fileA), _attachment('b', fileB)],
+      );
+
+      final processed = await uploader.processRequest(request, eagerError: false);
+
+      final updated = processed.getOrNull()!;
+      expect(updated.attachments?.map((it) => it.assetUrl), ['https://cdn/a']);
+      expect(updated.attachmentUploads?.map((it) => it.id), ['b']);
+    });
+  });
+
+  group('processRequestsBatch', () {
+    test('gives each request only its own uploaded attachments', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}),
+      );
+      final first = _TestRequest(attachmentUploads: [_attachment('a', fileA)]);
+      final second = _TestRequest(attachmentUploads: [_attachment('b', fileB)]);
+
+      final processed = await uploader.processRequestsBatch([first, second]);
+      final requests = processed.getOrNull()!;
+
+      expect(requests.first.attachments?.map((it) => it.assetUrl), ['https://cdn/a']);
+      expect(requests.last.attachments?.map((it) => it.assetUrl), ['https://cdn/b']);
+      expect(requests.every((it) => it.attachmentUploads!.isEmpty), isTrue);
+    });
+
+    test('leaves a request whose upload failed queued without touching the others', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}),
+      );
+      final ok = _TestRequest(attachmentUploads: [_attachment('a', fileA)]);
+      final failed = _TestRequest(attachmentUploads: [_attachment('b', fileB)]);
+
+      final processed = await uploader.processRequestsBatch([ok, failed], eagerError: false);
+      final requests = processed.getOrNull()!;
+
+      expect(requests.first.attachments?.map((it) => it.assetUrl), ['https://cdn/a']);
+      expect(requests.first.attachmentUploads, isEmpty);
+      expect(requests.last.attachments, isNull);
+      expect(requests.last.attachmentUploads?.map((it) => it.id), ['b']);
+    });
+
+    test('returns the requests in the order they were given', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}),
+      );
+      final withB = _TestRequest(attachmentUploads: [_attachment('b', fileB)]);
+      final withA = _TestRequest(attachmentUploads: [_attachment('a', fileA)]);
+
+      final processed = await uploader.processRequestsBatch([withB, withA]);
+      final requests = processed.getOrNull()!;
+
+      expect(requests.first.attachments?.map((it) => it.assetUrl), ['https://cdn/b']);
+      expect(requests.last.attachments?.map((it) => it.assetUrl), ['https://cdn/a']);
+    });
+
+    test('leaves a request with nothing to upload exactly as it was', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA)}),
+      );
+      const nothingToUpload = _TestRequest();
+      final withUpload = _TestRequest(attachmentUploads: [_attachment('a', fileA)]);
+
+      final processed = await uploader.processRequestsBatch([nothingToUpload, withUpload]);
+      final requests = processed.getOrNull()!;
+
+      expect(requests.first, same(nothingToUpload), reason: 'nothing about it changed');
+      expect(requests.last.attachments?.map((it) => it.assetUrl), ['https://cdn/a']);
+    });
+
+    test('bounds the uploads across every request, not within each one', () async {
+      final cdn = _GatedCdn();
+      final uploader = StreamAttachmentUploader(cdn: cdn);
+      final files = [for (var i = 0; i < 4; i++) AttachmentFile.fromData(Uint8List(0))];
+      final requests = <_TestRequest>[];
+      for (var i = 0; i < files.length; i++) {
+        requests.add(_TestRequest(attachmentUploads: [_attachment('a$i', files[i])]));
+      }
+
+      final pending = uploader.processRequestsBatch(requests, maxConcurrent: 2);
+      await pumpEventQueue();
+
+      expect(cdn.inFlight, 2, reason: 'four requests of one attachment each, two uploads at a time');
+
+      cdn.answer(files[0], const Result.success(uploadedA));
+      await pumpEventQueue();
+
+      expect(cdn.inFlight, 2, reason: 'a freed slot takes exactly one more');
+
+      for (final file in files.skip(1)) {
+        cdn.answer(file, const Result.success(uploadedA));
+        await pumpEventQueue();
+      }
+
+      expect((await pending).getOrNull(), hasLength(4));
+    });
+
+    test('refuses two requests that share an attachment id', () async {
+      final uploader = StreamAttachmentUploader(cdn: _FakeCdn({}));
+      final shared = _attachment('same', fileA);
+
+      await expectLater(
+        uploader.processRequestsBatch([
+          _TestRequest(attachmentUploads: [shared]),
+          _TestRequest(attachmentUploads: [shared]),
+        ]),
+        throwsArgumentError,
+        reason: 'one batch could not tell them apart',
+      );
+    });
+
+    test('succeeds with nothing when there are no requests', () async {
+      final uploader = StreamAttachmentUploader(cdn: _FakeCdn({}));
+
+      final processed = await uploader.processRequestsBatch<_TestRequest>([]);
+
+      expect(processed.getOrNull(), isEmpty);
+    });
+
+    test('processes every request when all uploads succeed', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.success(uploadedB)}),
+      );
+      final requests = [
+        _TestRequest(attachmentUploads: [_attachment('a', fileA)]),
+        _TestRequest(attachmentUploads: [_attachment('b', fileB)]),
+      ];
+
+      final processed = await uploader.processRequestsBatch(requests);
+
+      expect(processed.getOrNull(), hasLength(2));
+    });
+
+    test('fails as one when any request fails', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}),
+      );
+      final requests = [
+        _TestRequest(attachmentUploads: [_attachment('a', fileA)]),
+        _TestRequest(attachmentUploads: [_attachment('b', fileB)]),
+      ];
+
+      final processed = await uploader.processRequestsBatch(requests);
+
+      expect(processed.exceptionOrNull(), same(_refused));
+    });
+
+    test('without eagerError, keeps every request, the failed upload staying queued in its own', () async {
+      final uploader = StreamAttachmentUploader(
+        cdn: _FakeCdn({fileA: const Result.success(uploadedA), fileB: const Result.failure(_refused)}),
+      );
+      final requests = [
+        _TestRequest(attachmentUploads: [_attachment('a', fileA)]),
+        _TestRequest(attachmentUploads: [_attachment('b', fileB)]),
+      ];
+
+      final processed = await uploader.processRequestsBatch(requests, eagerError: false);
+
+      final kept = processed.getOrNull()!;
+      expect(kept, hasLength(2));
+      expect(kept.first.attachmentUploads, isEmpty);
+      expect(kept.last.attachmentUploads?.map((it) => it.id), ['b']);
+    });
+  });
+}
diff --git a/packages/stream_feeds_test/lib/src/helpers/test_data.dart b/packages/stream_feeds_test/lib/src/helpers/test_data.dart
index 583b5569..cecddd0a 100644
--- a/packages/stream_feeds_test/lib/src/helpers/test_data.dart
+++ b/packages/stream_feeds_test/lib/src/helpers/test_data.dart
@@ -103,6 +103,7 @@ ActivityResponse createDefaultActivityResponse({
   int? friendReactionCount,
   Map<String, int>? metrics,
   ActivityResponseRestrictReplies? restrictReplies,
+  FeedResponse? currentFeed,
 }) {
   latestReactions = latestReactions.isEmpty ? ownReactions : latestReactions;
   reactionGroups = switch (reactionGroups.isNotEmpty) {
@@ -157,6 +158,7 @@ ActivityResponse createDefaultActivityResponse({
     reactionCount: reactionGroups.values.sumOf((group) => group.count),
     reactionGroups: reactionGroups,
     restrictReplies: restrictReplies ?? ActivityResponseRestrictReplies.everyone,
+    currentFeed: currentFeed,
     score: 0,
     searchData: const {},
     shareCount: 0,
diff --git a/packages/stream_feeds_test/lib/src/testers/base_tester.dart b/packages/stream_feeds_test/lib/src/testers/base_tester.dart
index 04860786..34ce2db5 100644
--- a/packages/stream_feeds_test/lib/src/testers/base_tester.dart
+++ b/packages/stream_feeds_test/lib/src/testers/base_tester.dart
@@ -107,7 +107,7 @@ abstract base class BaseTester<S> with ApiMockerMixin, CdnMockerMixin {
   /// // Custom error code (triggers reconnection)
   /// tester.mockFailedAuth(errorCode: 5);
   ///
-  /// await expectLater(client.connect(), throwsA(isA<ClientException>()));
+  /// await expectLater(client.connect(), throwsA(isA<StreamApiException>()));
   /// ```
   void mockFailedAuth({int errorCode = 40}) {
     return _wsTester.mockFailedAuth(errorCode: errorCode);
@@ -122,7 +122,7 @@ abstract base class BaseTester<S> with ApiMockerMixin, CdnMockerMixin {
   /// ```dart
   /// tester.mockFailedSend();
   ///
-  /// await expectLater(client.connect(), throwsA(isA<ClientException>()));
+  /// await expectLater(client.connect(), throwsA(isA<StreamClientException>()));
   /// ```
   void mockFailedSend({Object? error}) {
     return _wsTester.mockFailedSend(error: error);
diff --git a/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart b/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart
index 8b637921..ccbe0281 100644
--- a/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart
+++ b/packages/stream_feeds_test/lib/src/testers/websocket_tester.dart
@@ -90,7 +90,7 @@ final class WebSocketTester {
   /// Example:
   /// ```dart
   /// wsTester.mockFailedSend();
-  /// await expectLater(client.connect(), throwsA(isA<ClientException>()));
+  /// await expectLater(client.connect(), throwsA(isA<StreamClientException>()));
   /// ```
   void mockFailedSend({Object? error}) {
     _resetFunction?.call(); // Reset previous mocks if any
@@ -137,7 +137,7 @@ final class WebSocketTester {
   ///
   /// await expectLater(
   ///   client.connect(),
-  ///   throwsA(isA<ClientException>()),
+  ///   throwsA(isA<StreamApiException>()),
   /// );
   /// ```
   void mockFailedAuth({int errorCode = 40}) {