diff --git a/shell-conduit.cabal b/shell-conduit.cabal
--- a/shell-conduit.cabal
+++ b/shell-conduit.cabal
@@ -1,5 +1,5 @@
 name:                shell-conduit
-version:             3.0
+version:             4.0
 synopsis:            Write shell scripts with Conduit
 description:         Write shell scripts with Conduit. See "Data.Conduit.Shell" for documentation.
 license:             BSD3
@@ -20,7 +20,8 @@
                      Data.Conduit.Shell.Process
                      Data.Conduit.Shell.Types
                      Data.Conduit.Shell.Variadic
-  build-depends:     base >= 4 && <5
+  build-depends:     async >= 2.0.1.5
+                   , base >= 4 && <5
                    , bytestring
                    , conduit
                    , conduit-extra
@@ -31,10 +32,11 @@
                    , monads-tf
                    , process
                    , resourcet
+                   , semigroups
                    , split
                    , template-haskell
-                   , transformers
-                   , transformers-base
                    , text
-                   , semigroups
                    , these
+                   , transformers
+                   , transformers-base
+                   , unix >= 2.7.0.1
diff --git a/src/Data/Conduit/Shell.hs b/src/Data/Conduit/Shell.hs
--- a/src/Data/Conduit/Shell.hs
+++ b/src/Data/Conduit/Shell.hs
@@ -15,35 +15,43 @@
 -- The monad instance of Conduit will simply pass along all stdout
 -- results:
 --
--- >>> run (do echo "Hello"; sed "s/l/a/"; echo "OK!")
--- Hello
--- OK!
---
 -- Piping with Conduit's normal pipe will predictably pipe things
 -- together, as in Bash:
 --
--- >>> run (do shell "echo Hello" $= sed "s/l/a/"; echo "OK!")
+-- >>> run (do shell "echo Hello" $| sed "s/l/a/"; echo "OK!")
 -- Healo
 -- OK!
 --
 -- Streaming pipes (aka lazy pipes) is also possible:
 --
