diff --git a/src/datadog/impl/core/storage/feature_event.cpp b/src/datadog/impl/core/storage/feature_event.cpp
index e5b19cb1..c08b2f3b 100644
--- a/src/datadog/impl/core/storage/feature_event.cpp
+++ b/src/datadog/impl/core/storage/feature_event.cpp
@@ -91,16 +91,21 @@ bool FeatureEventStorage::Initialize(
 bool FeatureEventStorage::DeletePendingBatches() {
   // Thread-safety/synchronization considerations:
   //
-  // 1. We only delete from _pending_root, which the upload thread never reads from, so
-  //    we don't have to worry about contention with other threads
-  // 2. The storage thread performs deletion synchronously, i.e. there can be no file
+  // 1. The storage thread performs deletion synchronously, i.e. there can be no file
   //    writes happening concurrently during deletion
-  // 3. When the storage thread _does_ perform writes, it does so atomically and closes
+  // 2. When the storage thread _does_ perform writes, it does so atomically and closes
   //    the file, so there are no open handles to any of the files we're deleting
-  // 4. This function is called on the storage thread
+  // 3. This function is called on the storage thread
   //
-  // Therefore, we can safely assume that we have exclusive access to _pending_root
-  // during this function call.
+  // Race with upload-thread age-based eviction:
+  //
+  // The upload thread runs an eviction pass over _pending_root (deleting files older
+  // than 18h) independently of this deletion. If the upload thread deletes a file
+  // between our ListFiles call and our Delete call for that file, the Delete returns
+  // DoesNotExist. We treat that as "already gone; continue" rather than a fatal error:
+  // the file was expired anyway and would never have been uploaded. Aborting on
+  // DoesNotExist would strand all younger files that appear later in the sorted
+  // listing.
 
   // Use a FilesystemWrapper to handle path encoding transparently
   FilesystemWrapper fsw(_fs);
@@ -137,20 +142,31 @@ bool FeatureEventStorage::DeletePendingBatches() {
       return false;
     }
 
-    // Attempt to delete the batch file, aborting on failure
+    // Attempt to delete the batch file
     const FilesystemResult delete_res = fsw.Delete(file_path.CStr());
-    if (delete_res != FilesystemResult::OK) {
-      _logger.Error(
-          "Could not delete batch file on consent change",
-          {{"path", file_path.Get()}, {"error", FilesystemResultStr(delete_res)}}
+    if (delete_res == FilesystemResult::OK) {
+      _logger.Debug(
+          "Deleted batch file due to consent change", {{"path", file_path.Get()}}
       );
-      return false;
+      continue;
+    }
+
+    // If the file is already gone, the upload thread's eviction pass beat us to it.
+    // The file was expired and would not have been uploaded; skip it and continue.
+    if (delete_res == FilesystemResult::DoesNotExist) {
+      _logger.Debug(
+          "Pending batch file already evicted before consent-revocation cleanup; "
+          "skipping",
+          {{"path", file_path.Get()}}
+      );
+      continue;
     }
 
-    // Batch file deleted successfully
-    _logger.Debug(
-        "Deleted batch file due to consent change", {{"path", file_path.Get()}}
+    _logger.Error(
+        "Could not delete batch file on consent change",
+        {{"path", file_path.Get()}, {"error", FilesystemResultStr(delete_res)}}
     );
+    return false;
   }
 
   // Success: all batch files in pending directory deleted
@@ -170,6 +186,16 @@ bool FeatureEventStorage::MigratePendingBatchesToGranted() {
   //    a file rename operation is atomic on both POSIX and Windows: the upload thread
   //    will never see a "partially-moved" file.
   //
+  // Race with upload-thread age-based eviction:
+  //
+  // The upload thread runs an eviction pass over _pending_root (deleting files older
+  // than 18h) independently of this migration. If the upload thread deletes a file
+  // between our ListFiles call and our Rename call for that file, the Rename returns
+  // DoesNotExist. We treat that as "already gone; continue" rather than a fatal error:
+  // the file was expired anyway and would never have been uploaded. Aborting on
+  // DoesNotExist would strand all younger files that appear later in the sorted
+  // listing.
+  //
   // Additional considerations re: filename conflicts:
   //
   // - The files we're moving are named with timestamps indicating file creation time,
@@ -261,6 +287,17 @@ bool FeatureEventStorage::MigratePendingBatchesToGranted() {
       continue;
     }
 
+    // If the rename failed because the source file no longer exists, the upload
+    // thread's age-based eviction has already removed it. The file was expired and
+    // would not have been uploaded; skip it and continue migrating the remaining files.
+    if (rename_res == FilesystemResult::DoesNotExist) {
+      _logger.Debug(
+          "Pending batch file already evicted before migration; skipping",
+          {{"path", src_file_path.Get()}}
+      );
+      continue;
+    }
+
     // If the move failed for any other reason, abort the migration operation
     _logger.Error(
         "Could not migrate batch file on consent change: rename failed",
diff --git a/src/datadog/impl/core/storage/filesystem.hpp b/src/datadog/impl/core/storage/filesystem.hpp
index 47c17f41..f67cb3f8 100644
--- a/src/datadog/impl/core/storage/filesystem.hpp
+++ b/src/datadog/impl/core/storage/filesystem.hpp
@@ -206,6 +206,19 @@ class IFilesystem {
    */
   virtual FilesystemResult Close(PlatformFileHandle handle) = 0;
 
+  struct GetFileSizeResult {
+    FilesystemResult value;
+    size_t size;  // valid only when value == OK
+  };
+
+  /**
+   * Returns the size of the regular file at `path` in bytes.
+   *
+   * On success, returns {OK, size}. If no file exists at the path, returns
+   * {DoesNotExist, 0}. Any other error code represents a failure.
+   */
+  virtual GetFileSizeResult GetFileSize(const PlatformPath& path) = 0;
+
   /**
    * Deletes a regular file.
    *
diff --git a/src/datadog/impl/core/storage/filesystem_posix.cpp b/src/datadog/impl/core/storage/filesystem_posix.cpp
index 318431e3..57dc95fc 100644
--- a/src/datadog/impl/core/storage/filesystem_posix.cpp
+++ b/src/datadog/impl/core/storage/filesystem_posix.cpp
@@ -306,6 +306,14 @@ class PosixFilesystem final : public IFilesystem {
     return map_errno(errno);
   }
 
+  GetFileSizeResult GetFileSize(const PlatformPath& path) override {
+    struct stat st;
+    if (stat(path.Get(), &st) != 0) {
+      return {map_errno(errno), 0};
+    }
+    return {FilesystemResult::OK, static_cast<size_t>(st.st_size)};
+  }
+
   FilesystemResult Delete(const PlatformPath& path) override {
     const int result = unlink(path.Get());
     if (result == 0) {
diff --git a/src/datadog/impl/core/storage/filesystem_windows.cpp b/src/datadog/impl/core/storage/filesystem_windows.cpp
index b9c71dbc..23dc2f64 100644
--- a/src/datadog/impl/core/storage/filesystem_windows.cpp
+++ b/src/datadog/impl/core/storage/filesystem_windows.cpp
@@ -352,6 +352,17 @@ class WindowsFilesystem final : public IFilesystem {
     return FilesystemResult::OK;
   }
 
+  GetFileSizeResult GetFileSize(const PlatformPath& path) override {
+    WIN32_FILE_ATTRIBUTE_DATA data;
+    if (GetFileAttributesExW(path.Get(), GetFileExInfoStandard, &data) == 0) {
+      return {map_error(GetLastError()), 0};
+    }
+    const size_t size = static_cast<size_t>(
+        (static_cast<uint64_t>(data.nFileSizeHigh) << 32) | data.nFileSizeLow
+    );
+    return {FilesystemResult::OK, size};
+  }
+
   FilesystemResult Delete(const PlatformPath& path) override {
     const BOOL result = DeleteFileW(path.Get());
     if (result == 0) {
diff --git a/src/datadog/impl/core/storage/filesystem_wrapper.cpp b/src/datadog/impl/core/storage/filesystem_wrapper.cpp
index 7da6a89a..0bb593ed 100644
--- a/src/datadog/impl/core/storage/filesystem_wrapper.cpp
+++ b/src/datadog/impl/core/storage/filesystem_wrapper.cpp
@@ -116,6 +116,13 @@ FilesystemWrapper::OpenFileResult FilesystemWrapper::OpenForRead(
   return {res.value, File{_fs, res.handle}};
 }
 
+IFilesystem::GetFileSizeResult FilesystemWrapper::GetFileSize(const char* path) {
+  if (!_path.Encode(path)) {
+    return {FilesystemResult::PathEncodingFailed, 0};
+  }
+  return _fs.GetFileSize(_path);
+}
+
 FilesystemResult FilesystemWrapper::Delete(const char* path) {
   if (!_path.Encode(path)) {
     return FilesystemResult::PathEncodingFailed;
diff --git a/src/datadog/impl/core/storage/filesystem_wrapper.hpp b/src/datadog/impl/core/storage/filesystem_wrapper.hpp
index 31852cbf..131e0305 100644
--- a/src/datadog/impl/core/storage/filesystem_wrapper.hpp
+++ b/src/datadog/impl/core/storage/filesystem_wrapper.hpp
@@ -96,6 +96,7 @@ class FilesystemWrapper {
   );
   OpenFileResult OpenForWrite(const char* path, bool append, bool hold_advisory_lock);
   OpenFileResult OpenForRead(const char* path, bool hold_advisory_lock);
+  IFilesystem::GetFileSizeResult GetFileSize(const char* path);
   FilesystemResult Delete(const char* path);
   FilesystemResult DeleteDirectory(const char* path);
   FilesystemResult Rename(const char* src, const char* dst);
diff --git a/src/datadog/impl/core/storage_write.cpp b/src/datadog/impl/core/storage_write.cpp
index b90ca7f6..03de174f 100644
--- a/src/datadog/impl/core/storage_write.cpp
+++ b/src/datadog/impl/core/storage_write.cpp
@@ -8,6 +8,7 @@
 
 #include <algorithm>
 #include <array>
+#include <cctype>
 #include <charconv>
 #include <chrono>
 #include <iostream>
@@ -162,12 +163,17 @@ bool BatchWriter::FlushEvent(
     num_bytes += TLVBlockHeader::SIZE + event_metadata.size();
   }
 
-  // If the encoded size of this write (metadata + event, with headers) exceeds the
-  // maximum configured size for a single batch file, reject the event outright: even a
-  // brand new batch file could not contain it
-  if (num_bytes > _config.max_file_size) {
+  // Reject the event if it exceeds either the per-event cap or the batch file size
+  // limit. In the normal production config max_event_size < max_file_size, so the event
+  // cap is the binding constraint; taking the minimum preserves the file-size guard for
+  // configurations where max_file_size may be set smaller than max_event_size.
+  const size_t max_write_size = std::min(_config.max_event_size, _config.max_file_size);
+  if (num_bytes > max_write_size) {
     _diagnostic_logger.Error(
-        "Event dropped; size of single event exceeds max batch file size"
+        "Event dropped; size of single event exceeds max write size",
+        {{"event_size", num_bytes},
+         {"max_event_size", _config.max_event_size},
+         {"max_file_size", _config.max_file_size}}
     );
     return false;
   }
@@ -288,11 +294,21 @@ BatchWriter::FileDetails* BatchWriter::PrepareFileForNextWrite(
     return &active_file;
   }
 
-  // If not, prepare to write a new file: start by figuring out what to name it
-  const auto next = GetFilenameForNextWrite(consent_dir_path, current_time);
+  // If we can't reuse the last file, prepare to write a new file. List the directory
+  // contents once here so that both the quota check and the filename search can share
+  // the same listing.
+  if (!CacheKnownFilenames(consent_dir_path)) {
+    active_file.Clear();
+    return nullptr;
+  }
+
+  // Enforce the directory size quota, using and updating the cached listing
+  PurgeDirectoryIfNeeded(consent_dir_path);
+
+  // Figure out what to name the new file, using the post-purge cached listing
+  const auto next = GetFilenameForNextWrite(current_time);
   if (!next) {
-    // Failed to resolve new filename (i.e. listing directory contents failed, or all
-    // potential filenames are in use); can't proceed with file creation
+    // All potential filenames are in use; can't proceed with file creation
     active_file.Clear();
     return nullptr;
   }
@@ -354,16 +370,11 @@ bool BatchWriter::CanReuseFileForNextWrite(
 }
 
 std::optional<std::pair<uint64_t, std::string>> BatchWriter::GetFilenameForNextWrite(
-    const StoragePath& consent_dir_path, Timestamp current_time
+    Timestamp current_time
 ) const {
-  // Our new file will use a filename that reflects the current timestamp, but it's
-  // possible that a file already exists with that name, and we don't want to reuse any
-  // existing files that we didn't create ourselves: so start by retrieving the list of
-  // existing filenames in our target directory
-  if (!CacheKnownFilenames(consent_dir_path)) {
-    // Failed to list directory contents; can't proceed with file creation
-    return std::nullopt;
-  }
+  // Our new file will use a filename reflecting the current timestamp. If a file with
+  // that name already exists (per the pre-populated _last_known_filenames), we
+  // increment by 1ms at a time to find a free slot, up to a limit of 100 attempts.
 
   // Use stack memory to convert uint64 to string for candidate filenames
   std::array<char, MAX_UINT64_DECIMAL_DIGITS> buffer{0};
@@ -427,9 +438,96 @@ bool BatchWriter::CacheKnownFilenames(const StoragePath& consent_dir_path) const
     return false;
   }
 
-  // Sort filenames for deterministic iteration in timestamp-name-order
+  // Sort filenames for deterministic iteration in timestamp-name-order, then strip any
+  // non-numeric names (files not written by the SDK) so that both
+  // PurgeDirectoryIfNeeded and GetFilenameForNextWrite operate only on batch files we
+  // own.
   std::sort(_last_known_filenames.begin(), _last_known_filenames.end());
+  _last_known_filenames.erase(
+      std::remove_if(
+          _last_known_filenames.begin(),
+          _last_known_filenames.end(),
+          [](const std::string& name) {
+            return !std::all_of(name.begin(), name.end(), [](unsigned char c) {
+              return std::isdigit(c);
+            });
+          }
+      ),
+      _last_known_filenames.end()
+  );
   return true;
 }
 
+void BatchWriter::PurgeDirectoryIfNeeded(const StoragePath& dir) {
+  // _last_known_filenames is pre-populated, sorted, and already filtered to
+  // numeric-named (SDK-written) batch files by CacheKnownFilenames.
+
+  // Accumulate sizes for each file using a reusable path buffer. Files whose size
+  // cannot be determined are treated as zero: they don't push the running total toward
+  // the quota threshold, but remain eligible for deletion if the purge loop runs and
+  // reaches them (without contributing to total reduction when deleted).
+  FilesystemWrapper fsw(_fs);
+  StoragePath path_buf;
+  std::vector<size_t> sizes;
+  sizes.reserve(_last_known_filenames.size());
+  size_t total = 0;
+  for (const std::string& name : _last_known_filenames) {
+    path_buf.MustSet(dir);
+    if (!path_buf.Append(name)) {
+      sizes.push_back(0);
+      continue;
+    }
+    const auto res = fsw.GetFileSize(path_buf.CStr());
+    const size_t file_size = res.value == FilesystemResult::OK ? res.size : 0;
+    sizes.push_back(file_size);
+    total += file_size;
+  }
+
+  // Note: this check covers only the files already on disk; it does not account for the
+  // new batch file that PrepareFileForNextWrite is about to create. The directory may
+  // therefore exceed max_directory_size by up to max_file_size after each rotation.
+  // This matches the iOS SDK's check-on-create design and the overshoot is bounded and
+  // transient, so it is accepted as a known limitation.
+  if (total <= _config.max_directory_size) {
+    return;
+  }
+
+  // Delete oldest files until we're within quota, erasing each from
+  // _last_known_filenames and the parallel sizes vector in lock-step so
+  // GetFilenameForNextWrite sees the post-purge state. Iterate by index rather than
+  // iterator so we can erase in-place: on a successful deletion we don't advance i (the
+  // next element slides into position); on failure we skip the file and increment.
+  size_t i = 0;
+  while (i < _last_known_filenames.size() && total > _config.max_directory_size) {
+    path_buf.MustSet(dir);
+    if (!path_buf.Append(_last_known_filenames[i])) {
+      ++i;
+      continue;
+    }
+    const auto delete_res = fsw.Delete(path_buf.CStr());
+    // DoesNotExist means the upload thread already removed this file concurrently;
+    // treat it as a successful removal so total and _last_known_filenames stay
+    // consistent and we don't over-delete younger files to compensate.
+    if (delete_res == FilesystemResult::OK ||
+        delete_res == FilesystemResult::DoesNotExist) {
+      total -= sizes[i];
+      if (delete_res == FilesystemResult::OK) {
+        _diagnostic_logger.Debug(
+            "Deleted batch file to enforce directory size quota",
+            {{"path", path_buf.Get()}}
+        );
+      }
+      const auto idx = static_cast<std::ptrdiff_t>(i);
+      _last_known_filenames.erase(_last_known_filenames.begin() + idx);
+      sizes.erase(sizes.begin() + idx);
+    } else {
+      _diagnostic_logger.Warning(
+          "Failed to delete batch file during directory quota enforcement",
+          {{"path", path_buf.Get()}, {"error", FilesystemResultStr(delete_res)}}
+      );
+      ++i;
+    }
+  }
+}
+
 }  // namespace datadog::impl
diff --git a/src/datadog/impl/core/storage_write.hpp b/src/datadog/impl/core/storage_write.hpp
index cf5d346b..cfceaab9 100644
--- a/src/datadog/impl/core/storage_write.hpp
+++ b/src/datadog/impl/core/storage_write.hpp
@@ -37,6 +37,18 @@ struct BatchWriterConfig {
    * bytes), we will not write to that file and will instead create a new one.
    */
   size_t max_file_size{0x400000};  // 4 MB
+  /**
+   * If the encoded size of a single event write (metadata + event, with TLV headers)
+   * exceeds this threshold, the event is dropped rather than written.
+   */
+  size_t max_event_size{512UL * 1024};  // 512 KB
+  /**
+   * Maximum total size (in bytes) of all batch files in a single consent-level
+   * directory (pending or granted) for one feature. When a new batch file is about to
+   * be created and the directory already exceeds this limit, the oldest files are
+   * deleted until the total is within the quota.
+   */
+  size_t max_directory_size{512ULL * 1024 * 1024};  // 512 MB
   /**
    * Maximum number of events that we will write to a single file. A single write
    * operation may include both metadata and event blocks: i.e. if configured with a
@@ -115,8 +127,9 @@ class BatchWriter {
 
   /**
    * Buffers the set of filenames retrieved in our last call to CacheKnownFilenames().
-   * Contains the names of all files present in the target directory for the current
-   * consent value. Reset when tracking consent changes.
+   * Contains the names of all SDK-written (numeric-named) batch files present in the
+   * target directory for the current consent value, sorted in ascending order. Reset
+   * when tracking consent changes.
    */
   mutable std::vector<std::string> _last_known_filenames;
 
@@ -240,19 +253,40 @@ class BatchWriter {
   ) const;
 
   /**
-   * Designates a name to use for a newly-created batch file in the given directory,
-   * returning both the integer millisecond count reflecting file creation time, as well
-   * as the string-formatted version of that same value.
+   * Designates a name for a newly-created batch file, returning both the integer
+   * millisecond count reflecting file creation time and the string-formatted value.
+   *
+   * Requires that `_last_known_filenames` has already been populated by
+   * `CacheKnownFilenames` before this function is called.
    */
   std::optional<std::pair<uint64_t, std::string>> GetFilenameForNextWrite(
-      const StoragePath& consent_dir_path, Timestamp current_time
+      Timestamp current_time
   ) const;
 
   /**
-   * Retrieves a list of all files in the current directory, caching that set of names
-   * in _last_known_filenames.
+   * Populates `_last_known_filenames` with a sorted listing of all numeric-named
+   * (SDK-written) regular files in `consent_dir_path`, excluding any foreign files
+   * whose names contain non-digit characters. Must be called in
+   * `PrepareFileForNextWrite` before invoking `PurgeDirectoryIfNeeded` or
+   * `GetFilenameForNextWrite`.
+   *
+   * Returns true on success, false if the directory listing fails.
    */
   bool CacheKnownFilenames(const StoragePath& consent_dir_path) const;
+
+  /**
+   * Enforces the per-directory size quota for `dir` by deleting the oldest batch files
+   * until the total size of remaining files falls within `_config.max_directory_size`.
+   *
+   * Reads and mutates `_last_known_filenames` (pre-populated and pre-filtered to
+   * numeric-named files by `CacheKnownFilenames`), removing deleted entries so that
+   * `GetFilenameForNextWrite` sees an accurate post-purge listing.
+   *
+   * Called only when a new batch file is about to be created, mirroring the iOS SDK's
+   * approach of deferring the quota check to file-creation time. The check is
+   * best-effort: errors from sizing individual files are tolerated.
+   */
+  void PurgeDirectoryIfNeeded(const StoragePath& dir);
 };
 
 }  // namespace datadog::impl
diff --git a/src/datadog/impl/core/upload_thread.cpp b/src/datadog/impl/core/upload_thread.cpp
index 1b6cce1d..6ca33529 100644
--- a/src/datadog/impl/core/upload_thread.cpp
+++ b/src/datadog/impl/core/upload_thread.cpp
@@ -11,6 +11,7 @@
 #include <charconv>
 #include <chrono>
 #include <iostream>
+#include <optional>
 #include <queue>
 #include <string>
 #include <thread>
@@ -109,8 +110,13 @@ static _process_and_upload_batch_result _process_and_upload_batch(
   const bool hold_advisory_lock = false;
   auto open_result = fsw.OpenForRead(file_path.CStr(), hold_advisory_lock);
   if (open_result.value != FilesystemResult::OK) {
-    // If we failed to open the file for any reason, leave it in place and continue to
-    // the next file
+    if (open_result.value == FilesystemResult::DoesNotExist) {
+      // The file was removed by another agent (quota purge or age eviction) between
+      // the directory listing and this open. Treat it as a consumed batch so the
+      // upload cycle continues rather than backing off.
+      return _process_and_upload_batch_result::bad_batch;
+    }
+    // For any other open failure, leave the file in place and retry later
     return _process_and_upload_batch_result::retryable_failure;
   }
 
@@ -165,6 +171,24 @@ static _process_and_upload_batch_result _process_and_upload_batch(
   return _interpret_http_result(res);
 }
 
+// Parses a batch filename as a millisecond-precision Unix timestamp and returns its age
+// relative to `now`. Returns nullopt for non-numeric filenames (non-SDK files that
+// should be left alone). Guards against underflow if the timestamp is in the future by
+// clamping the result to zero.
+static std::optional<Duration> batch_file_age(
+    const std::string& filename, Timestamp now
+) {
+  uint64_t timestamp_ms{0};
+  const auto parse_result =
+      std::from_chars(filename.data(), filename.data() + filename.size(), timestamp_ms);
+  if (parse_result.ec != std::errc{} ||
+      parse_result.ptr != filename.data() + filename.size()) {
+    return std::nullopt;
+  }
+  const Timestamp file_time{std::chrono::milliseconds(timestamp_ms)};
+  return file_time < now ? now - file_time : Duration::zero();
+}
+
 static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-complexity)
     DiagnosticLogger& diagnostic_logger,
     UploadThreadConfig config,
@@ -193,6 +217,52 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp
   // Prepare a FilesystemWrapper that will handle path encoding transparently
   FilesystemWrapper fsw(fs);
 
+  const Timestamp now = clock.Now();
+
+  // Evict stale files from the pending-consent directory. These accumulate when consent
+  // is set to Pending and never resolved. The upload thread never reads from this
+  // directory, so its only clean-up mechanism is this age-based eviction pass.
+  //
+  // This runs before the granted-directory listing so that a transient failure listing
+  // the granted directory does not prevent eviction of stale pending files.
+  const StoragePath& pending_dir_path = feature.storage->GetPendingPath();
+  mut_filenames.clear();
+  const auto pending_list_res = fsw.ListFiles(pending_dir_path.CStr(), mut_filenames);
+  if (pending_list_res == FilesystemResult::OK) {
+    std::sort(mut_filenames.begin(), mut_filenames.end());
+    for (const std::string& pending_filename : mut_filenames) {
+      const auto file_age = batch_file_age(pending_filename, now);
+      if (!file_age) {
+        continue;
+      }
+      if (*file_age < config.max_file_age_for_read) {
+        // Files are sorted oldest-first; all subsequent files are also too young.
+        break;
+      }
+
+      StoragePath pending_file_path;
+      pending_file_path.MustSet(pending_dir_path);
+      if (!pending_file_path.Append(pending_filename)) {
+        continue;
+      }
+
+      const auto delete_res = fsw.Delete(pending_file_path.CStr());
+      if (delete_res == FilesystemResult::OK) {
+        diagnostic_logger.Debug(
+            "Deleted outdated pending batch file",
+            {{"feature", feature.name}, {"filename", pending_filename}}
+        );
+      } else {
+        diagnostic_logger.Warning(
+            "Failed to delete outdated pending batch file",
+            {{"feature", feature.name},
+             {"filename", pending_filename},
+             {"error", FilesystemResultStr(delete_res)}}
+        );
+      }
+    }
+  }
+
   // Retrieve a list of all filenames in the consent-granted event directory
   mut_filenames.clear();
   const StoragePath& granted_dir_path = feature.storage->GetGrantedPath();
@@ -212,39 +282,17 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp
       _process_and_upload_batch_result::success
   };
   for (const std::string& filename : mut_filenames) {
-    // If a filename is all-integer, it's taken to be a Unix timestamp in milliseconds,
-    // corresponding to the system_clock value that the storage thread read when it
-    // created the file
-    uint64_t timestamp_ms{0};
-    const auto parse_result = std::from_chars(
-        filename.data(), filename.data() + filename.size(), timestamp_ms
-    );
-
-    // Skip any files whose names are not strictly numeric
-    const bool parse_ok = parse_result.ec == std::errc{};
-    const bool parsed_full_string =
-        parse_result.ptr == (filename.data() + filename.size());
-    if (!parse_ok || !parsed_full_string) {
+    const auto file_age = batch_file_age(filename, now);
+    if (!file_age) {
       continue;
     }
 
-    // Read the system clock so we can determine the relative age of the file
-    const Timestamp file_time{std::chrono::milliseconds(timestamp_ms)};
-    const Timestamp now = clock.Now();
-
-    // Defensive: guard against integer underflow on file age calculation. If the file
-    // appears to be from the future, clamp its age to 0.
-    Duration file_age = Duration::zero();
-    if (file_time < now) {
-      file_age = now - file_time;
-    }
-
     // Under normal circumstances, we have to respect a minimum file age threshold
     // before we can read from files, as the storage thread may still be writing to them
     // before they hit that age. If our threshold is 0, it means we're permitted to
     // bypass these checks and read from all files, as long as they're not too _old_ to
     // upload.
-    if (file_age < config.min_file_age_for_read) {
+    if (*file_age < config.min_file_age_for_read) {
       // If we've encountered our first file that's too new to process, we're done. We
       // process files in order, sorted lexically by timestamp, so every file that we'd
       // encounter hereafter would be even newer than this one.
@@ -262,7 +310,7 @@ static Duration _run_upload_cycle( // NOLINT(readability-function-cognitive-comp
     }
 
     // If this file is too old to process, delete it and continue
-    if (file_age >= config.max_file_age_for_read) {
+    if (*file_age >= config.max_file_age_for_read) {
       const auto delete_res = fsw.Delete(file_path.CStr());
       if (delete_res == FilesystemResult::OK) {
         diagnostic_logger.Debug(
diff --git a/tests/impl/core/storage_write_test.cpp b/tests/impl/core/storage_write_test.cpp
index 7eaf9096..8c99b07d 100644
--- a/tests/impl/core/storage_write_test.cpp
+++ b/tests/impl/core/storage_write_test.cpp
@@ -336,24 +336,6 @@ TEST_CASE("BatchWriter", "[unit]") {
     );
   }
 
-  SECTION("M drop event W single write exceeds max file size") {
-    // Given a batch writer configured with a max batch file size of 16 bytes
-    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
-    config.max_file_size = 16;
-    BatchWriter writer = init_writer(config);
-
-    // When we attempt to write a 20-byte payload, which is larger than any file
-    // could contain
-    const bool write_ok = writer.HandleWrite("twenty-bytes-0", {}, false);
-
-    // Then the event is not accepted
-    REQUIRE(write_ok == false);
-
-    // And nothing is written to disk
-    std::vector<std::string> names = fs.Ls(pending_dir_path);
-    REQUIRE(names.empty());
-  }
-
   SECTION("M start new file W max writes per file exceeded") {
     // Given a batch writer configured to write no more than 2 events per file
     auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
@@ -559,6 +541,44 @@ TEST_CASE("BatchWriter", "[unit]") {
     REQUIRE(fs.Ls(granted_dir_path).size() == 0);
   }
 
+  SECTION(
+      "M delete remaining pending files W upload thread has already evicted one file "
+      "before consent-revocation cleanup runs"
+  ) {
+    // Given a BatchWriter that has written two batch files to the pending directory
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Small);
+    BatchWriter writer = init_writer(config);
+    REQUIRE(writer.HandleWrite("event-0", {}, false));
+    clock.Tick(std::chrono::seconds(60));
+    REQUIRE(writer.HandleWrite("event-1", {}, false));
+
+    // Then we have two batch files in the pending directory
+    auto names = fs.Ls(pending_dir_path);
+    std::sort(names.begin(), names.end());
+    REQUIRE(names.size() == 2);
+
+    // When the upload thread concurrently evicts the oldest file between our ListFiles
+    // call and our Delete call: simulate this by making Delete return DoesNotExist for
+    // the oldest pending file
+    fs.SimulateFailure(
+        pending_prefix + names[0],
+        FilesystemResult::DoesNotExist,
+        MockFilesystem::FailureFlags::Delete
+    );
+
+    // And consent is revoked
+    const bool ok = writer.SetTrackingConsent(TrackingConsent::NotGranted);
+
+    // Then the revocation succeeds despite the race: the DoesNotExist is treated as
+    // "already gone" and deletion continues to the younger file
+    REQUIRE(ok);
+
+    // The younger file is deleted; the older file remains in the mock because
+    // SimulateFailure intercepted (but did not execute) its Delete call
+    names = fs.Ls(pending_dir_path);
+    REQUIRE(names.size() == 1);
+  }
+
   SECTION(
       "M move all files from pending to granted W tracking consent changes to Granted"
   ) {
@@ -787,6 +807,258 @@ TEST_CASE("BatchWriter", "[unit]") {
     REQUIRE(names.size() == 1);
   }
 
+  SECTION("M drop event W single write exceeds max_event_size but not max_file_size") {
+    // Given a batch writer with a generous max_file_size but a tight max_event_size
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_file_size = 1024 * 1024;  // 1 MB batch file — plenty of room
+    config.max_event_size = 20;          // only 20 bytes per event
+    BatchWriter writer = init_writer(config);
+
+    // When we attempt to write a payload where header (6 bytes) + data (15 bytes) = 21
+    REQUIRE_FALSE(writer.HandleWrite("fifteen-bytes!!", {}, false));
+
+    // Then nothing is written to disk
+    REQUIRE(fs.Ls(pending_dir_path).empty());
+  }
+
+  SECTION("M accept event W single write exactly equals max_event_size") {
+    // Given a batch writer with max_event_size set to exactly fit a 14-byte write
+    // (6-byte TLV header + 8 bytes data)
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_event_size = 14;
+    config.max_file_size = 1024 * 1024;
+    BatchWriter writer = init_writer(config);
+
+    // When we write an 8-byte event (total encoded size = 14 bytes)
+    REQUIRE(writer.HandleWrite("8-bytes!", {}, false));
+    REQUIRE(fs.Ls(pending_dir_path).size() == 1);
+  }
+
+  SECTION("M purge oldest files W directory size exceeds max_directory_size") {
+    // Given a batch writer with a directory quota of 50 bytes
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 50;
+    BatchWriter writer = init_writer(config);
+
+    // And three existing batch files each 40 bytes at timestamps 0, 1, and 2
+    // (total 120 bytes, well over the 50-byte quota)
+    fs.Touch(pending_prefix + "1700000000000", std::string(40, 'x'));
+    fs.Touch(pending_prefix + "1700000000001", std::string(40, 'y'));
+    fs.Touch(pending_prefix + "1700000000002", std::string(40, 'z'));
+
+    // When we write a new event 100ms later (triggers new file + quota check)
+    clock.TickMilliseconds(100);
+    REQUIRE(writer.HandleWrite("event-new", {}, false));
+
+    // Then the two oldest files are purged (oldest: 40 bytes → 80 > 50; next: 40
+    // bytes → 40 ≤ 50, stop), leaving only the youngest pre-existing file and the
+    // new batch file
+    std::vector<std::string> names = fs.Ls(pending_dir_path);
+    std::sort(names.begin(), names.end());
+    REQUIRE(names.size() == 2);
+    REQUIRE(names[0] == "1700000000002");
+    REQUIRE(names[1] == "1700000000100");
+  }
+
+  SECTION("M not purge any files W directory size is within max_directory_size") {
+    // Given a batch writer with a generous quota
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 10000;
+    BatchWriter writer = init_writer(config);
+
+    // And a small existing file in the pending directory
+    fs.Touch(pending_prefix + "1700000000000", std::string(10, 'x'));
+
+    // When we write a new event after triggering a file rotation
+    clock.TickMilliseconds(100);
+    REQUIRE(writer.HandleWrite("event-new", {}, false));
+
+    // Then no files are deleted
+    std::vector<std::string> names = fs.Ls(pending_dir_path);
+    std::sort(names.begin(), names.end());
+    REQUIRE(names.size() == 2);
+    REQUIRE(names[0] == "1700000000000");
+    REQUIRE(names[1] == "1700000000100");
+  }
+
+  SECTION("M not purge non-numeric-named files W directory contains foreign files") {
+    // Given a batch writer with a very small quota
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 1;
+    BatchWriter writer = init_writer(config);
+
+    // And a non-numerically-named file (not written by the SDK) that would push the
+    // directory over quota if counted
+    fs.Touch(pending_prefix + "not-a-batch", std::string(100, 'x'));
+
+    // When a new event triggers the quota check
+    clock.TickMilliseconds(100);
+    REQUIRE(writer.HandleWrite("event-new", {}, false));
+
+    // Then the non-SDK file is left untouched
+    REQUIRE(fs.IsFile(pending_prefix + "not-a-batch"));
+  }
+
+  SECTION(
+      "M leave existing files intact W directory listing fails on new file creation"
+  ) {
+    // Given a batch writer with a directory quota so tight that any existing file
+    // would trigger a purge
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 1;
+    BatchWriter writer = init_writer(config);
+
+    // And an existing batch file in the pending directory
+    fs.Touch(pending_prefix + "1700000000000", std::string(100, 'x'));
+
+    // And a filesystem that refuses to list the pending directory
+    fs.SimulateFailure(
+        pending_dir_path,
+        FilesystemResult::PermissionDenied,
+        MockFilesystem::FailureFlags::Ls
+    );
+
+    // When we attempt to write an event (no active file, so new-file path is taken and
+    // the quota check fires). The write also fails because filename resolution requires
+    // listing the same directory.
+    clock.TickMilliseconds(100);
+    writer.HandleWrite("event-new", {}, false);
+
+    // Then the pre-existing file is untouched — purge bailed out rather than
+    // deleting files based on an incomplete view of the directory
+    fs.ClearSimulatedFailure(pending_dir_path);
+    REQUIRE(fs.IsFile(pending_prefix + "1700000000000"));
+  }
+
+  SECTION(
+      "M drop event W metadata and event together exceed max_event_size but neither "
+      "alone would"
+  ) {
+    // Given a batch writer where max_event_size = 24 bytes.
+    // A 5-byte event alone encodes to 11 bytes (6-byte TLV header + 5 data), which is
+    // within the limit. An 8-byte metadata block encodes to 14 bytes. Combined they
+    // require 25 bytes, which exceeds the 24-byte cap.
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_file_size = 1024 * 1024;
+    config.max_event_size = 24;
+    BatchWriter writer = init_writer(config);
+
+    // When we write the 5-byte event with no metadata, it fits (11 bytes ≤ 24)
+    REQUIRE(writer.HandleWrite("event", {}, false));
+
+    // When we write the same event with an 8-byte metadata block, the combined size
+    // of 25 bytes exceeds the per-event cap
+    REQUIRE_FALSE(writer.HandleWrite("event", "metadata", false));
+
+    // Then only the first write produced a file; the second was dropped
+    REQUIRE(fs.Ls(pending_dir_path).size() == 1);
+  }
+
+  SECTION("M purge oldest files W quota exceeded in granted directory") {
+    // Given a batch writer writing to the granted directory with a 50-byte quota
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 50;
+    BatchWriter writer = init_writer(config, TrackingConsent::Granted);
+
+    // And three existing files of 40 bytes each (total 120 bytes, over the 50-byte
+    // quota)
+    fs.Touch(granted_prefix + "1700000000000", std::string(40, 'x'));
+    fs.Touch(granted_prefix + "1700000000001", std::string(40, 'y'));
+    fs.Touch(granted_prefix + "1700000000002", std::string(40, 'z'));
+
+    // When we write a new event 100ms later (triggers new file + quota check)
+    clock.TickMilliseconds(100);
+    REQUIRE(writer.HandleWrite("event-new", {}, false));
+
+    // Then the two oldest files are purged, leaving only the youngest pre-existing
+    // file and the new batch file
+    std::vector<std::string> names = fs.Ls(granted_dir_path);
+    std::sort(names.begin(), names.end());
+    REQUIRE(names.size() == 2);
+    REQUIRE(names[0] == "1700000000002");
+    REQUIRE(names[1] == "1700000000100");
+  }
+
+  SECTION("M treat GetFileSize failure as zero W file stat fails during quota check") {
+    // Given a batch writer with a 50-byte quota
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 50;
+    BatchWriter writer = init_writer(config);
+
+    // And an oldest file that is 40 bytes but has a simulated stat failure, so it
+    // contributes 0 bytes to the quota total
+    fs.Touch(pending_prefix + "1700000000000", std::string(40, 'x'));
+    fs.SimulateFailure(
+        pending_prefix + "1700000000000",
+        FilesystemResult::UnknownError,
+        MockFilesystem::FailureFlags::Stat
+    );
+
+    // And two younger files of 10 bytes each (true total 60 bytes, but apparent total
+    // with the stat failure counted as 0 is only 20 bytes — within the 50-byte quota)
+    fs.Touch(pending_prefix + "1700000000001", std::string(10, 'y'));
+    fs.Touch(pending_prefix + "1700000000002", std::string(10, 'z'));
+
+    // When we write a new event 100ms later (triggers new file + quota check)
+    clock.TickMilliseconds(100);
+    REQUIRE(writer.HandleWrite("event-new", {}, false));
+
+    // Then no files are purged: the failed stat caused the directory to appear to be
+    // within quota, so the purge was a no-op
+    REQUIRE(fs.IsFile(pending_prefix + "1700000000000"));
+    REQUIRE(fs.IsFile(pending_prefix + "1700000000001"));
+    REQUIRE(fs.IsFile(pending_prefix + "1700000000002"));
+  }
+
+  SECTION(
+      "M not over-delete younger files W upload thread removes oldest file between "
+      "stat pass and delete pass"
+  ) {
+    // Given a batch writer with a directory quota of 50 bytes
+    auto config = BatchWriterConfig::FromBatchSize(BatchSize::Large);
+    config.max_directory_size = 50;
+    BatchWriter writer = init_writer(config);
+
+    // And three existing batch files at t=0, t=1, t=2, each 40 bytes — total 120 bytes,
+    // well over the 50-byte quota
+    fs.Touch(pending_prefix + "1700000000000", std::string(40, 'x'));
+    fs.Touch(pending_prefix + "1700000000001", std::string(40, 'y'));
+    fs.Touch(pending_prefix + "1700000000002", std::string(40, 'z'));
+
+    // The upload thread concurrently removes the oldest file between the stat pass and
+    // the deletion pass: simulate by making Delete return DoesNotExist for t=0 while
+    // leaving GetFileSize intact so the stat pass still sees its size (40 bytes).
+    //
+    // Note: the mock retains the file on disk even though Delete returns DoesNotExist;
+    // this mirrors the real scenario where the file is already gone externally but the
+    // mock cannot remove it via the simulated return-code path.
+    fs.SimulateFailure(
+        pending_prefix + "1700000000000",
+        FilesystemResult::DoesNotExist,
+        MockFilesystem::FailureFlags::Delete
+    );
+
+    // When a new write triggers the quota check
+    clock.TickMilliseconds(100);
+    REQUIRE(writer.HandleWrite("event-new", {}, false));
+
+    // Then the DoesNotExist result for t=0 is treated as a successful removal: its 40
+    // bytes are subtracted, leaving 80 bytes > 50. The loop then removes t=1 (40
+    // bytes), leaving 40 bytes ≤ 50 and stopping before t=2. Three files remain on
+    // disk: t=0 (physically present in mock despite DoesNotExist), t=2 (untouched),
+    // and the new batch file at t=100.
+    //
+    // Without the fix the DoesNotExist would be treated as a failure, keeping 120 bytes
+    // in total; the loop would then delete t=1 and t=2, leaving only t=0 and the new
+    // file — one more deletion than necessary.
+    std::vector<std::string> names = fs.Ls(pending_dir_path);
+    std::sort(names.begin(), names.end());
+    REQUIRE(names.size() == 3);
+    REQUIRE(names[0] == "1700000000000");
+    REQUIRE(names[1] == "1700000000002");
+    REQUIRE(names[2] == "1700000000100");
+  }
+
   SECTION("M write bypass event to granted dir W active file is warm in pending dir") {
     // Given a batch writer with consent Pending and a prior normal write that has
     // warmed up _active_file to a file in the pending directory
diff --git a/tests/impl/core/upload_thread_test.cpp b/tests/impl/core/upload_thread_test.cpp
index 0b46130e..c03694f2 100644
--- a/tests/impl/core/upload_thread_test.cpp
+++ b/tests/impl/core/upload_thread_test.cpp
@@ -693,6 +693,71 @@ TEST_CASE("HandleUploadProc", "[unit]") {
     REQUIRE(delay_until_next_cycle == std::chrono::milliseconds(55000));
   }
 
+  SECTION(
+      "M skip batch and continue processing W batch file is gone when opened"
+      " {race between quota purge and upload thread}"
+  ) {
+    // Given a single registered feature with three eligible batches, the second of
+    // which disappears between the directory listing and the open (simulating a
+    // concurrent quota purge)
+    clock.FreezeAtMilliseconds(1700000000000);
+    MockHttpClient client;
+    fs.Touch(
+        alpha_granted_prefix + "1699999955000",
+        MockTLVFile().AppendEvent("event-0").ToString()
+    );
+    fs.Touch(
+        alpha_granted_prefix + "1699999956000",
+        MockTLVFile().AppendEvent("event-1").ToString()
+    );
+    fs.Touch(
+        alpha_granted_prefix + "1699999957000",
+        MockTLVFile().AppendEvent("event-2").ToString()
+    );
+    const auto [core_config, config] = init_config(
+        BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium
+    );
+    auto features = register_feature(UploadFrequency::Average);
+
+    // And a filesystem that will report DoesNotExist when the second file is opened
+    // (simulating the file being removed by the quota purge after listing)
+    fs.SimulateFailure(
+        alpha_granted_prefix + "1699999956000",
+        FilesystemResult::DoesNotExist,
+        MockFilesystem::FailureFlags::Open
+    );
+
+    // When we process uploads for that feature
+    std::vector<std::string> filenames;
+    std::vector<char> read_buffer;
+    Duration delay_until_next_cycle = Internal_HandleUploadProc(
+        logger,
+        config,
+        clock,
+        CreateFeatureId("ALFA"),
+        features,
+        fs,
+        client,
+        filenames,
+        read_buffer
+    );
+
+    // Then our first and third batches are uploaded (the missing second is skipped)
+    REQUIRE(alpha->reports.size() == 2);
+    REQUIRE(client.requests.size() == 2);
+    REQUIRE(client.requests[0].body == "event-0");
+    REQUIRE(client.requests[1].body == "event-2");
+
+    // And the upload cycle did not abort early: all three files are gone from disk
+    // (the first and third were deleted after successful upload; the second was deleted
+    // by the attempted fsw.Delete after bad_batch, since only its Open was blocked)
+    REQUIRE(fs.Ls(alpha_granted_dir).empty());
+
+    // And this feature's upload delay is reduced because the cycle finished with a
+    // successful upload
+    REQUIRE(delay_until_next_cycle == std::chrono::seconds(10));
+  }
+
   SECTION("M reject batch and continue processing W unable to process batch") {
     // Given a single registered feature with three eligible batches, the second of
     // which does not contain valid TLV data
@@ -832,6 +897,200 @@ TEST_CASE("HandleUploadProc", "[unit]") {
     REQUIRE(delay_until_next_cycle == std::chrono::milliseconds(55000));
   }
 
+  SECTION("M delete outdated files W pending directory contains stale batch files") {
+    // Given an upload cycle configuration with an 18-hour TTL
+    clock.FreezeAtMilliseconds(1700000000000);
+    MockHttpClient client;
+    const auto [core_config, config] = init_config(
+        BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium
+    );
+    auto features = register_feature(UploadFrequency::Average);
+
+    // And a pending directory containing one file that is over 18 hours old and one
+    // that is fresh
+    const std::string alpha_pending_dir =
+        "app/.datadog/main/12345/alpha/intermediate-v1";
+    const std::string alpha_pending_prefix = alpha_pending_dir + "/";
+    const uint64_t eighteen_hours_ms = 18ULL * 60 * 60 * 1000;
+    const uint64_t old_ts = 1700000000000ULL - eighteen_hours_ms - 1000;
+    const uint64_t fresh_ts = 1700000000000ULL - 1000;
+    fs.Touch(
+        alpha_pending_prefix + std::to_string(old_ts),
+        MockTLVFile().AppendEvent("old-event").ToString()
+    );
+    fs.Touch(
+        alpha_pending_prefix + std::to_string(fresh_ts),
+        MockTLVFile().AppendEvent("fresh-event").ToString()
+    );
+
+    // When we run an upload cycle
+    std::vector<std::string> filenames;
+    std::vector<char> read_buffer;
+    Internal_HandleUploadProc(
+        logger,
+        config,
+        clock,
+        CreateFeatureId("ALFA"),
+        features,
+        fs,
+        client,
+        filenames,
+        read_buffer
+    );
+
+    // Then the stale pending file is deleted
+    REQUIRE_FALSE(fs.IsFile(alpha_pending_prefix + std::to_string(old_ts)));
+
+    // And the fresh pending file is left intact
+    REQUIRE(fs.IsFile(alpha_pending_prefix + std::to_string(fresh_ts)));
+
+    // And no upload requests were made (pending files are never uploaded)
+    REQUIRE(client.requests.empty());
+  }
+
+  SECTION(
+      "M leave pending files intact W they are younger than max_file_age_for_read"
+  ) {
+    // Given an upload cycle configuration with an 18-hour TTL
+    clock.FreezeAtMilliseconds(1700000000000);
+    MockHttpClient client;
+    const auto [core_config, config] = init_config(
+        BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium
+    );
+    auto features = register_feature(UploadFrequency::Average);
+
+    // And a pending directory with two files both younger than 18 hours
+    const std::string alpha_pending_dir =
+        "app/.datadog/main/12345/alpha/intermediate-v1";
+    const std::string alpha_pending_prefix = alpha_pending_dir + "/";
+    fs.Touch(
+        alpha_pending_prefix + "1699999000000",
+        MockTLVFile().AppendEvent("event-a").ToString()
+    );
+    fs.Touch(
+        alpha_pending_prefix + "1699999500000",
+        MockTLVFile().AppendEvent("event-b").ToString()
+    );
+
+    // When we run an upload cycle
+    std::vector<std::string> filenames;
+    std::vector<char> read_buffer;
+    Internal_HandleUploadProc(
+        logger,
+        config,
+        clock,
+        CreateFeatureId("ALFA"),
+        features,
+        fs,
+        client,
+        filenames,
+        read_buffer
+    );
+
+    // Then both pending files remain untouched
+    REQUIRE(fs.Ls(alpha_pending_dir).size() == 2);
+    REQUIRE(client.requests.empty());
+  }
+
+  SECTION("M not delete non-numeric files W pending directory contains non-SDK files") {
+    // Given an upload cycle with an 18-hour TTL
+    clock.FreezeAtMilliseconds(1700000000000);
+    MockHttpClient client;
+    const auto [core_config, config] = init_config(
+        BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium
+    );
+    auto features = register_feature(UploadFrequency::Average);
+
+    const std::string alpha_pending_dir =
+        "app/.datadog/main/12345/alpha/intermediate-v1";
+    const std::string alpha_pending_prefix = alpha_pending_dir + "/";
+
+    // And a pending directory containing one stale SDK batch file and one
+    // non-numerically-named file that was not written by the SDK
+    const uint64_t eighteen_hours_ms = 18ULL * 60 * 60 * 1000;
+    const uint64_t old_ts = 1700000000000ULL - eighteen_hours_ms - 1000;
+    fs.Touch(
+        alpha_pending_prefix + std::to_string(old_ts),
+        MockTLVFile().AppendEvent("old-event").ToString()
+    );
+    fs.Touch(alpha_pending_prefix + "not-an-sdk-file", "foreign-data");
+
+    // When we run an upload cycle
+    std::vector<std::string> filenames;
+    std::vector<char> read_buffer;
+    Internal_HandleUploadProc(
+        logger,
+        config,
+        clock,
+        CreateFeatureId("ALFA"),
+        features,
+        fs,
+        client,
+        filenames,
+        read_buffer
+    );
+
+    // Then the stale SDK batch file is evicted
+    REQUIRE_FALSE(fs.IsFile(alpha_pending_prefix + std::to_string(old_ts)));
+
+    // But the non-SDK file is left untouched
+    REQUIRE(fs.IsFile(alpha_pending_prefix + "not-an-sdk-file"));
+
+    // And no upload requests were made
+    REQUIRE(client.requests.empty());
+  }
+
+  SECTION("M silently skip pending eviction W ListFiles fails on pending directory") {
+    // Given an upload cycle with an 18-hour TTL
+    clock.FreezeAtMilliseconds(1700000000000);
+    MockHttpClient client;
+    const auto [core_config, config] = init_config(
+        BatchSize::Medium, UploadFrequency::Average, BatchProcessingLevel::Medium
+    );
+    auto features = register_feature(UploadFrequency::Average);
+
+    const std::string alpha_pending_dir =
+        "app/.datadog/main/12345/alpha/intermediate-v1";
+    const std::string alpha_pending_prefix = alpha_pending_dir + "/";
+
+    // And a stale pending file that would normally be evicted
+    const uint64_t eighteen_hours_ms = 18ULL * 60 * 60 * 1000;
+    const uint64_t old_ts = 1700000000000ULL - eighteen_hours_ms - 1000;
+    fs.Touch(
+        alpha_pending_prefix + std::to_string(old_ts),
+        MockTLVFile().AppendEvent("old-event").ToString()
+    );
+
+    // But the pending directory refuses to be listed
+    fs.SimulateFailure(
+        alpha_pending_dir,
+        FilesystemResult::PermissionDenied,
+        MockFilesystem::FailureFlags::Ls
+    );
+
+    // When we run an upload cycle
+    std::vector<std::string> filenames;
+    std::vector<char> read_buffer;
+    Internal_HandleUploadProc(
+        logger,
+        config,
+        clock,
+        CreateFeatureId("ALFA"),
+        features,
+        fs,
+        client,
+        filenames,
+        read_buffer
+    );
+
+    // Then the pending eviction pass is silently skipped: the stale file remains
+    fs.ClearSimulatedFailure(alpha_pending_dir);
+    REQUIRE(fs.IsFile(alpha_pending_prefix + std::to_string(old_ts)));
+
+    // And no upload requests were made
+    REQUIRE(client.requests.empty());
+  }
+
   SECTION("M reject batch and continue processing W upload gets HTTP client error") {
     // Given a single registered feature with two eligible batches that can be read
     // and processed without error
diff --git a/tests/mock/filesystem.cpp b/tests/mock/filesystem.cpp
index a7736cf0..556d2fe0 100644
--- a/tests/mock/filesystem.cpp
+++ b/tests/mock/filesystem.cpp
@@ -485,6 +485,20 @@ FilesystemResult MockFilesystem::Close(PlatformFileHandle handle) {
   return FilesystemResult::OK;
 }
 
+IFilesystem::GetFileSizeResult MockFilesystem::GetFileSize(const PlatformPath& path) {
+  const std::string normalized_path = NormalizePath(path);
+  std::lock_guard lock(_mutex);
+
+  const auto found = _files.find(normalized_path);
+  if (found == _files.end()) {
+    return {FilesystemResult::DoesNotExist, 0};
+  }
+  if (auto status = HasSimulatedFailure(found->second, FailureFlags::Stat)) {
+    return {*status, 0};
+  }
+  return {FilesystemResult::OK, found->second.data.size()};
+}
+
 FilesystemResult MockFilesystem::Delete(const PlatformPath& path) {
   const std::string normalized_path = NormalizePath(path);
   std::lock_guard lock(_mutex);
@@ -931,7 +945,7 @@ bool MockFilesystem::IsFileLocked(std::string_view path) {
   if (file != _files.end()) {
     return file->second.advisory_lock_holder != INVALID_FILE_HANDLE;
   }
-  return "";
+  return false;
 }
 
 std::string MockFilesystem::Cat(std::string_view path) {
diff --git a/tests/mock/filesystem.hpp b/tests/mock/filesystem.hpp
index 7795b39a..11183494 100644
--- a/tests/mock/filesystem.hpp
+++ b/tests/mock/filesystem.hpp
@@ -32,6 +32,7 @@ class MockFilesystem : public impl::IFilesystem {
    * at (or, in some cases, one level beneath) the given path.
    */
   enum class FailureFlags : uint8_t {
+    Stat = (1 << 0),    // GetFileSize will fail for the target file
     Mkdir = (1 << 1),   // CreateDirectory will fail with target dir as parent
     Ls = (1 << 2),      // ListFiles and ListSubdirectories will fail in target dir
     Open = (1 << 3),    // OpenForRead and OpenForWrite will fail on file; in target dir
@@ -131,6 +132,7 @@ class MockFilesystem : public impl::IFilesystem {
   ) override;
   ReadResult Read(impl::PlatformFileHandle handle, char* dst, size_t n) override;
   impl::FilesystemResult Close(impl::PlatformFileHandle handle) override;
+  GetFileSizeResult GetFileSize(const impl::PlatformPath& path) override;
   impl::FilesystemResult Delete(const impl::PlatformPath& path) override;
   impl::FilesystemResult DeleteDirectory(const impl::PlatformPath& path) override;
   impl::FilesystemResult Rename(