diff --git a/lib_eio/process.ml b/lib_eio/process.ml
index 587db13a..0cca7157 100644
--- a/lib_eio/process.ml
+++ b/lib_eio/process.ml
@@ -143,7 +143,10 @@ let run t ?cwd ?stdin ?stdout ?stderr ?(is_success = Int.equal 0) ?env ?executab
 
 let pipe (type tag) ~sw ((Resource.T (v, ops)) : [> tag mgr_ty] r) =
   let module X = (val (Resource.get ops Pi.Mgr)) in
-  X.pipe v ~sw
+  let r, w = X.pipe v ~sw in
+  let r = (r : [Flow.source_ty | Resource.close_ty] r :> [< Flow.source_ty | Resource.close_ty] r) in
+  let w = (w : [Flow.sink_ty   | Resource.close_ty] r :> [< Flow.sink_ty   | Resource.close_ty] r) in
+  r, w
 
 let parse_out (type tag) (t : [> tag mgr_ty] r) parse ?cwd ?stdin ?stderr ?is_success ?env ?executable args =
   Switch.run ~name:"Process.parse_out" @@ fun sw ->
diff --git a/lib_eio/process.mli b/lib_eio/process.mli
index 5966bbd8..fb19ac64 100644
--- a/lib_eio/process.mli
+++ b/lib_eio/process.mli
@@ -140,7 +140,7 @@ val parse_out :
 
 (** {2 Pipes} *)
 
-val pipe : sw:Switch.t -> _ mgr -> [Flow.source_ty | Resource.close_ty] r * [Flow.sink_ty | Resource.close_ty] r
+val pipe : sw:Switch.t -> _ mgr -> [< Flow.source_ty | Resource.close_ty] r * [< Flow.sink_ty | Resource.close_ty] r
 (** [pipe ~sw mgr] creates a pipe backed by the OS.
 
     The flows can be used by {!spawn} without the need for extra fibers to copy the data.
diff --git a/lib_eio/unix/eio_unix.mli b/lib_eio/unix/eio_unix.mli
index 774ba5f6..1b94254c 100644
--- a/lib_eio/unix/eio_unix.mli
+++ b/lib_eio/unix/eio_unix.mli
@@ -58,7 +58,7 @@ val run_in_systhread : ?label:string -> (unit -> 'a) -> 'a
 
     @param label The operation name to use in trace output. *)
 
-val pipe : Switch.t -> source_ty r * sink_ty r
+val pipe : Switch.t -> [< source_ty] r * [< sink_ty] r
 (** [pipe sw] returns a connected pair of flows [src] and [sink]. Data written to [sink]
     can be read from [src].
     Note that, like all FDs created by Eio, they are both marked as close-on-exec by default. *)