diff --git a/runtime/bin/file_system_watcher_macos.cc b/runtime/bin/file_system_watcher_macos.cc
index 50c4f9aca78..ee8c84c5766 100644
--- a/runtime/bin/file_system_watcher_macos.cc
+++ b/runtime/bin/file_system_watcher_macos.cc
@@ -36,15 +36,75 @@ union FSEvent {
   uint8_t bytes[PATH_MAX + 8];
 };
 
+// A helper for creating Dart_CObject array using a single allocation.
+//
+// We can't use CObject helpers because those rely on Dart_ScopeAllocate.
+namespace {
+
+template <typename T>
+struct SetCObjectValue;
+
+template <>
+struct SetCObjectValue<const char*> {
+  static void Assign(Dart_CObject* o, const char* str) {
+    o->type = Dart_CObject_kString;
+    o->value.as_string = str;
+  }
+};
+
+template <>
+struct SetCObjectValue<char*> {
+  static void Assign(Dart_CObject* o, const char* str) {
+    o->type = Dart_CObject_kString;
+    o->value.as_string = str;
+  }
+};
+
+template <>
+struct SetCObjectValue<int64_t> {
+  static void Assign(Dart_CObject* o, int64_t v) {
+    o->type = Dart_CObject_kInt64;
+    o->value.as_int64 = v;
+  }
+};
+
+template <typename... Ts>
+Dart_CObject* CreateCObjectArray(Ts... elements) {
+  const auto length = sizeof...(elements);
+  auto array = static_cast<Dart_CObject*>(
+      malloc(sizeof(Dart_CObject) +
+             (sizeof(Dart_CObject*) + sizeof(Dart_CObject)) * length));
+  array->type = Dart_CObject_kArray;
+  array->value.as_array.values = reinterpret_cast<Dart_CObject**>(array + 1);
+  array->value.as_array.length = length;
+  for (uintptr_t i = 0; i < length; i++) {
+    array->value.as_array.values[i] =
+        reinterpret_cast<Dart_CObject*>(array->value.as_array.values + length) +
+        i;
+  }
+
+  int index = 0;
+  (
+      [&] {
+        SetCObjectValue<Ts>::Assign(array->value.as_array.values[index],
+                                    elements);
+        ++index;
+      }(),
+      ...);
+
+  return array;
+}
+
+}  // namespace
+
 class Node {
  public:
-  Node(char* base_path, int read_fd, int write_fd, bool recursive)
-      : base_path_length_(strlen(base_path)),
+  Node(Dart_Port port, char* base_path, bool recursive)
+      : port_(port),
+        base_path_length_(strlen(base_path)),
         path_ref_(CFStringCreateWithCString(nullptr,
                                             base_path,
                                             kCFStringEncodingUTF8)),
-        read_fd_(read_fd),
-        write_fd_(write_fd),
         recursive_(recursive),
         ref_(nullptr) {
     Start();
@@ -56,7 +116,6 @@ class Node {
     // deallocated, the same [FSEventStream] that [Callback] gets a reference
     // to during its execution. [Callback] holding a reference prevents stream
     // from deallocation.
-    close(write_fd_);
     CFRelease(path_ref_);
   }
 
@@ -96,20 +155,16 @@ class Node {
   }
 
   intptr_t base_path_length() const { return base_path_length_; }
-  int read_fd() const { return read_fd_; }
-  int write_fd() const { return write_fd_; }
   bool recursive() const { return recursive_; }
 
-  static Node* Watch(const char* path, int events, bool recursive) {
-    int fds[2];
-    VOID_NO_RETRY_EXPECTED(pipe(fds));
-    FDUtils::SetNonBlocking(fds[0]);
-    FDUtils::SetBlocking(fds[1]);
-
+  static Node* Watch(Dart_Port port,
+                     const char* path,
+                     int events,
+                     bool recursive) {
     char base_path[PATH_MAX];
     realpath(path, base_path);
 
-    return new Node(base_path, fds[0], fds[1], recursive);
+    return new Node(port, base_path, recursive);
   }
 
   static void Unwatch(Node* node) {
@@ -147,11 +202,16 @@ class Node {
       TimerUtils::Sleep(1000 /* ms */);
     }
     Node* node = static_cast<Node*>(client);
+
+    // Can't use CObject helpers because they expect Dart_ScopeAllocate to work
+    // and this thread is not attached to any isolate or native message handler.
+    Dart_CObject events;
+    events.type = Dart_CObject_kArray;
+    events.value.as_array.values =
+        static_cast<Dart_CObject**>(malloc(sizeof(Dart_CObject*) * num_events));
+    events.value.as_array.length = 0;
     for (size_t i = 0; i < num_events; i++) {
       char* path = reinterpret_cast<char**>(event_paths)[i];
-      FSEvent event;
-      event.data.exists =
-          File::GetType(nullptr, path, false) != File::kDoesNotExist;
       path += node->base_path_length();
       // If path is longer the base, skip next character ('/').
       if (path[0] != '\0') {
@@ -160,18 +220,72 @@ class Node {
       if (!node->recursive() && (strstr(path, "/") != nullptr)) {
         continue;
       }
-      event.data.flags = event_flags[i];
-      memmove(event.data.path, path, strlen(path) + 1);
-      write(node->write_fd(), event.bytes, sizeof(event));
+
+      const bool is_path_empty = path[0] == '\0';
+      const bool path_exists =
+          !is_path_empty &&
+          File::GetType(nullptr, path, false) != File::kDoesNotExist;
+
+      events.value.as_array.values[events.value.as_array.length++] =
+          CreateCObjectArray(
+              /*flags=*/ConvertEventFlags(event_flags[i], is_path_empty,
+                                          path_exists),
+              /*cookie=*/static_cast<int64_t>(0), path,
+              /*path_id=*/reinterpret_cast<int64_t>(node));
     }
+
+    if (events.value.as_array.length != 0) {
+      Dart_PostCObject(node->port_, &events);
+    }
+
+    for (int i = 0; i < events.value.as_array.length; i++) {
+      free(events.value.as_array.values[i]);
+    }
+    free(events.value.as_array.values);
+  }
+
+  static int64_t ConvertEventFlags(FSEventStreamEventFlags flags,
+                                   bool is_path_empty,
+                                   bool path_exists) {
+    int64_t mask = 0;
+    if ((flags & kFSEventStreamEventFlagItemRenamed) != 0) {
+      if (is_path_empty) {
+        // The moved path is the path being watched.
+        mask |= FileSystemWatcher::kDeleteSelf;
+      } else if (path_exists) {
+        mask |= FileSystemWatcher::kCreate;
+      } else {
+        mask |= FileSystemWatcher::kDelete;
+      }
+    }
+    if ((flags & kFSEventStreamEventFlagItemModified) != 0) {
+      mask |= FileSystemWatcher::kModifyContent;
+    }
+    if ((flags & kFSEventStreamEventFlagItemXattrMod) != 0) {
+      mask |= FileSystemWatcher::kModifyAttribute;
+    }
+    if ((flags & kFSEventStreamEventFlagItemCreated) != 0) {
+      mask |= FileSystemWatcher::kCreate;
+    }
+    if ((flags & kFSEventStreamEventFlagItemIsDir) != 0) {
+      mask |= FileSystemWatcher::kIsDir;
+    }
+    if ((flags & kFSEventStreamEventFlagItemRemoved) != 0) {
+      if (is_path_empty) {
+        // The removed path is the path being watched.
+        mask |= FileSystemWatcher::kDeleteSelf;
+      } else {
+        mask |= FileSystemWatcher::kDelete;
+      }
+    }
+    return mask;
   }
 
   static dispatch_queue_t notification_queue_;
 
+  Dart_Port port_;
   intptr_t base_path_length_;
   CFStringRef path_ref_;
-  int read_fd_;
-  int write_fd_;
   bool recursive_;
   FSEventStreamRef ref_;
   Monitor monitor_;
@@ -203,7 +317,8 @@ intptr_t FileSystemWatcher::WatchPath(intptr_t id,
                                       const char* path,
                                       int events,
                                       bool recursive) {
-  return reinterpret_cast<intptr_t>(Node::Watch(path, events, recursive));
+  return reinterpret_cast<intptr_t>(
+      Node::Watch(static_cast<Dart_Port>(id), path, events, recursive));
 }
 
 void FileSystemWatcher::UnwatchPath(intptr_t id, intptr_t path_id) {
@@ -216,67 +331,13 @@ void FileSystemWatcher::DestroyWatch(intptr_t path_id) {
 }
 
 intptr_t FileSystemWatcher::GetSocketId(intptr_t id, intptr_t path_id) {
-  return reinterpret_cast<Node*>(path_id)->read_fd();
+  // This API should not be called. We are communicating over ports instead.
+  return -1;
 }
 
 Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) {
-  intptr_t fd = GetSocketId(id, path_id);
-  intptr_t avail = FDUtils::AvailableBytes(fd);
-  int count = avail / sizeof(FSEvent);
-  if (count <= 0) {
-    return Dart_NewList(0);
-  }
-  Dart_Handle events = Dart_NewList(count);
-  FSEvent e;
-  for (int i = 0; i < count; i++) {
-    intptr_t bytes = TEMP_FAILURE_RETRY(read(fd, e.bytes, sizeof(e)));
-    if (bytes < 0) {
-      return DartUtils::NewDartOSError();
-    }
-    size_t path_len = strlen(e.data.path);
-    Dart_Handle event = Dart_NewList(kEventNumElements);
-    int flags = e.data.flags;
-    int mask = 0;
-    if ((flags & kFSEventStreamEventFlagItemRenamed) != 0) {
-      if (path_len == 0) {
-        // The moved path is the path being watched.
-        mask |= kDeleteSelf;
-      } else {
-        mask |= e.data.exists ? kCreate : kDelete;
-      }
-    }
-    if ((flags & kFSEventStreamEventFlagItemModified) != 0) {
-      mask |= kModifyContent;
-    }
-    if ((flags & kFSEventStreamEventFlagItemXattrMod) != 0) {
-      mask |= kModifyAttribute;
-    }
-    if ((flags & kFSEventStreamEventFlagItemCreated) != 0) {
-      mask |= kCreate;
-    }
-    if ((flags & kFSEventStreamEventFlagItemIsDir) != 0) {
-      mask |= kIsDir;
-    }
-    if ((flags & kFSEventStreamEventFlagItemRemoved) != 0) {
-      if (path_len == 0) {
-        // The removed path is the path being watched.
-        mask |= kDeleteSelf;
-      } else {
-        mask |= kDelete;
-      }
-    }
-    Dart_ListSetAt(event, kEventFlagsIndex, Dart_NewInteger(mask));
-    Dart_ListSetAt(event, kEventCookieIndex, Dart_NewInteger(0));
-    Dart_Handle name = Dart_NewStringFromUTF8(
-        reinterpret_cast<uint8_t*>(e.data.path), path_len);
-    if (Dart_IsError(name)) {
-      return name;
-    }
-    Dart_ListSetAt(event, kEventPathIndex, name);
-    Dart_ListSetAt(event, kEventPathIdIndex, Dart_NewInteger(path_id));
-    Dart_ListSetAt(events, i, event);
-  }
-  return events;
+  // This API should not be called. We are communicating over ports instead.
+  return DartUtils::NewDartOSError();
 }
 
 }  // namespace bin
diff --git a/runtime/tests/vm/dart/regress_61551_test.dart b/runtime/tests/vm/dart/regress_61551_test.dart
new file mode 100644
index 00000000000..f86b6180d9a
--- /dev/null
+++ b/runtime/tests/vm/dart/regress_61551_test.dart
@@ -0,0 +1,99 @@
+// Copyright (c) 2025, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+//
+// Make sure that file system watcher works when Mac OS X restricts pipe buffer
+// size to 512 bytes (see https://github.com/dart-lang/sdk/issues/61551).
+
+import 'dart:async';
+import 'dart:io';
+import 'dart:ffi';
+
+import 'package:expect/expect.dart';
+import 'package:ffi/ffi.dart';
+import 'package:path/path.dart' as p;
+
+void main() async {
+  if (!Platform.isMacOS) {
+    return;
+  }
+
+  exhaustMaxPipeKVA();
+
+  var tempDir = await Directory.systemTemp.createTemp('fsevents_test_');
+  try {
+    var watcher = tempDir.watch();
+    var eventsReceived = 0;
+
+    var subscription = watcher.listen((event) {
+      eventsReceived++;
+    });
+
+    // Wait for the watcher to become active.
+    await Future.delayed(Duration(milliseconds: 500));
+
+    // Make some changes in the directory.
+    final testFile = File(p.join(tempDir.path, 'test_file.txt'));
+    await testFile.writeAsString('test content');
+    await testFile.writeAsString('modified content', mode: FileMode.append);
+    await testFile.delete();
+
+    // Wait a bit for events to arrive.
+    await Future.delayed(Duration(seconds: 2));
+
+    // Cancel the watcher.
+    await subscription.cancel();
+
+    // We should have received at least some events.
+    Expect.isTrue(eventsReceived > 0);
+  } finally {
+    await tempDir.delete(recursive: true);
+  }
+}
+
+// Create pipes and force their buffers to grow to 64KB until we reach
+// kern.ipc.maxpipekva. This should not take more than 256 pipes because
+// the limit is 16 MB.
+void exhaustMaxPipeKVA() {
+  final fds = calloc<Int>(2);
+
+  const writeSize = 64 * 1024;
+  final buf = calloc<Int8>(writeSize);
+
+  for (int i = 0; i < 256; i++) {
+    final pipeRes = pipe(fds);
+    // We do not expect to run out of file descriptors before we run out of space
+    // in kernel for pipebuffers.
+    Expect.equals(0, pipeRes, 'Failed to create a pipe');
+    var writeFd = (fds + 1).value;
+
+    const F_GETFL = 3;
+    const F_SETFL = 4;
+    const O_NONBLOCK = 0x00000004;
+
+    final flags = fcntl0(writeFd, F_GETFL);
+    Expect.isTrue(flags != -1, 'Failed to call fcntl($writeFd, F_GETFL)');
+
+    final setFlagsRes = fcntl1(writeFd, F_SETFL, flags | O_NONBLOCK);
+    Expect.isTrue(setFlagsRes != -1, 'Failed to call fcntl($writeFd, F_SETFL)');
+
+    if (write(writeFd, buf.cast(), writeSize) != writeSize) {
+      break;
+    }
+  }
+}
+
+@Native<Int Function(Pointer<Int>)>()
+external int pipe(Pointer<Int> fds);
+
+@Native<Int Function(Pointer<Int>)>()
+external int fnctl(Pointer<Int> fds);
+
+@Native<Int Function(Int, Int, VarArgs<()>)>(symbol: 'fcntl')
+external int fcntl0(int fd, int cmd);
+
+@Native<Int Function(Int, Int, VarArgs<(Int,)>)>(symbol: 'fcntl')
+external int fcntl1(int fd, int cmd, int val);
+
+@Native<Int Function(Int, Pointer<Void> buf, Size)>()
+external int write(int fd, Pointer<Void> buf, int size);
diff --git a/sdk/lib/_internal/vm/bin/file_patch.dart b/sdk/lib/_internal/vm/bin/file_patch.dart
index 271b2953f33..2df5886baed 100644
--- a/sdk/lib/_internal/vm/bin/file_patch.dart
+++ b/sdk/lib/_internal/vm/bin/file_patch.dart
@@ -450,11 +450,11 @@ abstract class _FileSystemWatcher {
       }
 
       if (Platform.isWindows) {
-        return _SocketPerPathFileSystemWatcher();
+        return _Win32FileSystemWatcher();
       }
 
       if (Platform.isMacOS) {
-        return _SocketPerPathFileSystemWatcher();
+        return _FSEventStreamFileSystemWatcher();
       }
     }
 
@@ -497,16 +497,54 @@ abstract class _FileSystemWatcher {
   external static void _destroyWatch(ffi.Pointer<ffi.Void> pathId);
 }
 
-class _InotifyFileSystemWatcher extends _FileSystemWatcher {
+/// A watcher that receives events for multiple `pathId` on a single channel.
+abstract class _MultiplexingFileSystemWatcher extends _FileSystemWatcher {
+  /// Map of [_watchedPath] indexed by `pathId` values.
   final Map<int, _WatchedPath> _watchedPaths = <int, _WatchedPath>{};
 
+  /// Perform necessary initialization of the native state for the watcher.
+  void _ensureWatcherIsRunning();
+
+  /// Shutdown the watcher when there is no actively watched paths.
+  ///
+  /// If shutdown requires asynchronous actions returns [Future] which will
+  /// complete when shutdown is finished.
+  FutureOr<void> _stopWatcher();
+
+  @override
+  _WatchedPath _startWatching(String path, int events, bool recursive) {
+    _ensureWatcherIsRunning();
+    final pathId = super._watchPath(path, events, recursive);
+    // On Linux inotify_add_watch will return an existing watch descriptor
+    // for the inode if there is already one associated with it. Thus we
+    // need accept the possibility that calling _watchPath twice will
+    // return the same pathId. Other OSes do not reuse pathId values.
+    assert(Platform.isLinux || !_watchedPaths.containsKey(pathId));
+    return _watchedPaths[pathId] ??= _WatchedPath(pathId, path, events);
+  }
+
+  @override
+  Future<void> _stopWatching(_WatchedPath wp) async {
+    assert(_watchedPaths[wp.pathId] == wp);
+    _watchedPaths.remove(wp.pathId);
+    await super._stopWatching(wp);
+
+    // If there are no more active watcher close inotify descriptor.
+    if (_watchedPaths.isEmpty) {
+      await _stopWatcher();
+    }
+  }
+}
+
+class _InotifyFileSystemWatcher extends _MultiplexingFileSystemWatcher {
   int? _inotifyFd;
   StreamSubscription<List<_NativeFSEvent>>? _inotifySubscription;
 
   @override
   int get _watcherId => _inotifyFd!;
 
-  void _ensureInotifyFD() {
+  @override
+  void _ensureWatcherIsRunning() {
     if (_inotifyFd != null) {
       return;
     }
@@ -518,22 +556,6 @@ class _InotifyFileSystemWatcher extends _FileSystemWatcher {
     ).listen(_handleEvents);
   }
 
-  @override
-  _WatchedPath _startWatching(String path, int events, bool recursive) {
-    _ensureInotifyFD();
-    // On Linux inotify_add_watch will return an existing watch descriptor
-    // for the inode if there is already one associated with it. Thus we
-    // need accept the possibility that calling _watchPath twice will
-    // return the same pathId.
-    final pathId = super._watchPath(path, events, recursive);
-    final watchedPath = _watchedPaths[pathId] ??= _WatchedPath(
-      pathId,
-      path,
-      events,
-    );
-    return watchedPath;
-  }
-
   void _handleEvents(List<_NativeFSEvent> events) {
     Set<_WatchedPath>? dirty;
 
@@ -556,22 +578,46 @@ class _InotifyFileSystemWatcher extends _FileSystemWatcher {
   }
 
   @override
-  Future<void> _stopWatching(_WatchedPath wp) async {
-    assert(_watchedPaths[wp.pathId] == wp);
-    _watchedPaths.remove(wp.pathId);
-    await super._stopWatching(wp);
-
-    // If there are no more active watcher close inotify descriptor.
+  FutureOr<void> _stopWatcher() {
     final subscription = _inotifySubscription;
-    if (_watchedPaths.isEmpty && subscription != null) {
-      _inotifyFd = null;
-      _inotifySubscription = null;
-      return subscription.cancel();
+    _inotifyFd = null;
+    _inotifySubscription = null;
+    return subscription?.cancel();
+  }
+}
+
+class _FSEventStreamFileSystemWatcher extends _MultiplexingFileSystemWatcher {
+  final _port = RawReceivePort()..keepIsolateAlive = false;
+
+  @override
+  late final _watcherId = ffi.NativePort(_port.sendPort).nativePort;
+
+  @override
+  void _ensureWatcherIsRunning() {
+    _port.keepIsolateAlive = true;
+    _port.handler = _handleEvents;
+  }
+
+  @override
+  FutureOr<void> _stopWatcher() {
+    _port.keepIsolateAlive = false;
+    _port.handler = null;
+  }
+
+  void _handleEvents(List events) {
+    // All events in a bundle have the same pathId, and we never get an empty
+    // bundle.
+    final pathId = (events[0] as _NativeFSEvent).pathId;
+    if (_watchedPaths[pathId] case final watchedPath?) {
+      for (_NativeFSEvent event in events) {
+        watchedPath.addEvent(event);
+      }
+      watchedPath.flushUnmatchedMoves();
     }
   }
 }
 
-class _SocketPerPathFileSystemWatcher extends _FileSystemWatcher {
+class _Win32FileSystemWatcher extends _FileSystemWatcher {
   @override
   _WatchedPath _startWatching(String path, int events, bool recursive) {
     final watchedPath = _WatchedPath(