--- >>> run (do tail' "foo.txt" "-f" $= grep "--line-buffered" "Hello")
+-- >>> run (tail' "/tmp/foo.txt" "-f" $| grep "--line-buffered" "Hello")
 -- Hello, world!
 -- Oh, hello!
 --
 -- (Remember that @grep@ needs @--line-buffered@ if it is to output
 -- things line-by-line).
 --
+-- Run custom processes via the @proc@ function:
+--
+-- >>> run (proc "ls" [])
+-- dist  LICENSE  README.md  Setup.hs  shell-conduit.cabal  src  TAGS  TODO.org
+--
+-- Run shell commands via the @shell@ function:
+--
+-- >>> run (shell "ls")
+-- dist  LICENSE  README.md  Setup.hs  shell-conduit.cabal  src  TAGS  TODO.org
+--
+-- Run conduits via the @conduit@ function:
+--
+-- >>> run (cat "/tmp/foo.txt" $| conduit (do Just x <- await; yield x))
+-- Hello!
+--
 -- == How it works
 --
 -- All executable names in the @PATH@ at compile-time are brought into
 -- scope as runnable process conduits e.g. @ls@ or @grep@.
 --
--- Stdin/out and stderr are handled as an 'Either' type: 'Chunk'
 --
--- 'Left' is stderr, 'Right' is @stdin@/@stdout@.
---
 -- All processes are bound as variadic process calling functions, like this:
 --
 -- @
@@ -54,9 +62,9 @@
 -- But ultimately the types end up being:
 --
 -- @
--- rmdir "foo" :: Conduit Chunk m Chunk
--- ls :: Conduit Chunk m Chunk
--- ls "." :: Conduit Chunk m Chunk
+-- rmdir "foo" :: Segment ()
+-- ls :: Segment ()
+-- ls "." :: Segment ()
 -- @
 --
 -- Etc.
@@ -64,8 +72,7 @@
 -- Run all shell scripts with 'run':
 --
 -- @
--- run :: (MonadIO m, MonadBaseControl IO m)
---     => Conduit Chunk (ShellT m) Chunk -> m ()
+-- run :: Segment r -> IO r
 -- @
 --
 -- == String types
@@ -86,19 +93,16 @@
 module Data.Conduit.Shell
   (-- * Running scripts
    run
-   -- * Running custom processes
+   -- * Making segments
   ,shell
   ,proc
-  -- * I/O chunks
-  ,bytes
-  ,unbytes
-  ,withRights
-  ,redirect
-  ,quiet
-  ,writeChunks
-  ,discardChunks
-   -- * Re-exports
-   -- $exports
+  ,conduit
+  -- * Composition of segments
+  ,($|)
+  ,Segment
+  ,ProcessException(..)
+  -- * Re-exports
+  -- $exports
   ,module Data.Conduit.Shell.PATH
   ,module Data.Conduit.Shell.Types
   ,module Data.Conduit.Shell.Variadic
diff --git a/src/Data/Conduit/Shell/Process.hs b/src/Data/Conduit/Shell/Process.hs
--- a/src/Data/Conduit/Shell/Process.hs
+++ b/src/Data/Conduit/Shell/Process.hs
@@ -1,5 +1,7 @@
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE DeriveDataTypeable #-}
 
 -- | Reading from the process.
 
@@ -7,238 +9,243 @@
   (-- * Running scripts
    run
    -- * Running processes
+  ,conduit
   ,Data.Conduit.Shell.Process.shell
   ,Data.Conduit.Shell.Process.proc
-   -- * I/O chunks
-  ,bytes
-  ,unbytes
-  ,withRights
-  ,redirect
-  ,quiet
-  ,writeChunks
-  ,discardChunks
-  -- * Low-level internals
-  ,conduitProcess
+  ,($|)
+  ,Segment
+  ,ProcessException(..)
+  ,ToChunk(..)
   )
   where
 
-import           Data.Conduit.Shell.Types
-
 import           Control.Applicative
-import           Control.Concurrent (forkIO)
-import           Control.Concurrent.Chan
-import qualified Control.Exception as E
+import           Control.Concurrent.Async
+import           Control.Exception
 import           Control.Monad
-import           Control.Monad.Fix
-import           Control.Monad.Trans
-import           Control.Monad.Trans.Resource
+import           Control.Monad.IO.Class
 import           Data.ByteString (ByteString)
 import qualified Data.ByteString as S
 import           Data.Conduit
-import           Data.Conduit.List (sourceList)
+import           Data.Conduit.Binary
 import qualified Data.Conduit.List as CL
-import           Data.Conduit.Process
-import           Data.Semigroup
-import           Data.These
-import           System.Exit (ExitCode(..))
+import           Data.Typeable
+import           System.Exit
 import           System.IO
-import qualified System.Process
+import           System.Posix.IO (createPipe, fdToHandle)
+import           System.Process
+import           System.Process.Internals (createProcess_)
 
--- | Extract the stdout values from the stream, discarding any errors.
-bytes :: Monad m => Conduit Chunk m ByteString
-bytes = CL.mapMaybe (either (const Nothing) Just)
+-- | A pipeable segment. Either a conduit or a process.
+data Segment r
+  = SegmentConduit (ConduitM ByteString (Either ByteString ByteString) IO r)
+  | SegmentProcess (Handles -> IO r)
 
--- | Extract the stdout values from the stream, discarding any errors.
-unbytes :: Monad m => Conduit ByteString m Chunk
-unbytes = CL.map Right
+instance Monad Segment where
+  return = SegmentConduit . return
+  SegmentConduit c >>= f =
+    SegmentProcess (conduitToProcess c) >>=
+    f
+  SegmentProcess f >>= g =
+    SegmentProcess
+      (\handles ->
+         do x <- f handles
+            case g x of
+              SegmentConduit c ->
+                conduitToProcess c handles
+              SegmentProcess p -> p handles)
 
--- | Run a shell command.
-shell :: (MonadResource m) => String -> Conduit Chunk m Chunk
-shell = conduitProcess . System.Process.shell
+instance Functor Segment where
+  fmap = liftM
 
--- | Run a shell command.
-proc :: (MonadResource m) => String -> [String] -> Conduit Chunk m Chunk
-proc px args = conduitProcess (System.Process.proc px args)
+instance Applicative Segment where
+  (<*>) = ap; pure = return
 
--- | Size of buffer used to read from process.
-bufSize :: Int
-bufSize = 64 * 1024
+instance MonadIO Segment where
+  liftIO x = SegmentProcess (const x)
 
--- | Do something with just the rights.
-withRights :: (Monad m)
-           => Conduit ByteString m ByteString -> Conduit Chunk m Chunk
-withRights f =
-  getZipConduit
-    (ZipConduit f' *>
-     ZipConduit g')
-  where f' =
-          CL.mapMaybe (either (const Nothing) Just) =$=
-          f =$=
-          CL.map Right
-        g' = CL.filter isLeft
+-- | Process handles: @stdin@, @stdout@, @stderr@
+data Handles =
+  Handles Handle Handle Handle
 
--- | Redirect the given chunk type to the other type.
-redirect :: Monad m
-         => ChunkType -> Conduit Chunk m Chunk
-redirect ty =
-  CL.map (\c' ->
-            case c' of
-              Left x' ->
-                case ty of
-                  Stderr -> Right x'
-                  Stdout -> c'
-              Right x' ->
-                case ty of
-                  Stderr -> c'
-                  Stdout -> Left x')
+-- | Process running exception.
+data ProcessException =
+  ProcessException CreateProcess
+                   ExitCode
+  deriving (Typeable)
 
--- | Discard any output from the command: make it quiet.
-quiet :: (Monad m,MonadIO m) => Conduit Chunk m Chunk -> Conduit Chunk m Chunk
-quiet m = m $= discardChunks
+instance Exception ProcessException
 
--- | Run a shell scripting conduit.
-run :: (MonadIO m,MonadBaseControl IO m)
-    => Conduit Chunk (ShellT m) Chunk -> m ()
-run p =
-  runResourceT
-    (runShellT (sourceList [] $=
-                p $$
-                writeChunks))
+instance Show ProcessException where
+  show (ProcessException cp ec) =
+    concat ["The "
+           ,case cmdspec cp of
+              ShellCommand s -> "shell command " ++ show s
+              RawCommand f args ->
+                "raw command: " ++
+                show (f : args)
+           ," returned a failure exit code: "
+           ,case ec of
+              ExitFailure i -> show i
+              _ -> show ec]
 
--- | Write chunks to stdout and stderr.
-writeChunks :: (MonadIO m)
-            => Consumer Chunk m ()
-writeChunks =
-  awaitForever
-    (\c ->
-       case c of
-         Left e -> liftIO (S.hPut stderr e)
-         Right o -> liftIO (S.hPut stdout o))
+-- | Convert a process or a conduit to a segment.
+class ToSegment a where
+  type SegmentResult a
+  toSegment :: a -> Segment (SegmentResult a)
 
--- | Discard all chunks.
-discardChunks :: (MonadIO m)
-              => Consumer Chunk m ()
-discardChunks = awaitForever (const (return ()))
+instance ToSegment (Segment r) where
+  type SegmentResult (Segment r) = r
+  toSegment = id
 
--- | Conduit of process.
-conduitProcess :: (MonadResource m)
-               => CreateProcess -> Conduit Chunk m Chunk
-conduitProcess cp =
-  bracketP createp closep startProxy
-  where createp =
-          createProcess
-            cp {std_in = CreatePipe
-               ,std_out = CreatePipe
-               ,std_err = CreatePipe}
-        closep (Just cin,Just cout,Just cerr,ph) =
-          do hClose cin
-             hClose cout
-             hClose cerr
-             _ <- waitForProcess' ph
-             return ()
-        closep _ = error "conduitProcess: unexpected arguments to closep"
+instance (a ~ ByteString,ToChunk b,m ~ IO) => ToSegment (ConduitM a b m r) where
+  type SegmentResult (ConduitM a b m r) = r
+  toSegment f =
+    SegmentConduit (f `fuseUpstream` CL.map toChunk)
 
--- | Start proxying from conduit to process back to conduit.
-startProxy :: (MonadIO m,MonadThrow m)
-           => (Maybe Handle,Maybe Handle,Maybe Handle,ProcessHandle)
-           -> ConduitM Chunk Chunk m ()
-startProxy (Just cin,Just cout,Just cerr,ph) = interleave
-  where interleave =
-          do end <- proxyInterleaved
-             liftIO (hClose cin)
-             remainders cout cerr
-             ec <- liftIO (maybe (waitForProcess' ph) return end)
+instance ToSegment CreateProcess where
+  type SegmentResult CreateProcess = ()
+  toSegment = liftProcess
+
+-- | Used to allow outputting stdout or stderr.
+class ToChunk a where
+  toChunk :: a -> Either ByteString ByteString
+
+instance ToChunk ByteString where
+  toChunk = Left
+
+instance ToChunk (Either ByteString ByteString) where
+  toChunk = id
+
+-- | Run a shell command.
+shell :: String -> Segment ()
+shell = liftProcess . System.Process.shell
+
+-- | Run a process command.
+proc :: String -> [String] -> Segment ()
+proc name args = liftProcess (System.Process.proc name args)
+
+-- | Run a segment.
+run :: Segment r -> IO r
+run (SegmentConduit c) =
+  run (SegmentProcess (conduitToProcess c))
+run (SegmentProcess p) =
+  p (Handles stdin stdout stderr)
+
+-- | Fuse two segments (either processes or conduits).
+($|) :: Segment () -> Segment b -> Segment b
+x $| y = x `fuseSegment` y
+infixl 0 $|
+
+-- | Lift a conduit into a segment.
+conduit :: (a ~ ByteString,ToChunk b,m ~ IO) => ConduitM a b m r -> Segment r
+conduit f = SegmentConduit (f `fuseUpstream` CL.map toChunk)
+
+-- | Lift a process into a segment.
+liftProcess :: CreateProcess -> Segment ()
+liftProcess cp =
+  SegmentProcess
+    (\(Handles inh outh errh) ->
+       let config =
+             cp {std_in = UseHandle inh
+                ,std_out = UseHandle outh
+                ,std_err = UseHandle errh
+                ,close_fds = True}
+       in do (Nothing,Nothing,Nothing,ph) <- createProcess_ "liftProcess" config
+             ec <- waitForProcess ph
              case ec of
                ExitSuccess -> return ()
-               ExitFailure i ->
-                 monadThrow (ShellExitFailure i)
-        proxyInterleaved =
-          do proxy cout Right
-             proxy cerr Left
-             ended <- liftIO (getProcessExitCode ph)
-             case ended of
-               Just{} -> return ended
-               Nothing ->
-                 do minp <- await
-                    case minp of
-                      Nothing -> return Nothing
-                      Just chunk ->
-                        do case chunk of
-                             Left{} -> yield chunk
-                             Right bytes ->
-                               liftIO (do S.hPut cin bytes
-                                          hFlush cin)
-                           proxyInterleaved
-startProxy _ = error "startProxy: unexpected arguments"
+               _ ->
+                 throwIO (ProcessException cp ec))
 
--- | Concurrently yield eithers downstream.
-remainders :: MonadIO m
-           => Handle -> Handle -> ConduitM i (Either ByteString ByteString) m ()
-remainders cout cerr =
-  do chan <- liftIO newChan
-     void (liftIO (forkIO (remainder chan cout Right)))
-     void (liftIO (forkIO (remainder chan cerr Left)))
-     fix (\loop done ->
-            case done of
-              Just (These () ()) -> return ()
-              _ ->
-                do chunk <- liftIO (readChan chan)
-                   case chunk of
-                     Left mchunk ->
-                       case mchunk of
-                         Nothing ->
-                           loop (done <>
-                                 Just (This ()))
-                         Just chunk ->
-                           do yield (Left chunk)
-                              loop done
-                     Right mchunk ->
-                       case mchunk of
-                         Nothing ->
-                           loop (done <>
-                                 Just (That ()))
-                         Just chunk ->
-                           do yield (Right chunk)
-                              loop done)
-         (Nothing :: Maybe (These () ()))
+-- | Convert a conduit to a process.
+conduitToProcess :: ConduitM ByteString (Either ByteString ByteString) IO r
+                 -> (Handles -> IO r)
+conduitToProcess c (Handles inh outh errh) =
+  sourceHandle inh $$ c `fuseUpstream`
+  sinkHandles outh errh
 
--- | Proxy final results from the handle and yield them.
-remainder :: Chan (Either (Maybe ByteString) (Maybe ByteString))
-          -> Handle
-          -> (Maybe ByteString -> Either (Maybe ByteString) (Maybe ByteString))
-          -> IO ()
-remainder chan h cons =
-  do bytes <- S.hGetSome h bufSize
-     if S.null bytes
-        then writeChan chan (cons Nothing)
-        else do writeChan chan (cons (Just bytes))
-                remainder chan h cons
+-- | Sink everything into the two handles.
+sinkHandles :: Handle -> Handle -> Consumer (Either ByteString ByteString) IO ()
+sinkHandles out err =
+  CL.mapM_ (\ebs ->
+              case ebs of
+                Left bs -> S.hPut out bs
+                Right bs -> S.hPut err bs)
 
--- | Proxy live results from the given handle and yield them.
-proxy :: MonadIO m
-      => Handle -> (ByteString -> o) -> ConduitM i o m ()
-proxy h cons =
-  do ready <- liftIO (hReady' h)
-     if not ready
-        then return ()
-        else do bytes <- liftIO (S.hGetSome h bufSize)
-                yield (cons bytes)
-                proxy h cons
+-- | Create a pipe.
+createHandles :: IO (Handle, Handle)
+createHandles =
+  mask_ (do (inFD,outFD) <- createPipe
+            x <- fdToHandle inFD
+            y <- fdToHandle outFD
+            hSetBuffering x NoBuffering
+            hSetBuffering y NoBuffering
+            return (x,y))
 
--- | Is the handle ready? Catches any exceptions.
-hReady' :: Handle -> IO Bool
-hReady' h =
-  E.catch (hReady h)
-          (\(E.SomeException _) -> return False)
+-- | Fuse two processes.
+fuseProcess :: (Handles -> IO ()) -> (Handles -> IO r) -> (Handles -> IO r)
+fuseProcess left right (Handles in1 out2 err) =
+  do (in2,out1) <- createHandles
+     runConcurrently
+       (Concurrently
+          (left (Handles in1 out1 err) `finally`
+           hClose out1) *>
+        Concurrently
+          (right (Handles in2 out2 err) `finally`
+           hClose in2))
 
--- | A safer 'waitForProcess'.
-waitForProcess' :: ProcessHandle -> IO ExitCode
-waitForProcess' ph =
-  E.catch (waitForProcess ph)
-          (\(E.SomeException _) ->
-             return ExitSuccess)
+-- | Fuse two conduits.
+fuseConduit :: Monad m
+            => ConduitM ByteString (Either ByteString ByteString) m ()
+            -> ConduitM ByteString (Either ByteString ByteString) m r
+            -> ConduitM ByteString (Either ByteString ByteString) m r
+fuseConduit left right = left =$= getZipConduit right'
+  where right' =
+          ZipConduit (CL.filter isRight) *>
+          ZipConduit
+            (CL.mapMaybe (either (const Nothing) Just) =$=
+             right)
+        isRight Right{} = True
+        isRight Left{} = False
 
--- | Polyfill for base < 4.7
-isLeft :: Either a b -> Bool
-isLeft (Left _) = True
-isLeft _        = False
+-- | Fuse a conduit with a process.
+fuseConduitProcess :: ConduitM ByteString (Either ByteString ByteString) IO ()
+                   -> (Handles -> IO r)
+                   -> (Handles -> IO r)
+fuseConduitProcess left right (Handles in1 out2 err) =
+  do (in2,out1) <- createHandles
+     runConcurrently
+       (Concurrently
+          ((sourceHandle in1 $$ left =$
+            sinkHandles out1 err) `finally`
+           hClose out1) *>
+        Concurrently
+          (right (Handles in2 out2 err) `finally`
+           hClose in2))
+
+-- | Fuse a process with a conduit.
+fuseProcessConduit :: (Handles -> IO ())
+                   -> ConduitM ByteString (Either ByteString ByteString) IO r
+                   -> (Handles -> IO r)
+fuseProcessConduit left right (Handles in1 out2 err) =
+  do (in2,out1) <- createHandles
+     runConcurrently
+       (Concurrently
+          (left (Handles in1 out1 err) `finally`
+           hClose out1) *>
+        Concurrently
+          ((sourceHandle in2 $$ right `fuseUpstream`
+            sinkHandles out2 err) `finally`
+           hClose in2))
+
+-- | Fuse one segment with another.
+fuseSegment :: Segment () -> Segment r -> Segment r
+SegmentConduit x `fuseSegment` SegmentConduit y =
+  SegmentConduit (fuseConduit x y)
+SegmentConduit x `fuseSegment` SegmentProcess y =
+  SegmentProcess (fuseConduitProcess x y)
+SegmentProcess x `fuseSegment` SegmentConduit y =
+  SegmentProcess (fuseProcessConduit x y)
+SegmentProcess x `fuseSegment` SegmentProcess y =
+  SegmentProcess (fuseProcess x y)
diff --git a/src/Data/Conduit/Shell/Types.hs b/src/Data/Conduit/Shell/Types.hs
--- a/src/Data/Conduit/Shell/Types.hs
+++ b/src/Data/Conduit/Shell/Types.hs
@@ -21,19 +21,8 @@
 import Control.Monad.Trans.Class
 import Control.Monad.Trans.Control
 import Control.Monad.Trans.Resource
-import Data.ByteString (ByteString)
 import Data.Conduit
 import Data.Typeable
-
--- | A chunk, either stdout/stdin or stderr. Used for both input and
--- output.
-type Chunk = Either ByteString ByteString
-
--- | Either stdout or stderr.
-data ChunkType
-  = Stderr -- ^ Stderr.
-  | Stdout -- ^ Stdin or stdout.
-  deriving (Eq,Enum,Bounded)
 
 -- | Shell transformer.
 newtype ShellT m a =
diff --git a/src/Data/Conduit/Shell/Variadic.hs b/src/Data/Conduit/Shell/Variadic.hs
--- a/src/Data/Conduit/Shell/Variadic.hs
+++ b/src/Data/Conduit/Shell/Variadic.hs
@@ -9,12 +9,9 @@
   ,CmdArg(..))
   where
 
-import           Control.Monad.Trans.Resource
 import qualified Data.ByteString as SB
 import qualified Data.ByteString.Lazy as LB
-import           Data.Conduit
 import           Data.Conduit.Shell.Process
-import           Data.Conduit.Shell.Types
 import qualified Data.Text as ST
 import qualified Data.Text.Encoding as ST
 import qualified Data.Text.Lazy as LT
@@ -26,22 +23,14 @@
 variadicProcess name = spr name []
 
 -- | Make the final conduit.
-makeProcessLauncher :: (MonadResource m)
-                    => String -> [ST.Text] -> Conduit Chunk m Chunk
+makeProcessLauncher :: String -> [ST.Text] -> Segment ()
 makeProcessLauncher name args = proc name (map ST.unpack args)
 
 -- | Process return type.
 class ProcessType t where
     spr :: String -> [ST.Text] -> t
 
--- | The real type should be:
---
--- @ConduitM Chunk Chunk m ()@
---
--- But with this more liberal instance head we catch all cases in the
--- instance resolver, and then apply the equality restrictions later.
---
-instance (MonadResource m, c ~ Chunk, c' ~ Chunk, r ~ ()) => ProcessType (ConduitM c c' m r) where
+instance (r ~ ()) => ProcessType (Segment r) where
     spr name args = makeProcessLauncher name (reverse args)
 
 -- | Accept strings as arguments.