diff --git a/lib_eio/process.ml b/lib_eio/process.ml
index 11049eff..63ec52ee 100644
--- a/lib_eio/process.ml
+++ b/lib_eio/process.ml
@@ -61,12 +61,12 @@ module Pi = struct
 
   module type MGR = sig
     type tag
-    type t
+    type t 
 
     val pipe :
       t ->
       sw:Switch.t ->
-      [< `Flow_source | `Resource_close] r * [< `Flow_sink | `Resource_close] r
+      [< Flow.source_ty | Resource.close_ty] r * [< Flow.sink_ty | Resource.close_ty] r
 
     val spawn :
       t ->
@@ -141,7 +141,8 @@ let run t ?cwd ?stdin ?stdout ?stderr ?(is_success = Int.equal 0) ?env ?executab
     let ex = err (Child_error status) in
     raise (Exn.add_context ex "running command: %a" pp_args args)
 
-let pipe (type tag) ~sw ((Resource.T (v, ops)) : [> tag mgr_ty] r) =
+let pipe (type tag) ~sw ((Resource.T (v, ops)) : [> tag mgr_ty] r) :
+  ([< Flow.source_ty | Resource.close_ty ] r * [< Flow.sink_ty | Resource.close_ty ] r) =
   let module X = (val (Resource.get ops Pi.Mgr)) in
   X.pipe v ~sw
 
diff --git a/lib_eio/process.mli b/lib_eio/process.mli
index 84a0bddf..2f13ebc6 100644
--- a/lib_eio/process.mli
+++ b/lib_eio/process.mli
@@ -140,10 +140,10 @@ 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 7969163a..061db060 100644
--- a/lib_eio/unix/eio_unix.mli
+++ b/lib_eio/unix/eio_unix.mli
@@ -58,9 +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 ->
-  [< Flow.source_ty | Resource.close_ty] r * [< Flow.sink_ty | Resource.close_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. *)
@@ -99,7 +97,7 @@ module Private : sig
     | Await_readable : Unix.file_descr -> unit Effect.t      (** See {!await_readable} *)
     | Await_writable : Unix.file_descr -> unit Effect.t      (** See {!await_writable} *)
     | Get_monotonic_clock : Eio.Time.Mono.ty r Effect.t
-    | Pipe : Eio.Switch.t -> (source_ty r * sink_ty r) Effect.t    (** See {!pipe} *)
+    | Pipe : Eio.Switch.t -> ([< source_ty] r * [< sink_ty] r) Effect.t   
 
   module Rcfd = Rcfd
 
diff --git a/lib_eio/unix/private.ml b/lib_eio/unix/private.ml
index 3e7c81f7..ddaa2fbc 100644
--- a/lib_eio/unix/private.ml
+++ b/lib_eio/unix/private.ml
@@ -7,7 +7,7 @@ type _ Effect.t +=
   | Await_readable : Unix.file_descr -> unit Effect.t
   | Await_writable : Unix.file_descr -> unit Effect.t
   | Get_monotonic_clock : Eio.Time.Mono.ty r Effect.t
-  | Pipe : Switch.t -> ([< Flow.source_ty | Resource.close_ty] r * [< Flow.sink_ty | Resource.close_ty] r) Effect.t
+  | Pipe : Switch.t -> ([< source_ty] r * [< sink_ty] r) Effect.t
 
 let await_readable fd = Effect.perform (Await_readable fd)
 let await_writable fd = Effect.perform (Await_writable fd)
diff --git a/lib_eio/unix/process.ml b/lib_eio/unix/process.ml
index f1a504ca..84c89973 100644
--- a/lib_eio/unix/process.ml
+++ b/lib_eio/unix/process.ml
@@ -118,9 +118,8 @@ end) = struct
   type tag = [ `Generic | `Unix ]
 
   let pipe _ ~sw =
-  (Eio.Process.pipe ~sw :> ([< Eio.Resource.close_ty | Eio.Flow.source_ty] r *
-  [< Eio.Resource.close_ty | Eio.Flow.sink_ty] r))
-
+    (Private.pipe sw :> ([< Eio.Resource.close_ty | Eio.Flow.source_ty] r *
+    [< Eio.Resource.close_ty | Eio.Flow.sink_ty] r))
 
   let spawn v ~sw ?cwd ?stdin ?stdout ?stderr ?env ?executable args =
     let executable = get_executable executable ~args in
diff --git a/lib_eio_linux/eio_linux.ml b/lib_eio_linux/eio_linux.ml
index ec9c8eb0..63598028 100644
--- a/lib_eio_linux/eio_linux.ml
+++ b/lib_eio_linux/eio_linux.ml
@@ -537,8 +537,8 @@ let run_event_loop (type a) ?fallback config (main : _ -> a) arg : a =
       | Eio_unix.Private.Pipe sw -> Some (fun k ->
           match
             let r, w = Low_level.pipe ~sw in
-            let r = (Flow.of_fd r :> _ Eio_unix.source) in
-            let w = (Flow.of_fd w :> _ Eio_unix.sink) in
+            let r = (Flow.of_fd r :> [< Eio_unix.source_ty] Eio.Std.r) in
+            let w = (Flow.of_fd w :> [< Eio_unix.sink_ty] Eio.Std.r) in
             (r, w)
           with
           | r -> continue k r