diff --git a/lix/libutil/async-io.cc b/lix/libutil/async-io.cc
index 22d943437..7488c3a62 100644
--- a/lix/libutil/async-io.cc
+++ b/lix/libutil/async-io.cc
@@ -1,4 +1,11 @@
 #include "async-io.hh"
+#include "async.hh"
+#include "error.hh"
+#include "file-descriptor.hh"
+#include "result.hh"
+#include <cerrno>
+#include <exception>
+#include <fcntl.h>
 
 namespace nix {
 kj::Promise<Result<void>> AsyncInputStream::drainInto(Sink & sink)
@@ -133,4 +140,54 @@ try {
 } catch (...) {
     co_return result::current_exception();
 }
+
+AsyncFdIoStream::AsyncFdIoStream(AutoCloseFD fd) : AsyncFdIoStream(shared_fd{}, fd.get())
+{
+    ownedFd = std::move(fd);
+}
+
+AsyncFdIoStream::AsyncFdIoStream(shared_fd, int fd)
+    : fd(fd)
+    , observer(AIO().unixEventPort, fd, kj::UnixEventPort::FdObserver::OBSERVE_READ_WRITE)
+{
+    oldFlags = fcntl(fd, F_GETFL, 0);
+    if (oldFlags == -1 || fcntl(fd, F_SETFL, oldFlags | O_NONBLOCK)) {
+        throw SysError("making file descriptor non-blocking");
+    }
+}
+
+AsyncFdIoStream::~AsyncFdIoStream() noexcept(false)
+{
+    if (fcntl(fd, F_SETFL, oldFlags)) {
+        try {
+            throw SysError("restoring file descriptor flags");
+        } catch (...) {
+            ignoreExceptionInDestructor();
+        }
+    }
+}
+
+kj::Promise<Result<size_t>> AsyncFdIoStream::read(void * tgt, size_t size)
+{
+    auto got = ::read(fd, tgt, size);
+    if (got >= 0) {
+        return {result::success(got)};
+    } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+        return observer.whenBecomesReadable().then([=, this] { return read(tgt, size); });
+    } else {
+        return {result::failure(std::make_exception_ptr(SysError(errno, "read failed")))};
+    }
+}
+
+kj::Promise<Result<size_t>> AsyncFdIoStream::write(const void * src, size_t size)
+{
+    auto got = ::write(fd, src, size);
+    if (got >= 0) {
+        return {result::success(got)};
+    } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+        return observer.whenBecomesWritable().then([=, this] { return write(src, size); });
+    } else {
+        return {result::failure(std::make_exception_ptr(SysError(errno, "write failed")))};
+    }
+}
 }
diff --git a/lix/libutil/async-io.hh b/lix/libutil/async-io.hh
index d71b043b0..680307ca5 100644
--- a/lix/libutil/async-io.hh
+++ b/lix/libutil/async-io.hh
@@ -3,10 +3,13 @@
 
 #include "lix/libutil/async.hh"
 #include "lix/libutil/box_ptr.hh"
+#include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/io-buffer.hh"
 #include "lix/libutil/ref.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/serialise.hh"
+#include <kj/async-io.h>
+#include <kj/async-unix.h>
 #include <kj/async.h>
 #include <kj/common.h>
 #include <memory>
@@ -152,4 +155,31 @@ public:
     kj::Promise<Result<size_t>> write(const void * src, size_t size) override;
     kj::Promise<Result<void>> flush();
 };
+
+class AsyncStream : public AsyncInputStream, public AsyncOutputStream
+{};
+
+class AsyncFdIoStream : public AsyncStream
+{
+    int fd, oldFlags;
+    AutoCloseFD ownedFd; // only for closing automatically, must equal fd if set
+    kj::UnixEventPort::FdObserver observer;
+
+public:
+    struct shared_fd
+    {};
+
+    explicit AsyncFdIoStream(AutoCloseFD fd);
+    AsyncFdIoStream(shared_fd, int fd);
+
+    ~AsyncFdIoStream() noexcept(false);
+
+    int getFD() const
+    {
+        return fd;
+    }
+
+    kj::Promise<Result<size_t>> read(void * tgt, size_t size) override;
+    kj::Promise<Result<size_t>> write(const void * src, size_t size) override;
+};
 }