diff --git a/CHANGELOG b/CHANGELOG
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,38 +1,45 @@
-0.5.0.2
--------
-
-- A bug slipped by :(
-
-0.5.0.1
--------
-
-- Changes in the internals to reduce the number of spawned threads.
-
-0.5.0.0
--------
-
-- Now the constructors for "PipingPolicy" take "Siphons" directly, instead of
-  continuation functions.
-- Removed "separated" and "combined" functions, added new "PipingPolicy"
-  constructors in their place.
-- Removed "LeftoverPolicy", its function is now performed by Siphons.
-- Removed "surely", "safely", "monoidally" functions with confusing signatures.
-- Removed all occurrences of unbounded buffers in the code.
-- Implemented support for branching pipelines of processes.
-
-0.3.0.0
--------
-
-- Renamed many functions (the number-based naming schema is gone).
-- Decisions about how to consume the standard streams were moved to the new
-"PipingPolicy" type.
-- "ForkProd" is now "Siphon". 
-- "execute" can now throw IOExceptions (but see "safeExecute").
-- Some complicated types were hidden behind data declarations.
-- Functor, Bifunctor and Profunctor instances for some types.
-
-0.0.1.1
--------
-
-- Added some examples.
-- Re-exported System.Process from System.Process.Streaming.
+0.6.0.0
+-------
+
+- Now compatible with earlier versions of transformers.
+- Simplified LinePolicy.
+- Simplified pipeline functions and datatypes.
+
+0.5.0.2
+-------
+
+- A bug slipped by :(
+
+0.5.0.1
+-------
+
+- Changes in the internals to reduce the number of spawned threads.
+
+0.5.0.0
+-------
+
+- Now the constructors for "PipingPolicy" take "Siphons" directly, instead of
+  continuation functions.
+- Removed "separated" and "combined" functions, added new "PipingPolicy"
+  constructors in their place.
+- Removed "LeftoverPolicy", its function is now performed by Siphons.
+- Removed "surely", "safely", "monoidally" functions with confusing signatures.
+- Removed all occurrences of unbounded buffers in the code.
+- Implemented support for branching pipelines of processes.
+
+0.3.0.0
+-------
+
+- Renamed many functions (the number-based naming schema is gone).
+- Decisions about how to consume the standard streams were moved to the new
+"PipingPolicy" type.
+- "ForkProd" is now "Siphon". 
+- "execute" can now throw IOExceptions (but see "safeExecute").
+- Some complicated types were hidden behind data declarations.
+- Functor, Bifunctor and Profunctor instances for some types.
+
+0.0.1.1
+-------
+
+- Added some examples.
+- Re-exported System.Process from System.Process.Streaming.
diff --git a/LICENSE b/LICENSE
--- a/LICENSE
+++ b/LICENSE
@@ -1,27 +1,27 @@
-Copyright (c) 2014, Daniel Díaz Carrete
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without modification,
-are permitted provided that the following conditions are met:
-
-* Redistributions of source code must retain the above copyright notice, this
-  list of conditions and the following disclaimer.
-
-* Redistributions in binary form must reproduce the above copyright notice, this
-  list of conditions and the following disclaimer in the documentation and/or
-  other materials provided with the distribution.
-
-* Neither the name of the {organization} nor the names of its
-  contributors may be used to endorse or promote products derived from
-  this software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
-ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
-WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
-ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
-(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
-LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
-ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
-SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+Copyright (c) 2014, Daniel Díaz Carrete
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice, this
+  list of conditions and the following disclaimer in the documentation and/or
+  other materials provided with the distribution.
+
+* Neither the name of the {organization} nor the names of its
+  contributors may be used to endorse or promote products derived from
+  this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -1,19 +1,19 @@
-process-streaming
-=================
-
-Exploring how to interact with system processes using a streaming library
-(pipes).
-
-The basic goals:
-
-- Concurrent, streaming access to stdin, stdout and stderr.
-
-- Easy integration with regular consumers, parsers from pipes-parse and various
-  folds.
-
-- Avoid launching exceptions: use Either or similar solution to signal non-IO
-  related error conditions.
-
-Relevant thread in the Haskell Pipes Google Group:
-
-https://groups.google.com/forum/#!searchin/haskell-pipes/pipes$20process/haskell-pipes/JFfyquj5HAg/Lxz7p50JOh4J
+process-streaming
+=================
+
+Exploring how to interact with system processes using a streaming library
+(pipes).
+
+The basic goals:
+
+- Concurrent, streaming access to stdin, stdout and stderr.
+
+- Easy integration with regular consumers, parsers from pipes-parse and various
+  folds.
+
+- Avoid launching exceptions: use Either or similar solution to signal non-IO
+  related error conditions.
+
+Relevant thread in the Haskell Pipes Google Group:
+
+https://groups.google.com/forum/#!searchin/haskell-pipes/pipes$20process/haskell-pipes/JFfyquj5HAg/Lxz7p50JOh4J
diff --git a/Setup.hs b/Setup.hs
--- a/Setup.hs
+++ b/Setup.hs
@@ -1,2 +1,2 @@
-import Distribution.Simple
-main = defaultMain
+import Distribution.Simple
+main = defaultMain
diff --git a/process-streaming.cabal b/process-streaming.cabal
--- a/process-streaming.cabal
+++ b/process-streaming.cabal
@@ -1,86 +1,86 @@
-name:          process-streaming
-version:       0.5.0.2
-license:       BSD3
-license-file:  LICENSE
-data-files:    
-author:        Daniel Díaz Carrete
-maintainer:    diaz_carrete@yahoo.com
-category:      System
-build-type:    Simple
-cabal-version: >= 1.10
-Synopsis:      Streaming interface to system processes. 
-Description:   Concurrent, streaming access to the input and outputs of system processes.
-
-Extra-Source-Files:
-    README.md
-    CHANGELOG
-
-Library
-    default-language: Haskell2010
-    hs-source-dirs: src
-    exposed-modules: 
-        System.Process.Streaming
-        System.Process.Lens
-    other-modules: 
-    build-depends:         
-        base >= 4.4 && < 5,
-        transformers >= 0.4 && < 0.5,
-        mtl >= 2.2 && < 2.3,
-        free >= 4.2 && < 5,
-        bifunctors >= 4.1 && < 5,
-        async >= 2.0.1 && < 2.1,
-        process >= 1.2.0 && < 1.3,
-        pipes >= 4.1.2 && < 4.2,
-        pipes-bytestring >= 2.1.0 && < 2.2,
-        pipes-text >= 0.0.0.10 && < 0.0.2,
-        text >= 0.11.2 && < 1.2,
-        pipes-concurrency >= 2.0.2 && < 3,
-        pipes-safe >= 2.2.0 && < 3,
-        pipes-parse >=3.0.1 && <3.1,
-        exceptions >= 0.6.0 && < 0.7,
-        void >= 0.6 && < 0.7,
-        containers >= 0.4,
-        semigroups >= 0.15 && < 0.16
-        
-Test-suite test
-    default-language:
-      Haskell2010
-    type:
-      exitcode-stdio-1.0
-    hs-source-dirs:
-      tests
-    main-is:
-      test.hs
-    build-depends:
-        base >= 4.4 && < 5
-      , transformers >= 0.4 && < 0.5
-      , mtl >= 2.2 && < 2.3
-      , free >= 4.2 && < 5
-      , bifunctors >= 4.1 && < 5
-      , async >= 2.0.1 && < 2.1
-      , process >= 1.2.0 && < 1.3
-      , pipes >= 4.1.2 && < 4.2
-      , pipes-bytestring >= 2.1.0 && < 2.2
-      , pipes-text >= 0.0.0.10 && < 0.0.2
-      , text >= 0.11.2 && < 1.2
-      , pipes-concurrency >= 2.0.2 && < 3
-      , pipes-safe >= 2.2.0 && < 3
-      , pipes-parse >=3.0.1 && <3.1
-      , exceptions >= 0.6.0 && < 0.7
-      , void >= 0.6 && < 0.7
-      , semigroups >= 0.15 && < 0.16
-      , containers >= 0.4
-      , process-streaming
-      , tasty >= 0.9
-      , tasty-hunit >= 0.9
-      , attoparsec >= 0.11
-      , pipes-attoparsec >= 0.5
-      , pipes-group >= 1.0.1
-      , bytestring >= 0.10
-      , lens >= 4
-      , directory >= 1.2
-
-Source-repository head
-    type:     git
-    location: https://github.com/danidiaz/process-streaming.git
-
+name:          process-streaming
+version:       0.6.0.0
+license:       BSD3
+license-file:  LICENSE
+data-files:    
+author:        Daniel Díaz Carrete
+maintainer:    diaz_carrete@yahoo.com
+category:      System
+build-type:    Simple
+cabal-version: >= 1.10
+Synopsis:      Streaming interface to system processes. 
+Description:   Concurrent, streaming access to the input and outputs of system processes.
+
+Extra-Source-Files:
+    README.md
+    CHANGELOG
+
+Library
+    default-language: Haskell2010
+    hs-source-dirs: src
+    exposed-modules: 
+        System.Process.Streaming
+        System.Process.Lens
+    other-modules: 
+    build-depends:         
+        base >= 4.4 && < 5,
+        transformers >= 0.2 && < 0.5,
+        transformers-compat == 0.3.*,
+        free >= 4.2 && < 5,
+        bifunctors >= 4.1 && < 5,
+        async >= 2.0.1 && < 2.1,
+        process >= 1.2.0 && < 1.3,
+        pipes >= 4.1.2 && < 4.2,
+        pipes-bytestring >= 2.1.0 && < 2.2,
+        pipes-text >= 0.0.0.10 && < 0.0.2,
+        text >= 0.11.2 && < 1.2,
+        pipes-concurrency >= 2.0.2 && < 3,
+        pipes-safe >= 2.2.0 && < 3,
+        pipes-parse >=3.0.1 && <3.1,
+        exceptions >= 0.6.0 && < 0.7,
+        void >= 0.6 && < 0.7,
+        containers >= 0.4,
+        semigroups >= 0.15 && < 0.16
+
+Test-suite test
+    default-language:
+      Haskell2010
+    type:
+      exitcode-stdio-1.0
+    hs-source-dirs:
+      tests
+    main-is:
+      test.hs
+    build-depends:
+        base >= 4.4 && < 5
+      , transformers >= 0.2 && < 0.5
+      , transformers-compat == 0.3.*
+      , free >= 4.2 && < 5
+      , bifunctors >= 4.1 && < 5
+      , async >= 2.0.1 && < 2.1
+      , process >= 1.2.0 && < 1.3
+      , pipes >= 4.1.2 && < 4.2
+      , pipes-bytestring >= 2.1.0 && < 2.2
+      , pipes-text >= 0.0.0.10 && < 0.0.2
+      , text >= 0.11.2 && < 1.2
+      , pipes-concurrency >= 2.0.2 && < 3
+      , pipes-safe >= 2.2.0 && < 3
+      , pipes-parse >=3.0.1 && <3.1
+      , exceptions >= 0.6.0 && < 0.7
+      , void >= 0.6 && < 0.7
+      , semigroups >= 0.15 && < 0.16
+      , containers >= 0.4
+      , process-streaming
+      , tasty >= 0.9
+      , tasty-hunit >= 0.9
+      , attoparsec >= 0.11
+      , pipes-attoparsec >= 0.5
+      , pipes-group >= 1.0.1
+      , bytestring >= 0.10
+      , lens >= 4
+      , directory >= 1.2
+
+Source-repository head
+    type:     git
+    location: https://github.com/danidiaz/process-streaming.git
+
diff --git a/src/System/Process/Lens.hs b/src/System/Process/Lens.hs
--- a/src/System/Process/Lens.hs
+++ b/src/System/Process/Lens.hs
@@ -1,229 +1,229 @@
-
--- |
--- Lenses and traversals for 'CreateProcess' and related types.
---
------------------------------------------------------------------------------
-
-{-# LANGUAGE DeriveDataTypeable #-}
-{-# LANGUAGE RankNTypes #-}
-
-module System.Process.Lens ( 
-         _cmdspec
-       , _ShellCommand
-       , _RawCommand
-       , _cwd
-       , _env
-       , streams
-       , _close_fds
-       , _create_group
-       , _delegate_ctlc 
-       , handles
-       , nohandles
-       , handleso
-       , handlese
-       , handlesoe
-       , handlesi
-       , handlesio
-       , handlesie
-       , handlesioe
-    ) where
-
-import Data.Maybe
-import Data.Functor.Identity
-import Data.Monoid
-import Data.Traversable
-import Control.Applicative
-import System.IO
-import System.Process
-
-{-|
-    > _cmdspec :: Lens' CreateProcess CmdSpec 
--}
-_cmdspec :: forall f. Functor f => (CmdSpec -> f CmdSpec) -> CreateProcess -> f CreateProcess 
-_cmdspec f c = setCmdSpec c <$> f (cmdspec c)
-    where
-    setCmdSpec c cmdspec' = c { cmdspec = cmdspec' } 
-
-{-|
-    > _ShellCommand :: Prism' CmdSpec String
--}
-_ShellCommand :: forall m. Applicative m => (String -> m String) -> CmdSpec -> m CmdSpec 
-_ShellCommand f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap ShellCommand (f r)
-    where    
-    impure (ShellCommand str) = Right str
-    impure x = Left x
-
-{-|
-    > _RawCommand :: Prism' CmdSpec (FilePath,[String])
--}
-_RawCommand :: forall m. Applicative m => ((FilePath,[String]) -> m (FilePath,[String])) -> CmdSpec -> m CmdSpec 
-_RawCommand f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-    impure (RawCommand fpath strs) = Right (fpath,strs)
-    impure x = Left x
-    justify (fpath,strs) = RawCommand fpath strs
-
-{-|
-    > _cwd :: Lens' CreateProcess (Maybe FilePath)
--}
-_cwd :: forall f. Functor f => (Maybe FilePath -> f (Maybe FilePath)) -> CreateProcess -> f CreateProcess 
-_cwd f c = setCwd c <$> f (cwd c)
-    where
-    setCwd c cwd' = c { cwd = cwd' } 
-
-{-|
-    > _env :: Lens' CreateProcess (Maybe [(String,String)])
--}
-_env :: forall f. Functor f => (Maybe [(String, String)] -> f (Maybe [(String, String)])) -> CreateProcess -> f CreateProcess 
-_env f c = setEnv c <$> f (env c)
-    where
-    setEnv c env' = c { env = env' } 
-
-{-| 
-    A lens for the @(std_in,std_out,std_err)@ triplet.  
-
-    > streams :: Lens' CreateProcess (StdStream,StdStream,StdStream)
--}
-streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess 
-streams f c = setStreams c <$> f (getStreams c)
-    where 
-        getStreams c = (std_in c,std_out c, std_err c)
-        setStreams c (s1,s2,s3) = c { std_in  = s1 
-                                    , std_out = s2 
-                                    , std_err = s3 
-                                    } 
-
-_close_fds :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess 
-_close_fds f c = set_close_fds c <$> f (close_fds c)
-    where
-    set_close_fds c cwd' = c { close_fds = cwd' } 
-
-
-_create_group :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess 
-_create_group f c = set_create_group c <$> f (create_group c)
-    where
-    set_create_group c cwd' = c { create_group = cwd' } 
-
-_delegate_ctlc :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess 
-_delegate_ctlc f c = set_delegate_ctlc c <$> f (delegate_ctlc c)
-    where
-    set_delegate_ctlc c cwd' = c { delegate_ctlc = cwd' } 
-
-{-|
-    A 'Lens' for the return value of 'createProcess' that focuses on the handles.
-
-    > handles :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle)
- -}
-handles :: forall m. Functor m => ((Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) -> m (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) 
-handles f quad = setHandles quad <$> f (getHandles quad)  
-    where
-        setHandles (c1'',c2'',c3'',c4'') (c1',c2',c3') = (c1',c2',c3',c4'')
-        getHandles (c1'',c2'',c3'',c4'') = (c1'',c2'',c3'')
-    
-
-{-|
-    A 'Prism' that matches when none of the standard streams have been piped.
-
-    > nohandles :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) ()
- -}
-nohandles :: forall m. Applicative m => (() -> m ()) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-nohandles f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Nothing, Nothing, Nothing) = Right () 
-        impure x = Left x
-        justify () = (Nothing, Nothing, Nothing)  
-
-
-{-|
-    A 'Prism' that matches when only @stdin@ has been piped.
-
-    > handlesi :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)
- -}
-handlesi :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handlesi f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Just h1, Nothing, Nothing) = Right h1
-        impure x = Left x
-        justify h1 = (Just h1, Nothing, Nothing)  
-
-handlesio :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handlesio f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Just h1, Just h2, Nothing) = Right (h1,h2)
-        impure x = Left x
-        justify (h1,h2) = (Just h1, Just h2, Nothing)  
-
-handlesie :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handlesie f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Just h1, Nothing, Just h2) = Right (h1,h2)
-        impure x = Left x
-        justify (h1,h2) = (Just h1, Nothing, Just h2)  
-
-{-|
-    A 'Prism' that matches when all three @stdin@, @stdout@ and @stderr@ have been piped.
-
-    > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle, Handle)
- -}
-handlesioe :: forall m. Applicative m => ((Handle, Handle, Handle) -> m (Handle, Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handlesioe f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Just h1, Just h2, Just h3) = Right (h1, h2, h3) 
-        impure x = Left x
-        justify (h1, h2, h3) = (Just h1, Just h2, Just h3)  
-
-{-|
-    A 'Prism' that matches when only @stdout@ and @stderr@ have been piped.
-
-    > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle)
- -}
-handlesoe :: forall m. Applicative m => ((Handle, Handle) -> m (Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handlesoe f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Nothing, Just h2, Just h3) = Right (h2, h3) 
-        impure x = Left x
-        justify (h2, h3) = (Nothing, Just h2, Just h3)  
-
-{-|
-    A 'Prism' that matches when only @stdout@ has been piped.
-
-    > handleso :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)
- -}
-handleso :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handleso f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Nothing, Just h2, Nothing) = Right h2
-        impure x = Left x
-        justify h2 = (Nothing, Just h2, Nothing)  
-
-{-|
-    A 'Prism' that matches when only @stderr@ has been piped.
-
-    > handlese :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)
- -}
-handlese :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
-handlese f quad = case impure quad of
-    Left l -> pure l
-    Right r -> fmap justify (f r)
-    where    
-        impure (Nothing, Nothing, Just h2) = Right h2
-        impure x = Left x
-        justify h2 = (Nothing, Nothing, Just h2)  
+
+-- |
+-- Lenses and traversals for 'CreateProcess' and related types.
+--
+-----------------------------------------------------------------------------
+
+{-# LANGUAGE DeriveDataTypeable #-}
+{-# LANGUAGE RankNTypes #-}
+
+module System.Process.Lens ( 
+         _cmdspec
+       , _ShellCommand
+       , _RawCommand
+       , _cwd
+       , _env
+       , streams
+       , _close_fds
+       , _create_group
+       , _delegate_ctlc 
+       , handles
+       , nohandles
+       , handleso
+       , handlese
+       , handlesoe
+       , handlesi
+       , handlesio
+       , handlesie
+       , handlesioe
+    ) where
+
+import Data.Maybe
+import Data.Functor.Identity
+import Data.Monoid
+import Data.Traversable
+import Control.Applicative
+import System.IO
+import System.Process
+
+{-|
+    > _cmdspec :: Lens' CreateProcess CmdSpec 
+-}
+_cmdspec :: forall f. Functor f => (CmdSpec -> f CmdSpec) -> CreateProcess -> f CreateProcess 
+_cmdspec f c = setCmdSpec c <$> f (cmdspec c)
+    where
+    setCmdSpec c cmdspec' = c { cmdspec = cmdspec' } 
+
+{-|
+    > _ShellCommand :: Prism' CmdSpec String
+-}
+_ShellCommand :: forall m. Applicative m => (String -> m String) -> CmdSpec -> m CmdSpec 
+_ShellCommand f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap ShellCommand (f r)
+    where    
+    impure (ShellCommand str) = Right str
+    impure x = Left x
+
+{-|
+    > _RawCommand :: Prism' CmdSpec (FilePath,[String])
+-}
+_RawCommand :: forall m. Applicative m => ((FilePath,[String]) -> m (FilePath,[String])) -> CmdSpec -> m CmdSpec 
+_RawCommand f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+    impure (RawCommand fpath strs) = Right (fpath,strs)
+    impure x = Left x
+    justify (fpath,strs) = RawCommand fpath strs
+
+{-|
+    > _cwd :: Lens' CreateProcess (Maybe FilePath)
+-}
+_cwd :: forall f. Functor f => (Maybe FilePath -> f (Maybe FilePath)) -> CreateProcess -> f CreateProcess 
+_cwd f c = setCwd c <$> f (cwd c)
+    where
+    setCwd c cwd' = c { cwd = cwd' } 
+
+{-|
+    > _env :: Lens' CreateProcess (Maybe [(String,String)])
+-}
+_env :: forall f. Functor f => (Maybe [(String, String)] -> f (Maybe [(String, String)])) -> CreateProcess -> f CreateProcess 
+_env f c = setEnv c <$> f (env c)
+    where
+    setEnv c env' = c { env = env' } 
+
+{-| 
+    A lens for the @(std_in,std_out,std_err)@ triplet.  
+
+    > streams :: Lens' CreateProcess (StdStream,StdStream,StdStream)
+-}
+streams :: forall f. Functor f => ((StdStream,StdStream,StdStream) -> f (StdStream,StdStream,StdStream)) -> CreateProcess -> f CreateProcess 
+streams f c = setStreams c <$> f (getStreams c)
+    where 
+        getStreams c = (std_in c,std_out c, std_err c)
+        setStreams c (s1,s2,s3) = c { std_in  = s1 
+                                    , std_out = s2 
+                                    , std_err = s3 
+                                    } 
+
+_close_fds :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess 
+_close_fds f c = set_close_fds c <$> f (close_fds c)
+    where
+    set_close_fds c cwd' = c { close_fds = cwd' } 
+
+
+_create_group :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess 
+_create_group f c = set_create_group c <$> f (create_group c)
+    where
+    set_create_group c cwd' = c { create_group = cwd' } 
+
+_delegate_ctlc :: forall f. Functor f => (Bool -> f Bool) -> CreateProcess -> f CreateProcess 
+_delegate_ctlc f c = set_delegate_ctlc c <$> f (delegate_ctlc c)
+    where
+    set_delegate_ctlc c cwd' = c { delegate_ctlc = cwd' } 
+
+{-|
+    A 'Lens' for the return value of 'createProcess' that focuses on the handles.
+
+    > handles :: Lens' (Maybe Handle, Maybe Handle, Maybe Handle,ProcessHandle) (Maybe Handle, Maybe Handle, Maybe Handle)
+ -}
+handles :: forall m. Functor m => ((Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) -> m (Maybe Handle,Maybe Handle ,Maybe Handle,ProcessHandle) 
+handles f quad = setHandles quad <$> f (getHandles quad)  
+    where
+        setHandles (c1'',c2'',c3'',c4'') (c1',c2',c3') = (c1',c2',c3',c4'')
+        getHandles (c1'',c2'',c3'',c4'') = (c1'',c2'',c3'')
+    
+
+{-|
+    A 'Prism' that matches when none of the standard streams have been piped.
+
+    > nohandles :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) ()
+ -}
+nohandles :: forall m. Applicative m => (() -> m ()) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+nohandles f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Nothing, Nothing, Nothing) = Right () 
+        impure x = Left x
+        justify () = (Nothing, Nothing, Nothing)  
+
+
+{-|
+    A 'Prism' that matches when only @stdin@ has been piped.
+
+    > handlesi :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)
+ -}
+handlesi :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handlesi f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Just h1, Nothing, Nothing) = Right h1
+        impure x = Left x
+        justify h1 = (Just h1, Nothing, Nothing)  
+
+handlesio :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handlesio f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Just h1, Just h2, Nothing) = Right (h1,h2)
+        impure x = Left x
+        justify (h1,h2) = (Just h1, Just h2, Nothing)  
+
+handlesie :: forall m. Applicative m => ((Handle,Handle) -> m (Handle,Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handlesie f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Just h1, Nothing, Just h2) = Right (h1,h2)
+        impure x = Left x
+        justify (h1,h2) = (Just h1, Nothing, Just h2)  
+
+{-|
+    A 'Prism' that matches when all three @stdin@, @stdout@ and @stderr@ have been piped.
+
+    > handlesioe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle, Handle)
+ -}
+handlesioe :: forall m. Applicative m => ((Handle, Handle, Handle) -> m (Handle, Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handlesioe f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Just h1, Just h2, Just h3) = Right (h1, h2, h3) 
+        impure x = Left x
+        justify (h1, h2, h3) = (Just h1, Just h2, Just h3)  
+
+{-|
+    A 'Prism' that matches when only @stdout@ and @stderr@ have been piped.
+
+    > handlesoe :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle, Handle)
+ -}
+handlesoe :: forall m. Applicative m => ((Handle, Handle) -> m (Handle, Handle)) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handlesoe f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Nothing, Just h2, Just h3) = Right (h2, h3) 
+        impure x = Left x
+        justify (h2, h3) = (Nothing, Just h2, Just h3)  
+
+{-|
+    A 'Prism' that matches when only @stdout@ has been piped.
+
+    > handleso :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)
+ -}
+handleso :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handleso f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Nothing, Just h2, Nothing) = Right h2
+        impure x = Left x
+        justify h2 = (Nothing, Just h2, Nothing)  
+
+{-|
+    A 'Prism' that matches when only @stderr@ has been piped.
+
+    > handlese :: Prism' (Maybe Handle, Maybe Handle, Maybe Handle) (Handle)
+ -}
+handlese :: forall m. Applicative m => (Handle -> m Handle) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)
+handlese f quad = case impure quad of
+    Left l -> pure l
+    Right r -> fmap justify (f r)
+    where    
+        impure (Nothing, Nothing, Just h2) = Right h2
+        impure x = Left x
+        justify h2 = (Nothing, Nothing, Just h2)  
diff --git a/src/System/Process/Streaming.hs b/src/System/Process/Streaming.hs
--- a/src/System/Process/Streaming.hs
+++ b/src/System/Process/Streaming.hs
@@ -1,860 +1,916 @@
-
--- |
--- This module contains helper functions and types built on top of
--- "System.Process" and "Pipes".
---
--- They provide concurrent, streaming access to the inputs and outputs of
--- system processes.
---
--- Error conditions that are not directly related to IO are made explicit
--- in the types.
---
--- Regular 'Consumer's, 'Parser's from @pipes-parse@ and folds from
--- "Pipes.Prelude" (also folds from @pipes-bytestring@ and @pipes-text@)
--- can be used to consume the output streams of the external processes.
---
------------------------------------------------------------------------------
-
-{-# LANGUAGE DeriveDataTypeable #-}
-{-# LANGUAGE DeriveFunctor #-}
-{-# LANGUAGE ExistentialQuantification #-}
-{-# LANGUAGE RankNTypes #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE ViewPatterns #-}
-
-module System.Process.Streaming ( 
-        -- * Execution
-          execute
-        , executeFallibly
-        -- * Piping Policies
-        , PipingPolicy
-        , nopiping
-        , pipeo
-        , pipee
-        , pipeoe
-        , pipeoec
-        , pipei
-        , pipeio
-        , pipeie
-        , pipeioe
-        , pipeioec
-
-        -- * Pumping bytes into stdin
-        , Pump (..)
-        , fromProducer
-        , fromSafeProducer
-        , fromFallibleProducer
-        -- * Siphoning bytes out of stdout/stderr
-        , Siphon
-        , siphon
-        , siphon'
-        , fromFold
-        , fromFold'
-        , fromFold'_
-        , fromConsumer
-        , fromSafeConsumer
-        , fromFallibleConsumer
-        , fromParser
-        , unwanted
-        , DecodingFunction
-        , encoded
-        -- * Line handling
-        , LinePolicy
-        , linePolicy
-        -- * Pipelines
-        , executePipeline
-        , executePipelineFallibly
-        , CreatePipeline (..)
-        , simplePipeline
-        , Stage (..)
-        , SubsequentStage (..)
-        -- * Re-exports
-        -- $reexports
-        , module System.Process
-    ) where
-
-import Data.Maybe
-import Data.Bifunctor
-import Data.Functor.Identity
-import Data.Either
-import Data.Monoid
-import Data.Foldable
-import Data.Traversable
-import Data.Typeable
-import Data.Tree
-import Data.Text 
-import Data.Text.Encoding 
-import Data.Void
-import Data.List.NonEmpty
-import qualified Data.List.NonEmpty as N
-import Control.Applicative
-import Control.Monad
-import Control.Monad.Trans.Free
-import Control.Monad.Except
-import Control.Monad.State
-import Control.Monad.Writer.Strict
-import qualified Control.Monad.Catch as C
-import Control.Exception
-import Control.Concurrent
-import Control.Concurrent.Async
-import Pipes
-import qualified Pipes as P
-import qualified Pipes.Prelude as P
-import Pipes.Lift
-import Pipes.ByteString
-import Pipes.Parse
-import qualified Pipes.Text as T
-import Pipes.Concurrent
-import Pipes.Safe (SafeT, runSafeT)
-import System.IO
-import System.IO.Error
-import System.Process
-import System.Process.Lens
-import System.Exit
-
-execute :: PipingPolicy Void a -> CreateProcess -> IO (ExitCode,a)
-execute pp cprocess = either absurd id <$> executeFallibly pp cprocess
-
-{-|
-   Executes an external process. The standard streams are piped and consumed in
-a way defined by the 'PipingPolicy' argument. 
-
-   This fuction re-throws any 'IOException's it encounters.
-
-   If the consumption of the standard streams fails with @e@, the whole
-computation is immediately aborted and @e@ is returned. (An exception is not
-thrown in this case.).  
-
-   If an error @e@ or an exception happens, the external process is
-terminated.
- -}
-executeFallibly :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a))
-executeFallibly pp record = case pp of
-      PPNone a -> executeInternal record nohandles $  
-          \() -> (return . Right $ a,return ())
-      PPOutput action -> executeInternal (record{std_out = CreatePipe}) handleso $
-          \h->(action (fromHandle h),hClose h) 
-      PPError action ->  executeInternal (record{std_err = CreatePipe}) handlese $
-          \h->(action (fromHandle h),hClose h)
-      PPOutputError action -> executeInternal (record{std_out = CreatePipe, std_err = CreatePipe}) handlesoe $
-          \(hout,herr)->(action (fromHandle hout,fromHandle herr),hClose hout `finally` hClose herr)
-      PPInput action -> executeInternal (record{std_in = CreatePipe}) handlesi $
-          \h -> (action (toHandle h, hClose h), return ())
-      PPInputOutput action -> executeInternal (record{std_in = CreatePipe,std_out = CreatePipe}) handlesio $
-          \(hin,hout) -> (action (toHandle hin,hClose hin,fromHandle hout), hClose hout)
-      PPInputError action -> executeInternal (record{std_in = CreatePipe,std_err = CreatePipe}) handlesie $
-          \(hin,herr) -> (action (toHandle hin,hClose hin,fromHandle herr), hClose herr)
-      PPInputOutputError action -> executeInternal (record{std_in = CreatePipe, std_out = CreatePipe, std_err = CreatePipe}) handlesioe $
-          \(hin,hout,herr) -> (action (toHandle hin,hClose hin,fromHandle hout,fromHandle herr), hClose hout `finally` hClose herr)
-
-executeInternal :: CreateProcess -> (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (t ->(IO (Either e a),IO ())) -> IO (Either e (ExitCode,a))
-executeInternal record somePrism allocator = mask $ \restore -> do
-    (min,mout,merr,phandle) <- createProcess record
-    case getFirst . getConst . somePrism (Const . First . Just) $ (min,mout,merr) of
-        Nothing -> 
-            throwIO (userError "stdin/stdout/stderr handle unwantedly null")
-            `finally`
-            terminateCarefully phandle 
-        Just t -> 
-            let (action,cleanup) = allocator t in
-            -- Handles must be closed *after* terminating the process, because a close
-            -- operation may block if the external process has unflushed bytes in the stream.
-            (restore (terminateOnError phandle action) `onException` terminateCarefully phandle) `finally` cleanup 
-
-exitCode :: (ExitCode,a) -> Either Int a
-exitCode (ec,a) = case ec of
-    ExitSuccess -> Right a 
-    ExitFailure i -> Left i
-
-terminateCarefully :: ProcessHandle -> IO ()
-terminateCarefully pHandle = do
-    mExitCode <- getProcessExitCode pHandle   
-    case mExitCode of 
-        Nothing -> terminateProcess pHandle  
-        Just _ -> return ()
-
-terminateOnError :: ProcessHandle 
-                 -> IO (Either e a)
-                 -> IO (Either e (ExitCode,a))
-terminateOnError pHandle action = do
-    result <- action
-    case result of
-        Left e -> do    
-            terminateCarefully pHandle
-            return $ Left e
-        Right r -> do 
-            exitCode <- waitForProcess pHandle 
-            return $ Right (exitCode,r)  
-
-{-|
-    A 'PipingPolicy' determines what standard streams will be piped and what to
-do with them.
-
-    The user doesn't need to manually set the 'std_in', 'std_out' and 'std_err'
-fields of the 'CreateProcess' record to 'CreatePipe', this is done
-automatically. 
-
-    A 'PipingPolicy' is parametrized by the type @e@ of errors that can abort
-the processing of the streams.
- -}
--- Knows that there is a stdin, stdout and a stderr,
--- but doesn't know anything about file handlers or CreateProcess.
-data PipingPolicy e a = 
-      PPNone a
-    | PPOutput (Producer ByteString IO () -> IO (Either e a))
-    | PPError (Producer ByteString IO () -> IO (Either e a))
-    | PPOutputError ((Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a))
-    | PPInput ((Consumer ByteString IO (), IO ()) -> IO (Either e a))
-    | PPInputOutput ((Consumer ByteString IO (), IO (),Producer ByteString IO ()) -> IO (Either e a))
-    | PPInputError ((Consumer ByteString IO (), IO (), Producer ByteString IO ()) -> IO (Either e a))
-    | PPInputOutputError ((Consumer ByteString IO (),IO (),Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a))
-    deriving (Functor)
-
-instance Bifunctor PipingPolicy where
-  bimap f g pp = case pp of
-        PPNone a -> PPNone $ g a 
-        PPOutput action -> PPOutput $ fmap (fmap (bimap f g)) action
-        PPError action -> PPError $ fmap (fmap (bimap f g)) action
-        PPOutputError action -> PPOutputError $ fmap (fmap (bimap f g)) action
-        PPInput action -> PPInput $ fmap (fmap (bimap f g)) action
-        PPInputOutput action -> PPInputOutput $ fmap (fmap (bimap f g)) action
-        PPInputError action -> PPInputError $ fmap (fmap (bimap f g)) action
-        PPInputOutputError action -> PPInputOutputError $ fmap (fmap (bimap f g)) action
-
-{-|
-    Do not pipe any standard stream. 
--}
-nopiping :: PipingPolicy e ()
-nopiping = PPNone ()
-
-{-|
-    Pipe @stdout@.
--}
-pipeo :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a
-pipeo (runSiphon -> siphonout) = PPOutput $ siphonout
-
-{-|
-    Pipe @stderr@.
--}
-pipee :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a
-pipee (runSiphon -> siphonout) = PPError $ siphonout
-
-{-|
-    Pipe @stdout@ and @stderr@.
--}
-pipeoe :: (Show e,Typeable e) => Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b)
-pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = 
-    PPOutputError $ uncurry $ separated siphonout siphonerr  
-
-{-|
-    Pipe @stdout@ and @stderr@ and consume them combined as 'Text'.  
--}
-pipeoec :: (Show e,Typeable e) => LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a
-pipeoec policy1 policy2 (runSiphon -> siphon) = 
-    PPOutputError $ uncurry $ combined policy1 policy2 siphon  
-
-{-|
-    Pipe @stdin@.
--}
-pipei :: (Show e, Typeable e) => Pump ByteString e i -> PipingPolicy e i
-pipei (Pump feeder) = PPInput $ \(consumer,cleanup) -> feeder consumer `finally` cleanup
-
-{-|
-    Pipe @stdin@ and @stdout@.
--}
-pipeio :: (Show e, Typeable e)
-        => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a)
-pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) ->
-        (conceit (feeder consumer `finally` cleanup) (siphonout producer))
-
-{-|
-    Pipe @stdin@ and @stderr@.
--}
-pipeie :: (Show e, Typeable e)
-        => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a)
-pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) ->
-        (conceit (feeder consumer `finally` cleanup) (siphonerr producer))
-
-{-|
-    Pipe @stdin@, @stdout@ and @stderr@.
--}
-pipeioe :: (Show e, Typeable e)
-        => Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b)
-pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $
-    \(consumer,cleanup,outprod,errprod) -> 
-             (conceit (feeder consumer `finally` cleanup) 
-                      (separated siphonout siphonerr outprod errprod))
-    where
-        flattenTuple (i, (a, b)) = (i,a,b)
-
-{-|
-    Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'.
--}
-pipeioec :: (Show e, Typeable e)
-        => Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a)
-pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $
-    \(consumer,cleanup,outprod,errprod) -> 
-             (conceit (feeder consumer `finally` cleanup) 
-                      (combined policy1 policy2 siphon outprod errprod))
-
-separated :: (Show e, Typeable e)
-          => (Producer ByteString IO () -> IO (Either e a))
-          -> (Producer ByteString IO () -> IO (Either e b))
-          ->  Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a,b))
-separated outfunc errfunc outprod errprod = 
-    conceit (outfunc outprod) (errfunc errprod)
-
-{-|
-   Defines how to decode a stream of bytes into text, including what to do
-   in presence of leftovers. Also defines how to manipulate each individual
-   line of text.  
- -}
-data LinePolicy e = LinePolicy ((FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> Producer ByteString IO () -> IO (Either e ()))
-
-instance Functor LinePolicy where
-  fmap f (LinePolicy func) = LinePolicy $ fmap (fmap (fmap (bimap f id))) func
-
-{-|
-    Constructs a 'LinePolicy'.
-
-    The second argument is a 'Siphon' value that specifies how to handle
-decoding failures. Passing @pure ()@ will ignore any leftovers. Passing
-@unwanted ()@ will abort the computation if leftovers remain.
-
-    The third argument is a function that modifies each individual line.
-    The line is represented as a 'Producer' to avoid having to keep it
-    wholly in memory. If you want the lines unmodified, just pass @id@.
-    Line prefixes are easy to add using applicative notation:
-
-  > (\x -> yield "prefix: " *> x)
-
- -}
-
-linePolicy :: (Show e, Typeable e)
-           => DecodingFunction ByteString Text 
-           -> Siphon ByteString e ()
-           -> (forall r. Producer T.Text IO r -> Producer T.Text IO r)
-           ->  LinePolicy e 
-linePolicy decoder lopo transform = LinePolicy $ \teardown producer -> do
-    let freeLines = transFreeT transform 
-                  . viewLines 
-                  . decoder
-                  $ producer
-        viewLines = getConst . T.lines Const
-    teardown freeLines >>= runSiphon lopo
-
--- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here
-combined :: (Show e, Typeable e) 
-         => LinePolicy e 
-         -> LinePolicy e 
-         -> (Producer T.Text IO () -> IO (Either e a))
-         -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)
-combined (LinePolicy fun1) (LinePolicy fun2) combinedConsumer prod1 prod2 = 
-    manyCombined [fmap ($prod1) fun1, fmap ($prod2) fun2] combinedConsumer 
-  where     
-    manyCombined :: (Show e, Typeable e) 
-                 => [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())]
-                 -> (Producer T.Text IO () -> IO (Either e a))
-                 -> IO (Either e a) 
-    manyCombined actions consumer = do
-        (outbox, inbox, seal) <- spawn' Single
-        mVar <- newMVar outbox
-        runConceit $ 
-            Conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal)
-            *>
-            Conceit (consumer (fromInput inbox) `finally` atomically seal)
-        where 
-        iterTLines mvar = iterT $ \textProducer -> do
-            -- the P.drain bit was difficult to figure out!!!
-            join $ withMVar mvar $ \output -> do
-                runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain)
-
-fromProducer :: Producer b IO r -> Pump b e ()
-fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) 
-
-fromSafeProducer :: Producer b (SafeT IO) r -> Pump b e ()
-fromSafeProducer producer = Pump $ safely $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) 
-
-fromFallibleProducer :: Producer b (ExceptT e IO) r -> Pump b e ()
-fromFallibleProducer producer = Pump $ \consumer -> runExceptT $ runEffect (mute producer >-> hoist lift consumer) 
-
-{-| 
-  Useful when we want to plug in a handler that does its work in the 'SafeT'
-transformer.
- -}
-safely :: (MFunctor t, C.MonadMask m, MonadIO m) 
-       => (t (SafeT m) l -> (SafeT m) x) 
-       ->  t m         l -> m         x 
-safely activity = runSafeT . activity . hoist lift 
-
-{-|
-    See the section /Non-lens decoding functions/ in the documentation for the
-@pipes-text@ package.  
--}
-type DecodingFunction bytes text = forall r. Producer bytes IO r -> Producer text IO (Producer bytes IO r)
-
-{-|
-    Constructs a 'Siphon' that works on encoded values out of a 'Siphon' that
-works on decoded values. 
-   
-    The two first arguments are a decoding function and a 'Siphon' that
-determines how to handle leftovers. Pass @pure id@ to ignore leftovers. Pass
-@unwanted id@ to abort the computation if leftovers remain.
- -}
-encoded :: (Show e, Typeable e) 
-        => DecodingFunction bytes text
-        -> Siphon bytes e (a -> b)
-        -> Siphon text  e a 
-        -> Siphon bytes e b
-encoded decoder policy activity = 
-    Unhalting $ \producer ->
-        runExceptT $ do
-            (a,leftovers) <- ExceptT $ unhalting activity $ decoder producer 
-            (f,r) <- ExceptT $ unhalting policy leftovers 
-            pure (f a,r)
-
-data WrappedError e = WrappedError e
-    deriving (Show, Typeable)
-
-instance (Show e, Typeable e) => Exception (WrappedError e)
-
-elideError :: (Show e, Typeable e) => IO (Either e a) -> IO a
-elideError action = action >>= either (throwIO . WrappedError) return
-
-revealError :: (Show e, Typeable e) => IO a -> IO (Either e a)  
-revealError action = catch (action >>= return . Right)
-                           (\(WrappedError e) -> return . Left $ e)   
-
-newtype Conceit e a = Conceit { runConceit :: IO (Either e a) }
-
-instance Functor (Conceit e) where
-  fmap f (Conceit x) = Conceit $ fmap (fmap f) x
-
-instance Bifunctor Conceit where
-  bimap f g (Conceit x) = Conceit $ liftM (bimap f g) x
-
-instance (Show e, Typeable e) => Applicative (Conceit e) where
-  pure = Conceit . pure . pure
-  Conceit fs <*> Conceit as =
-    Conceit . revealError $ 
-        uncurry ($) <$> concurrently (elideError fs) (elideError as)
-
-instance (Show e, Typeable e) => Alternative (Conceit e) where
-  empty = Conceit $ forever (threadDelay maxBound)
-  Conceit as <|> Conceit bs =
-    Conceit $ either id id <$> race as bs
-
-instance (Show e, Typeable e, Monoid a) => Monoid (Conceit e a) where
-   mempty = Conceit . pure . pure $ mempty
-   mappend c1 c2 = (<>) <$> c1 <*> c2
-
-conceit :: (Show e, Typeable e) 
-        => IO (Either e a)
-        -> IO (Either e b)
-        -> IO (Either e (a,b))
-conceit c1 c2 = runConceit $ (,) <$> Conceit c1 <*> Conceit c2
-
-{-| 
-      Works similarly to 'Control.Concurrent.Async.mapConcurrently' from the
-@async@ package, but if any of the computations fails with @e@, the others are
-immediately cancelled and the whole computation fails with @e@. 
- -}
-mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b))
-mapConceit f = revealError .  mapConcurrently (elideError . f)
-
-newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) }
-
-instance Functor (Pump b e) where
-  fmap f (Pump x) = Pump $ fmap (fmap (fmap f)) x
-
-instance Bifunctor (Pump b) where
-  bimap f g (Pump x) = Pump $ fmap (liftM  (bimap f g)) x
-
-instance (Show e, Typeable e) => Applicative (Pump b e) where
-  pure = Pump . pure . pure . pure
-  Pump fs <*> Pump as = 
-      Pump $ \consumer -> do
-          (outbox1,inbox1,seal1) <- spawn' Single
-          (outbox2,inbox2,seal2) <- spawn' Single
-          runConceit $ 
-              Conceit (runExceptT $ do
-                           r1 <- ExceptT $ (fs $ toOutput outbox1) 
-                                               `finally` atomically seal1
-                           r2 <- ExceptT $ (as $ toOutput outbox2) 
-                                               `finally` atomically seal2
-                           return $ r1 r2 
-                      )
-              <* 
-              Conceit (do
-                         (runEffect $
-                             (fromInput inbox1 >> fromInput inbox2) >-> consumer)
-                            `finally` atomically seal1
-                            `finally` atomically seal2
-                         runExceptT $ pure ()
-                      )
-
-instance (Show e, Typeable e, Monoid a) => Monoid (Pump b e a) where
-   mempty = Pump . pure . pure . pure $ mempty
-   mappend s1 s2 = (<>) <$> s1 <*> s2
-
-{-| 
-    A 'Siphon' represents a computation that completely drains a producer, but
-may fail early with an error of type @e@. 
-
-    'pure' creates a 'Siphon' that does nothing besides draining the
-'Producer'. 
-
-    '<*>' executes its arguments concurrently. The 'Producer' is forked so
-    that each argument receives its own copy of the data.
- -}
-data Siphon b e a = 
-         Trivial a 
-       | Unhalting (forall r. Producer b IO r -> IO (Either e (a,r)))
-       | Halting (Producer b IO () -> IO (Either e a))
-       deriving (Functor)
-
-instance Bifunctor (Siphon b) where
-  bimap f g s = case s of
-      Trivial a -> Trivial $ g a
-      Unhalting u -> Unhalting $ fmap (liftM  (bimap f (bimap g id))) u
-      Halting h -> Halting $ fmap (liftM  (bimap f g)) h
-
-instance (Show e, Typeable e) => Applicative (Siphon b e) where
-    pure = Trivial
-   
-    s1 <*> s2 = case (s1,s2) of
-        (Trivial f,_) -> fmap f s2
-        (_,Trivial a) -> fmap ($ a) s1
-        (_,_) -> bifurcate (halting s1) (halting s2)  
-      where 
-        bifurcate fs as =
-            Unhalting $ \producer -> do
-                (outbox1,inbox1,seal1) <- spawn' Single
-                (outbox2,inbox2,seal2) <- spawn' Single
-                runConceit $
-                    (,)
-                    <$>
-                    Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) 
-                                                            `finally` atomically seal1) 
-                                                            ((as $ fromInput inbox2) 
-                                                            `finally` atomically seal2) 
-                            )
-                    <*>
-                    Conceit ((fmap pure $ runEffect $ 
-                                  producer >-> P.tee (toOutput outbox1 >> P.drain) 
-                                           >->       (toOutput outbox2 >> P.drain))   
-                             `finally` atomically seal1 `finally` atomically seal2
-                            ) 
-
-runSiphon :: (Show e, Typeable e) => Siphon b e a  -> Producer b IO () -> IO (Either e a)
-runSiphon s = case s of 
-    h@(Halting _) -> halting $ Unhalting $ unhalting h 
-    _ -> halting s
-
--- This might return a computation that *doesn't* completely drain the
--- Producer.
-halting :: (Show e, Typeable e) => Siphon b e a  -> Producer b IO () -> IO (Either e a)
-halting s = case s of 
-    a@(Trivial _) -> halting $ Unhalting $ unhalting a
-    Unhalting u -> \producer -> liftM (fmap fst) $ u producer
-    Halting h -> h 
-
-unhalting :: (Show e, Typeable e) => Siphon b e a -> Producer b IO r -> IO (Either e (a,r))
-unhalting s = case s of 
-    Trivial a -> \producer -> do
-        r <- (runEffect $ producer >-> P.drain)
-        pure . pure $ (a,r)
-    Unhalting u -> u
-    Halting activity -> \producer -> do 
-        (outbox,inbox,seal) <- spawn' Single
-        runConceit $ 
-            (,) 
-            <$>
-            Conceit (activity (fromInput inbox) `finally` atomically seal)
-            <*>
-            Conceit ((fmap pure $ runEffect $ 
-                            producer >-> (toOutput outbox >> P.drain))
-                     `finally` atomically seal
-                    )
-
-instance (Show e, Typeable e, Monoid a) => Monoid (Siphon b e a) where
-   mempty = pure mempty
-   mappend s1 s2 = (<>) <$> s1 <*> s2
-
-fromConsumer :: Consumer b IO r -> Siphon b e ()
-fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer 
-
-fromSafeConsumer :: Consumer b (SafeT IO) r -> Siphon b e ()
-fromSafeConsumer consumer = siphon $ safely $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer 
-
-fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e ()
-fromFallibleConsumer consumer = siphon $ \producer -> runExceptT $ runEffect (hoist lift producer >-> mute consumer) 
-
-{-| 
-  Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'.
- -}
-fromParser :: Parser b IO (Either e a) -> Siphon b e a 
-fromParser parser = siphon $ Pipes.Parse.evalStateT parser 
-
-{-| 
-   Builds a 'Siphon' out of a computation that does something with
-   a 'Producer', but may fail with an error of type @e@.
-   
-   Even if the original computation doesn't completely drain the 'Producer',
-   the constructed 'Siphon' will.
--}
-siphon :: (Producer b IO () -> IO (Either e a))
-       -> Siphon b e a 
-siphon = Halting
-
-
-{-| 
-   Builds a 'Siphon' out of a computation that drains a 'Producer' completely,
-but may fail with an error of type @e@.
--}
-siphon' :: (forall r. Producer b IO r -> IO (Either e (a,r))) -> Siphon b e a 
-siphon' = Unhalting
-
-fromFold :: (Producer b IO () -> IO a) -> Siphon b e a 
-fromFold aFold = siphon $ fmap (fmap pure) $ aFold 
-
-{-| 
-   Builds a 'Siphon' out of a computation that folds a 'Producer' and drains it completely.
--}
-fromFold' :: (forall r. Producer b IO r -> IO (a,r)) -> Siphon b e a 
-fromFold' aFold = siphon' $ fmap (fmap pure) aFold
-
-fromFold'_ :: (forall r. Producer b IO r -> IO r) -> Siphon b e () 
-fromFold'_ aFold = fromFold' $ fmap (fmap ((,) ())) aFold
-
-{-|
-  Constructs a 'Siphon' that aborts the computation if the underlying
-'Producer' produces anything.
- -}
-unwanted :: a -> Siphon b b a
-unwanted a = Unhalting $ \producer -> do
-    n <- next producer  
-    return $ case n of 
-        Left r -> Right (a,r)
-        Right (b,_) -> Left b
-
-executePipeline :: PipingPolicy Void a -> CreatePipeline Void -> IO a 
-executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline
-
-
-{-|
-    Similar to 'executeFallibly', but instead of a single process it
-    executes a (possibly branching) pipeline of external processes. 
-
-    The 'PipingPolicy' argument views the pipeline as a synthetic process
-    for which @stdin@ is the @stdin@ of the first stage, @stdout@ is the
-    @stdout@ of the leftmost terminal stage among those closer to the root,
-    and @stderr@ is a combination of the @stderr@ streams of all the
-    stages.
-
-    The combined @stderr@ stream always has UTF-8 encoding.
-
-    This function has a limitation compared to the standard UNIX pipelines.
-    If a downstream process terminates early without error, the upstream
-    processes are not notified and keep going. There is no SIGPIPE-like
-    functionality, in other words. 
- -}
-executePipelineFallibly :: (Show e,Typeable e) => PipingPolicy e a -> CreatePipeline e -> IO (Either e a)
-executePipelineFallibly policy pipeline = case policy of 
-      PPNone a -> fmap (fmap (const a)) $
-           executePipelineInternal 
-                (\o _ -> mute $ pipeo o) 
-                (\i o _ -> mute $ pipeio i o) 
-                (\i _ -> mute $ pipei i) 
-                (\i _ -> mute $ pipei i) 
-                pipeline
-      PPOutput action -> do
-            (outbox, inbox, seal) <- spawn' Single
-            runConceit $  
-                (Conceit $ action $ fromInput inbox)
-                <* 
-                (Conceit $ executePipelineInternal 
-                                (\o _ -> pipeo o)
-                                (\i o _ -> mute $ pipeio i o) 
-                                (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ outbox)) 
-                                (\i _ -> mute $ pipei i)
-                                pipeline
-                           `finally` atomically seal
-                ) 
-      PPError action -> do
-            (eoutbox, einbox, eseal) <- spawn' Single
-            errf <- errorSiphonUTF8 <$> newMVar eoutbox
-            runConceit $  
-                (Conceit $ action $ fromInput einbox)
-                <*
-                (Conceit $ executePipelineInternal 
-                            (\o l -> mute $ pipeoe o (errf l)) 
-                            (\i o l -> mute $ pipeioe i o (errf l)) 
-                            (\i l -> mute $ pipeie i (errf l)) 
-                            (\i l -> mute $ pipeie i (errf l))
-                            pipeline
-                            `finally` atomically eseal)
-      PPOutputError action -> do
-            (outbox, inbox, seal) <- spawn' Single
-            (eoutbox, einbox, eseal) <- spawn' Single
-            errf <- errorSiphonUTF8 <$> newMVar eoutbox
-            runConceit $  
-                (Conceit $ action $ (fromInput inbox,fromInput einbox))
-                <* 
-                (Conceit $ executePipelineInternal 
-                                (\o l -> mute $ pipeoe o (errf l))
-                                (\i o l -> mute $ pipeioe i o (errf l)) 
-                                (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ outbox) (errf l)) 
-                                (\i l -> mute $ pipeie i (errf l))
-                                pipeline
-                           `finally` atomically seal `finally` atomically eseal
-                )
-      PPInput action -> do
-            (outbox, inbox, seal) <- spawn' Single
-            runConceit $  
-                (Conceit $ action (toOutput outbox,atomically seal))
-                <* 
-                (Conceit $ executePipelineInternal 
-                                (\o _ -> mute $ pipeio (fromProducer . fromInput $ inbox) o)
-                                (\i o _ -> mute $ pipeio i o) 
-                                (\i _ -> mute $ pipei i) 
-                                (\i _ -> mute $ pipei i) 
-                                pipeline
-                           `finally` atomically seal
-                )
-      PPInputOutput action -> do
-            (ioutbox, iinbox, iseal) <- spawn' Single
-            (ooutbox, oinbox, oseal) <- spawn' Single
-            runConceit $  
-                (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox))
-                <* 
-                (Conceit $ executePipelineInternal 
-                                (\o _ -> mute $ pipeio (fromProducer . fromInput $ iinbox) o)
-                                (\i o _ -> mute $ pipeio i o) 
-                                (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ ooutbox)) 
-                                (\i _ -> mute $ pipei i) 
-                                pipeline
-                           `finally` atomically iseal `finally` atomically oseal
-                )
-      PPInputError action -> do
-            (outbox, inbox, seal) <- spawn' Single
-            (eoutbox, einbox, eseal) <- spawn' Single
-            errf <- errorSiphonUTF8 <$> newMVar eoutbox
-            runConceit $  
-                (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))
-                <* 
-                (Conceit $ executePipelineInternal 
-                                (\o l -> mute $ pipeioe (fromProducer . fromInput $ inbox) o (errf l))
-                                (\i o l -> mute $ pipeioe i o (errf l)) 
-                                (\i l -> mute $ pipeie i (errf l)) 
-                                (\i l -> mute $ pipeie i (errf l)) 
-                                pipeline
-                           `finally` atomically seal `finally` atomically eseal
-                )
-      PPInputOutputError action -> do
-            (ioutbox, iinbox, iseal) <- spawn' Single
-            (ooutbox, oinbox, oseal) <- spawn' Single
-            (eoutbox, einbox, eseal) <- spawn' Single
-            errf <- errorSiphonUTF8 <$> newMVar eoutbox
-            runConceit $  
-                (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))
-                <* 
-                (Conceit $ executePipelineInternal 
-                                (\o l -> mute $ pipeioe (fromProducer . fromInput $ iinbox) o (errf l))
-                                (\i o l -> mute $ pipeioe i o (errf l)) 
-                                (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ ooutbox) (errf l)) 
-                                (\i l -> mute $ pipeie i (errf l))  
-                                pipeline
-                           `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal
-                )
-    where 
-      errorSiphonUTF8 :: MVar (Output ByteString) -> LinePolicy e -> Siphon ByteString e ()
-      errorSiphonUTF8 mvar (LinePolicy fun) = Halting $ fun iterTLines 
-        where     
-          iterTLines = iterT $ \textProducer -> do
-              -- the P.drain bit was difficult to figure out!!!
-              join $ withMVar mvar $ \output -> do
-                  runEffect $     (textProducer <* P.yield (singleton '\n')) 
-                              >->  P.map Data.Text.Encoding.encodeUtf8 
-                              >-> (toOutput output >> P.drain)
-
-mute :: Functor f => f a -> f ()
-mute = fmap (const ())
-
-{-|
-   An individual stage in a process pipeline. 
-   
-   The 'LinePolicy' field defines how to handle @stderr@ when @stderr@ is
-   piped. 
-   
-   Also required is a function that determines if the returned exit code
-   represents an error or not. This is necessary because some programs use
-   non-standard exit codes.
- -}
-data Stage e = Stage 
-           {
-             processDefinition :: CreateProcess 
-           , stderrLinePolicy :: LinePolicy e
-           , exitCodePolicy :: Int -> Maybe e
-           } deriving (Functor)
-
-{-|
-   Any stage beyond the first in a process pipeline. 
-
-   Incoming data is passed through the 'Pipe' before being fed to the process.
-
-   Use 'cat' (the identity 'Pipe' from 'Pipes') if no pre-processing is required.
- -}
-data SubsequentStage e = SubsequentStage (forall a.Pipe ByteString ByteString (ExceptT e IO) a) (Stage e) 
-
-instance Functor (SubsequentStage) where
-    fmap f (SubsequentStage bs s) = SubsequentStage (hoist (mapExceptT $ liftM (bimap f id)) bs) (fmap f s)
-
-data CreatePipeline e =  CreatePipeline (Stage e) (NonEmpty (Tree (SubsequentStage e))) deriving (Functor)
-
-{-|
-    Builds a (possibly branching) pipeline assuming that @stderr@ has the same
-encoding in all the stages, that no computation is perfored between the stages,
-and that any exit code besides 'ExitSuccess' in a stage actually represents an
-error.
- -}
-simplePipeline :: DecodingFunction ByteString Text -> CreateProcess -> NonEmpty (Tree (CreateProcess)) -> CreatePipeline String 
-simplePipeline decoder initial forest = CreatePipeline (simpleStage initial) (fmap (fmap simpleSubsequentStage) forest)   
-  where 
-     simpleStage cp = Stage cp simpleLinePolicy simpleErrorPolicy
-     simpleSubsequentStage = SubsequentStage P.cat . simpleStage
-     simpleLinePolicy = linePolicy decoder (pure ()) id
-     simpleErrorPolicy = Just . ("Exit failure: " ++) . show
-
-executePipelineInternal :: (Show e,Typeable e) 
-                        => (Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ())
-                        -> (Pump ByteString e () -> Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ())
-                        -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ())
-                        -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ())
-                        -> CreatePipeline e 
-                        -> IO (Either e ())
-executePipelineInternal ppinitial ppmiddle ppend ppend' (CreatePipeline (Stage cp lpol ecpol) a) =      
-    blende ecpol <$> executeFallibly (ppinitial (runNonEmpty ppend ppend' a) lpol) cp
-  where 
-    runTree ppend ppend' (Node (SubsequentStage pipe (Stage cp lpol ecpol)) forest) = case forest of
-        [] -> Halting $ \producer ->
-            blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ hoist lift producer >-> pipe) lpol) cp
-        c1 : cs -> Halting $ \producer ->
-           blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ hoist lift producer >-> pipe) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp
-
-    runNonEmpty ppend ppend' (b :| bs) = 
-        runTree ppend ppend' b <* Prelude.foldr (<*) (pure ()) (runTree ppend' ppend' <$> bs) 
-    
-    blende :: (Int -> Maybe e) -> Either e (ExitCode,()) -> Either e ()
-    blende f (Right (ExitFailure i,())) = case f i of
-        Nothing -> Right ()
-        Just e -> Left e
-    blende _ (Right (ExitSuccess,())) = Right () 
-    blende _ (Left e) = Left e
-
-{- $reexports
- 
-"System.Process" is re-exported for convenience.
-
--} 
-
+
+-- |
+-- This module contains helper functions and types built on top of
+-- "System.Process" and "Pipes".
+--
+-- They provide concurrent, streaming access to the inputs and outputs of
+-- system processes.
+--
+-- Error conditions not directly related to IO are made explicit
+-- in the types.
+--
+-- Regular 'Consumer's, 'Parser's from @pipes-parse@ and various folds can
+-- be used to consume the output streams of the external processes.
+--
+-----------------------------------------------------------------------------
+
+{-# LANGUAGE DeriveDataTypeable #-}
+{-# LANGUAGE DeriveFunctor #-}
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE ViewPatterns #-}
+
+module System.Process.Streaming ( 
+        -- * Execution
+          execute
+        , executeFallibly
+        -- * Piping Policies
+        , PipingPolicy
+        , nopiping
+        , pipeo
+        , pipee
+        , pipeoe
+        , pipeoec
+        , pipei
+        , pipeio
+        , pipeie
+        , pipeioe
+        , pipeioec
+
+        -- * Pumping bytes into stdin
+        , Pump (..)
+        , fromProducer
+        , fromSafeProducer
+        , fromFallibleProducer
+        -- * Siphoning bytes out of stdout/stderr
+        , Siphon
+        , siphon
+        , siphon'
+        , fromFold
+        , fromFold'
+        , fromFold'_
+        , fromConsumer
+        , fromSafeConsumer
+        , fromFallibleConsumer
+        , fromParser
+        , unwanted
+        , DecodingFunction
+        , encoded
+        -- * Line handling
+        , LinePolicy
+        , linePolicy
+        , tweakLines
+        -- * Pipelines
+        , executePipeline
+        , executePipelineFallibly
+--        , CreatePipeline (..)
+        --, simplePipeline
+        , Stage
+        , stage
+        , pipefail
+        , inbound
+       -- , SubsequentStage (..)
+        -- * Re-exports
+        -- $reexports
+        , module System.Process
+    ) where
+
+import Data.Maybe
+import Data.Bifunctor
+import Data.Functor.Identity
+import Data.Either
+import Data.Monoid
+import Data.Foldable
+import Data.Traversable
+import Data.Typeable
+import Data.Tree
+import Data.Text 
+import Data.Text.Encoding 
+import Data.Void
+import Data.List.NonEmpty
+import qualified Data.List.NonEmpty as N
+import Control.Applicative
+import Control.Monad
+import Control.Monad.Trans.Free
+import Control.Monad.Trans.Except
+import Control.Monad.Trans.State
+import Control.Monad.Trans.Writer.Strict
+import qualified Control.Monad.Catch as C
+import Control.Exception
+import Control.Concurrent
+import Control.Concurrent.Async
+import Pipes
+import qualified Pipes as P
+import qualified Pipes.Prelude as P
+import Pipes.Lift
+import Pipes.ByteString
+import Pipes.Parse
+import qualified Pipes.Text as T
+import Pipes.Concurrent
+import Pipes.Safe (SafeT, runSafeT)
+import System.IO
+import System.IO.Error
+import System.Process
+import System.Process.Lens
+import System.Exit
+
+execute :: PipingPolicy Void a -> CreateProcess -> IO (ExitCode,a)
+execute pp cprocess = either absurd id <$> executeFallibly pp cprocess
+
+{-|
+   Executes an external process. The standard streams are piped and consumed in
+a way defined by the 'PipingPolicy' argument. 
+
+   This function re-throws any 'IOException's it encounters.
+
+   If the consumption of the standard streams fails with @e@, the whole
+computation is immediately aborted and @e@ is returned. (An exception is not
+thrown in this case.).  
+
+   If an error @e@ or an exception happens, the external process is
+terminated.
+ -}
+executeFallibly :: PipingPolicy e a -> CreateProcess -> IO (Either e (ExitCode,a))
+executeFallibly pp record = case pp of
+      PPNone a -> executeInternal record nohandles $  
+          \() -> (return . Right $ a,return ())
+      PPOutput action -> executeInternal (record{std_out = CreatePipe}) handleso $
+          \h->(action (fromHandle h),hClose h) 
+      PPError action ->  executeInternal (record{std_err = CreatePipe}) handlese $
+          \h->(action (fromHandle h),hClose h)
+      PPOutputError action -> executeInternal (record{std_out = CreatePipe, std_err = CreatePipe}) handlesoe $
+          \(hout,herr)->(action (fromHandle hout,fromHandle herr),hClose hout `finally` hClose herr)
+      PPInput action -> executeInternal (record{std_in = CreatePipe}) handlesi $
+          \h -> (action (toHandle h, hClose h), return ())
+      PPInputOutput action -> executeInternal (record{std_in = CreatePipe,std_out = CreatePipe}) handlesio $
+          \(hin,hout) -> (action (toHandle hin,hClose hin,fromHandle hout), hClose hout)
+      PPInputError action -> executeInternal (record{std_in = CreatePipe,std_err = CreatePipe}) handlesie $
+          \(hin,herr) -> (action (toHandle hin,hClose hin,fromHandle herr), hClose herr)
+      PPInputOutputError action -> executeInternal (record{std_in = CreatePipe, std_out = CreatePipe, std_err = CreatePipe}) handlesioe $
+          \(hin,hout,herr) -> (action (toHandle hin,hClose hin,fromHandle hout,fromHandle herr), hClose hout `finally` hClose herr)
+
+executeInternal :: CreateProcess -> (forall m. Applicative m => (t -> m t) -> (Maybe Handle, Maybe Handle, Maybe Handle) -> m (Maybe Handle, Maybe Handle, Maybe Handle)) -> (t ->(IO (Either e a),IO ())) -> IO (Either e (ExitCode,a))
+executeInternal record somePrism allocator = mask $ \restore -> do
+    (min,mout,merr,phandle) <- createProcess record
+    case getFirst . getConst . somePrism (Const . First . Just) $ (min,mout,merr) of
+        Nothing -> 
+            throwIO (userError "stdin/stdout/stderr handle unwantedly null")
+            `finally`
+            terminateCarefully phandle 
+        Just t -> 
+            let (action,cleanup) = allocator t in
+            -- Handles must be closed *after* terminating the process, because a close
+            -- operation may block if the external process has unflushed bytes in the stream.
+            (restore (terminateOnError phandle action) `onException` terminateCarefully phandle) `finally` cleanup 
+
+exitCode :: (ExitCode,a) -> Either Int a
+exitCode (ec,a) = case ec of
+    ExitSuccess -> Right a 
+    ExitFailure i -> Left i
+
+terminateCarefully :: ProcessHandle -> IO ()
+terminateCarefully pHandle = do
+    mExitCode <- getProcessExitCode pHandle   
+    case mExitCode of 
+        Nothing -> terminateProcess pHandle  
+        Just _ -> return ()
+
+terminateOnError :: ProcessHandle 
+                 -> IO (Either e a)
+                 -> IO (Either e (ExitCode,a))
+terminateOnError pHandle action = do
+    result <- action
+    case result of
+        Left e -> do    
+            terminateCarefully pHandle
+            return $ Left e
+        Right r -> do 
+            exitCode <- waitForProcess pHandle 
+            return $ Right (exitCode,r)  
+
+{-|
+    A 'PipingPolicy' determines what standard streams will be piped and what to
+do with them.
+
+    The user doesn't need to manually set the 'std_in', 'std_out' and 'std_err'
+fields of the 'CreateProcess' record to 'CreatePipe', this is done
+automatically. 
+
+    A 'PipingPolicy' is parametrized by the type @e@ of errors that can abort
+the processing of the streams.
+ -}
+-- Knows that there is a stdin, stdout and a stderr,
+-- but doesn't know anything about file handlers or CreateProcess.
+data PipingPolicy e a = 
+      PPNone a
+    | PPOutput (Producer ByteString IO () -> IO (Either e a))
+    | PPError (Producer ByteString IO () -> IO (Either e a))
+    | PPOutputError ((Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a))
+    | PPInput ((Consumer ByteString IO (), IO ()) -> IO (Either e a))
+    | PPInputOutput ((Consumer ByteString IO (), IO (),Producer ByteString IO ()) -> IO (Either e a))
+    | PPInputError ((Consumer ByteString IO (), IO (), Producer ByteString IO ()) -> IO (Either e a))
+    | PPInputOutputError ((Consumer ByteString IO (),IO (),Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a))
+    deriving (Functor)
+
+instance Bifunctor PipingPolicy where
+  bimap f g pp = case pp of
+        PPNone a -> PPNone $ g a 
+        PPOutput action -> PPOutput $ fmap (fmap (bimap f g)) action
+        PPError action -> PPError $ fmap (fmap (bimap f g)) action
+        PPOutputError action -> PPOutputError $ fmap (fmap (bimap f g)) action
+        PPInput action -> PPInput $ fmap (fmap (bimap f g)) action
+        PPInputOutput action -> PPInputOutput $ fmap (fmap (bimap f g)) action
+        PPInputError action -> PPInputError $ fmap (fmap (bimap f g)) action
+        PPInputOutputError action -> PPInputOutputError $ fmap (fmap (bimap f g)) action
+
+{-|
+    Do not pipe any standard stream. 
+-}
+nopiping :: PipingPolicy e ()
+nopiping = PPNone ()
+
+{-|
+    Pipe @stdout@.
+-}
+pipeo :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a
+pipeo (runSiphon -> siphonout) = PPOutput $ siphonout
+
+{-|
+    Pipe @stderr@.
+-}
+pipee :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a
+pipee (runSiphon -> siphonout) = PPError $ siphonout
+
+{-|
+    Pipe @stdout@ and @stderr@.
+-}
+pipeoe :: (Show e,Typeable e) => Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b)
+pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = 
+    PPOutputError $ uncurry $ separated siphonout siphonerr  
+
+{-|
+    Pipe @stdout@ and @stderr@ and consume them combined as 'Text'.  
+-}
+pipeoec :: (Show e,Typeable e) => LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a
+pipeoec policy1 policy2 (runSiphon -> siphon) = 
+    PPOutputError $ uncurry $ combined policy1 policy2 siphon  
+
+{-|
+    Pipe @stdin@.
+-}
+pipei :: (Show e, Typeable e) => Pump ByteString e i -> PipingPolicy e i
+pipei (Pump feeder) = PPInput $ \(consumer,cleanup) -> feeder consumer `finally` cleanup
+
+{-|
+    Pipe @stdin@ and @stdout@.
+-}
+pipeio :: (Show e, Typeable e)
+        => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a)
+pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) ->
+        (conceit (feeder consumer `finally` cleanup) (siphonout producer))
+
+{-|
+    Pipe @stdin@ and @stderr@.
+-}
+pipeie :: (Show e, Typeable e)
+        => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a)
+pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) ->
+        (conceit (feeder consumer `finally` cleanup) (siphonerr producer))
+
+{-|
+    Pipe @stdin@, @stdout@ and @stderr@.
+-}
+pipeioe :: (Show e, Typeable e)
+        => Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b)
+pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $
+    \(consumer,cleanup,outprod,errprod) -> 
+             (conceit (feeder consumer `finally` cleanup) 
+                      (separated siphonout siphonerr outprod errprod))
+    where
+        flattenTuple (i, (a, b)) = (i,a,b)
+
+{-|
+    Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'.
+-}
+pipeioec :: (Show e, Typeable e)
+        => Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a)
+pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $
+    \(consumer,cleanup,outprod,errprod) -> 
+             (conceit (feeder consumer `finally` cleanup) 
+                      (combined policy1 policy2 siphon outprod errprod))
+
+separated :: (Show e, Typeable e)
+          => (Producer ByteString IO () -> IO (Either e a))
+          -> (Producer ByteString IO () -> IO (Either e b))
+          ->  Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e (a,b))
+separated outfunc errfunc outprod errprod = 
+    conceit (outfunc outprod) (errfunc errprod)
+
+{-|
+    A configuration parameter used in functions that combine lines from
+    multiple streams.
+ -}
+
+data LinePolicy e = LinePolicy 
+    {
+        teardown :: (forall r. Producer T.Text IO r -> Producer T.Text IO r)
+                 -> (FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) 
+                 -> Producer ByteString IO () -> IO (Either e ())
+    ,   lineTweaker :: forall r. Producer T.Text IO r -> Producer T.Text IO r
+    } 
+
+instance Functor LinePolicy where
+  fmap f (LinePolicy func lt) = LinePolicy (\x y z -> fmap (bimap f id) $ func x y z) lt
+
+
+{-|
+    Specifies a transformation that will be applied to each line of text,
+    represented as a 'Producer'.
+
+    Line prefixes are easy to add using applicative notation:
+
+  > (\x -> yield "prefix: " *> x)
+-}
+tweakLines :: (forall r. Producer T.Text IO r -> Producer T.Text IO r) -> LinePolicy e -> LinePolicy e 
+tweakLines lt' (LinePolicy tear lt) = LinePolicy tear (lt' . lt) 
+
+{-|
+    Constructs a 'LinePolicy' out of a 'DecodingFunction' and a 'Siphon'
+    that specifies how to handle decoding failures. Passing @pure ()@ as
+    the 'Siphon' will ignore any leftovers. Passing @unwanted ()@ will
+    abort the computation if leftovers remain.
+ -}
+linePolicy :: (Show e, Typeable e)
+           => DecodingFunction ByteString Text 
+           -> Siphon ByteString e ()
+           -> LinePolicy e 
+linePolicy decoder lopo = LinePolicy
+    (\tweaker teardown producer -> do
+        let freeLines = transFreeT tweaker 
+                      . viewLines 
+                      . decoder
+                      $ producer
+            viewLines = getConst . T.lines Const
+        teardown freeLines >>= runSiphon lopo)
+    id 
+
+-- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here
+combined :: (Show e, Typeable e) 
+         => LinePolicy e 
+         -> LinePolicy e 
+         -> (Producer T.Text IO () -> IO (Either e a))
+         -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)
+combined (LinePolicy fun1 twk1) (LinePolicy fun2 twk2) combinedConsumer prod1 prod2 = 
+    manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer 
+  where     
+    manyCombined :: (Show e, Typeable e) 
+                 => [(FreeT (Producer T.Text IO) IO (Producer ByteString IO ()) -> IO (Producer ByteString IO ())) -> IO (Either e ())]
+                 -> (Producer T.Text IO () -> IO (Either e a))
+                 -> IO (Either e a) 
+    manyCombined actions consumer = do
+        (outbox, inbox, seal) <- spawn' Single
+        mVar <- newMVar outbox
+        runConceit $ 
+            Conceit (mapConceit ($ iterTLines mVar) actions `finally` atomically seal)
+            *>
+            Conceit (consumer (fromInput inbox) `finally` atomically seal)
+        where 
+        iterTLines mvar = iterT $ \textProducer -> do
+            -- the P.drain bit was difficult to figure out!!!
+            join $ withMVar mvar $ \output -> do
+                runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain)
+
+fromProducer :: Producer b IO r -> Pump b e ()
+fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) 
+
+fromSafeProducer :: Producer b (SafeT IO) r -> Pump b e ()
+fromSafeProducer producer = Pump $ safely $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) 
+
+fromFallibleProducer :: Producer b (ExceptT e IO) r -> Pump b e ()
+fromFallibleProducer producer = Pump $ \consumer -> runExceptT $ runEffect (mute producer >-> hoist lift consumer) 
+
+{-| 
+  Useful when we want to plug in a handler that does its work in the 'SafeT'
+transformer.
+ -}
+safely :: (MFunctor t, C.MonadMask m, MonadIO m) 
+       => (t (SafeT m) l -> (SafeT m) x) 
+       ->  t m         l -> m         x 
+safely activity = runSafeT . activity . hoist lift 
+
+{-|
+    See the section /Non-lens decoding functions/ in the documentation for the
+@pipes-text@ package.  
+-}
+type DecodingFunction bytes text = forall r. Producer bytes IO r -> Producer text IO (Producer bytes IO r)
+
+{-|
+    Constructs a 'Siphon' that works on encoded values out of a 'Siphon' that
+works on decoded values. 
+   
+    The two first arguments are a decoding function and a 'Siphon' that
+determines how to handle leftovers. Pass @pure id@ to ignore leftovers. Pass
+@unwanted id@ to abort the computation if leftovers remain.
+ -}
+encoded :: (Show e, Typeable e) 
+        => DecodingFunction bytes text
+        -> Siphon bytes e (a -> b)
+        -> Siphon text  e a 
+        -> Siphon bytes e b
+encoded decoder policy activity = 
+    Unhalting $ \producer ->
+        runExceptT $ do
+            (a,leftovers) <- ExceptT $ unhalting activity $ decoder producer 
+            (f,r) <- ExceptT $ unhalting policy leftovers 
+            pure (f a,r)
+
+data WrappedError e = WrappedError e
+    deriving (Show, Typeable)
+
+instance (Show e, Typeable e) => Exception (WrappedError e)
+
+elideError :: (Show e, Typeable e) => IO (Either e a) -> IO a
+elideError action = action >>= either (throwIO . WrappedError) return
+
+revealError :: (Show e, Typeable e) => IO a -> IO (Either e a)  
+revealError action = catch (action >>= return . Right)
+                           (\(WrappedError e) -> return . Left $ e)   
+
+newtype Conceit e a = Conceit { runConceit :: IO (Either e a) } deriving Functor
+
+instance Bifunctor Conceit where
+  bimap f g (Conceit x) = Conceit $ liftM (bimap f g) x
+
+instance (Show e, Typeable e) => Applicative (Conceit e) where
+  pure = Conceit . pure . pure
+  Conceit fs <*> Conceit as =
+    Conceit . revealError $ 
+        uncurry ($) <$> concurrently (elideError fs) (elideError as)
+
+instance (Show e, Typeable e) => Alternative (Conceit e) where
+  empty = Conceit $ forever (threadDelay maxBound)
+  Conceit as <|> Conceit bs =
+    Conceit $ either id id <$> race as bs
+
+instance (Show e, Typeable e, Monoid a) => Monoid (Conceit e a) where
+   mempty = Conceit . pure . pure $ mempty
+   mappend c1 c2 = (<>) <$> c1 <*> c2
+
+conceit :: (Show e, Typeable e) 
+        => IO (Either e a)
+        -> IO (Either e b)
+        -> IO (Either e (a,b))
+conceit c1 c2 = runConceit $ (,) <$> Conceit c1 <*> Conceit c2
+
+{-| 
+      Works similarly to 'Control.Concurrent.Async.mapConcurrently' from the
+@async@ package, but if any of the computations fails with @e@, the others are
+immediately cancelled and the whole computation fails with @e@. 
+ -}
+mapConceit :: (Show e, Typeable e, Traversable t) => (a -> IO (Either e b)) -> t a -> IO (Either e (t b))
+mapConceit f = revealError .  mapConcurrently (elideError . f)
+
+newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor
+
+instance Bifunctor (Pump b) where
+  bimap f g (Pump x) = Pump $ fmap (liftM  (bimap f g)) x
+
+instance (Show e, Typeable e) => Applicative (Pump b e) where
+  pure = Pump . pure . pure . pure
+  Pump fs <*> Pump as = 
+      Pump $ \consumer -> do
+          (outbox1,inbox1,seal1) <- spawn' Single
+          (outbox2,inbox2,seal2) <- spawn' Single
+          runConceit $ 
+              Conceit (runExceptT $ do
+                           r1 <- ExceptT $ (fs $ toOutput outbox1) 
+                                               `finally` atomically seal1
+                           r2 <- ExceptT $ (as $ toOutput outbox2) 
+                                               `finally` atomically seal2
+                           return $ r1 r2 
+                      )
+              <* 
+              Conceit (do
+                         (runEffect $
+                             (fromInput inbox1 >> fromInput inbox2) >-> consumer)
+                            `finally` atomically seal1
+                            `finally` atomically seal2
+                         runExceptT $ pure ()
+                      )
+
+instance (Show e, Typeable e, Monoid a) => Monoid (Pump b e a) where
+   mempty = Pump . pure . pure . pure $ mempty
+   mappend s1 s2 = (<>) <$> s1 <*> s2
+
+{-| 
+    A 'Siphon' represents a computation that completely drains a producer, but
+may fail early with an error of type @e@. 
+
+    'pure' creates a 'Siphon' that does nothing besides draining the
+'Producer'. 
+
+    '<*>' executes its arguments concurrently. The 'Producer' is forked so
+    that each argument receives its own copy of the data.
+ -}
+data Siphon b e a = 
+         Trivial a 
+       | Unhalting (forall r. Producer b IO r -> IO (Either e (a,r)))
+       | Halting (Producer b IO () -> IO (Either e a))
+       deriving (Functor)
+
+instance Bifunctor (Siphon b) where
+  bimap f g s = case s of
+      Trivial a -> Trivial $ g a
+      Unhalting u -> Unhalting $ fmap (liftM  (bimap f (bimap g id))) u
+      Halting h -> Halting $ fmap (liftM  (bimap f g)) h
+
+instance (Show e, Typeable e) => Applicative (Siphon b e) where
+    pure = Trivial
+   
+    s1 <*> s2 = case (s1,s2) of
+        (Trivial f,_) -> fmap f s2
+        (_,Trivial a) -> fmap ($ a) s1
+        (_,_) -> bifurcate (halting s1) (halting s2)  
+      where 
+        bifurcate fs as =
+            Unhalting $ \producer -> do
+                (outbox1,inbox1,seal1) <- spawn' Single
+                (outbox2,inbox2,seal2) <- spawn' Single
+                runConceit $
+                    (,)
+                    <$>
+                    Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) 
+                                                            `finally` atomically seal1) 
+                                                            ((as $ fromInput inbox2) 
+                                                            `finally` atomically seal2) 
+                            )
+                    <*>
+                    Conceit ((fmap pure $ runEffect $ 
+                                  producer >-> P.tee (toOutput outbox1 >> P.drain) 
+                                           >->       (toOutput outbox2 >> P.drain))   
+                             `finally` atomically seal1 `finally` atomically seal2
+                            ) 
+
+runSiphon :: (Show e, Typeable e) => Siphon b e a  -> Producer b IO () -> IO (Either e a)
+runSiphon s = case s of 
+    h@(Halting _) -> halting $ Unhalting $ unhalting h 
+    _ -> halting s
+
+-- This might return a computation that *doesn't* completely drain the
+-- Producer.
+halting :: (Show e, Typeable e) => Siphon b e a  -> Producer b IO () -> IO (Either e a)
+halting s = case s of 
+    a@(Trivial _) -> halting $ Unhalting $ unhalting a
+    Unhalting u -> \producer -> liftM (fmap fst) $ u producer
+    Halting h -> h 
+
+unhalting :: (Show e, Typeable e) => Siphon b e a -> Producer b IO r -> IO (Either e (a,r))
+unhalting s = case s of 
+    Trivial a -> \producer -> do
+        r <- (runEffect $ producer >-> P.drain)
+        pure . pure $ (a,r)
+    Unhalting u -> u
+    Halting activity -> \producer -> do 
+        (outbox,inbox,seal) <- spawn' Single
+        runConceit $ 
+            (,) 
+            <$>
+            Conceit (activity (fromInput inbox) `finally` atomically seal)
+            <*>
+            Conceit ((fmap pure $ runEffect $ 
+                            producer >-> (toOutput outbox >> P.drain))
+                     `finally` atomically seal
+                    )
+
+instance (Show e, Typeable e, Monoid a) => Monoid (Siphon b e a) where
+   mempty = pure mempty
+   mappend s1 s2 = (<>) <$> s1 <*> s2
+
+fromConsumer :: Consumer b IO r -> Siphon b e ()
+fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer 
+
+fromSafeConsumer :: Consumer b (SafeT IO) r -> Siphon b e ()
+fromSafeConsumer consumer = siphon $ safely $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer 
+
+fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e ()
+fromFallibleConsumer consumer = siphon $ \producer -> runExceptT $ runEffect (hoist lift producer >-> mute consumer) 
+
+{-| 
+  Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'.
+ -}
+fromParser :: Parser b IO (Either e a) -> Siphon b e a 
+fromParser parser = siphon $ Pipes.Parse.evalStateT parser 
+
+{-| 
+   Builds a 'Siphon' out of a computation that does something with
+   a 'Producer', but may fail with an error of type @e@.
+   
+   Even if the original computation doesn't completely drain the 'Producer',
+   the constructed 'Siphon' will.
+-}
+siphon :: (Producer b IO () -> IO (Either e a))
+       -> Siphon b e a 
+siphon = Halting
+
+
+{-| 
+   Builds a 'Siphon' out of a computation that drains a 'Producer' completely,
+but may fail with an error of type @e@.
+-}
+siphon' :: (forall r. Producer b IO r -> IO (Either e (a,r))) -> Siphon b e a 
+siphon' = Unhalting
+
+{-| 
+    Useful in combination with 'Pipes.Text.toLazyM' from @pipes-text@ and
+    'Pipes.ByteString.toLazyM' from @pipes-bytestring@, when the user
+    wants to collect all the output. 
+-}
+fromFold :: (Producer b IO () -> IO a) -> Siphon b e a 
+fromFold aFold = siphon $ fmap (fmap pure) $ aFold 
+
+{-| 
+   Builds a 'Siphon' out of a computation that folds a 'Producer' and drains it completely.
+-}
+fromFold' :: (forall r. Producer b IO r -> IO (a,r)) -> Siphon b e a 
+fromFold' aFold = siphon' $ fmap (fmap pure) aFold
+
+fromFold'_ :: (forall r. Producer b IO r -> IO r) -> Siphon b e () 
+fromFold'_ aFold = fromFold' $ fmap (fmap ((,) ())) aFold
+
+{-|
+  Constructs a 'Siphon' that aborts the computation if the underlying
+'Producer' produces anything.
+ -}
+unwanted :: a -> Siphon b b a
+unwanted a = Unhalting $ \producer -> do
+    n <- next producer  
+    return $ case n of 
+        Left r -> Right (a,r)
+        Right (b,_) -> Left b
+
+executePipeline :: PipingPolicy Void a -> Tree (Stage Void) -> IO a 
+executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline
+
+
+{-|
+    Similar to 'executeFallibly', but instead of a single process it
+    executes a (possibly branching) pipeline of external processes. 
+
+    The 'PipingPolicy' argument views the pipeline as a synthetic process
+    for which @stdin@ is the @stdin@ of the first stage, @stdout@ is the
+    @stdout@ of the leftmost terminal stage among those closer to the root,
+    and @stderr@ is a combination of the @stderr@ streams of all the
+    stages.
+
+    The combined @stderr@ stream always has UTF-8 encoding.
+
+    This function has a limitation compared to the standard UNIX pipelines.
+    If a downstream process terminates early without error, the upstream
+    processes are not notified and keep going. There is no SIGPIPE-like
+    functionality, in other words. 
+ -}
+executePipelineFallibly :: (Show e,Typeable e) => PipingPolicy e a -> Tree (Stage e) -> IO (Either e a)
+executePipelineFallibly policy (Node (Stage cp lpol ecpol _) []) = case policy of
+          PPNone a -> blende ecpol <$> executeFallibly policy cp 
+          PPOutput action -> blende ecpol <$> executeFallibly policy cp 
+          PPError action -> do
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action $ fromInput einbox)
+                    <*
+                    (Conceit $ blende ecpol <$> executeFallibly (pipee (errf lpol)) cp `finally` atomically eseal)
+          PPOutputError action -> do 
+                (outbox, inbox, seal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action $ (fromInput inbox,fromInput einbox))
+                    <* 
+                    (Conceit $ blende ecpol <$> executeFallibly
+                                    (pipeoe (fromConsumer.toOutput $ outbox) (errf lpol)) cp
+                               `finally` atomically seal `finally` atomically eseal
+                    )
+          PPInput action -> blende ecpol <$> executeFallibly policy cp
+          PPInputOutput action -> blende ecpol <$> executeFallibly policy cp
+          PPInputError action -> do
+                (outbox, inbox, seal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))
+                    <* 
+                    (Conceit $ blende ecpol <$> executeFallibly
+                                    (pipeie (fromProducer . fromInput $ inbox) (errf lpol)) cp
+                               `finally` atomically seal `finally` atomically eseal
+                    )
+          PPInputOutputError action -> do
+                (ioutbox, iinbox, iseal) <- spawn' Single
+                (ooutbox, oinbox, oseal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))
+                    <* 
+                    (Conceit $ blende ecpol <$> executeFallibly
+                                    (pipeioe (fromProducer . fromInput $ iinbox) 
+                                            (fromConsumer . toOutput $ ooutbox) 
+                                            (errf lpol) 
+                                    )
+                                    cp
+                               `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal
+                    )
+executePipelineFallibly policy (Node s (s':ss)) = 
+      let pipeline = CreatePipeline s $ s' :| ss 
+      in case policy of 
+          PPNone a -> fmap (fmap (const a)) $
+               executePipelineInternal 
+                    (\o _ -> mute $ pipeo o) 
+                    (\i o _ -> mute $ pipeio i o) 
+                    (\i _ -> mute $ pipei i) 
+                    (\i _ -> mute $ pipei i) 
+                    pipeline
+          PPOutput action -> do
+                (outbox, inbox, seal) <- spawn' Single
+                runConceit $  
+                    (Conceit $ action $ fromInput inbox)
+                    <* 
+                    (Conceit $ executePipelineInternal 
+                                    (\o _ -> pipeo o)
+                                    (\i o _ -> mute $ pipeio i o) 
+                                    (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ outbox)) 
+                                    (\i _ -> mute $ pipei i)
+                                    pipeline
+                               `finally` atomically seal
+                    ) 
+          PPError action -> do
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action $ fromInput einbox)
+                    <*
+                    (Conceit $ executePipelineInternal 
+                                (\o l -> mute $ pipeoe o (errf l)) 
+                                (\i o l -> mute $ pipeioe i o (errf l)) 
+                                (\i l -> mute $ pipeie i (errf l)) 
+                                (\i l -> mute $ pipeie i (errf l))
+                                pipeline
+                                `finally` atomically eseal)
+          PPOutputError action -> do
+                (outbox, inbox, seal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action $ (fromInput inbox,fromInput einbox))
+                    <* 
+                    (Conceit $ executePipelineInternal 
+                                    (\o l -> mute $ pipeoe o (errf l))
+                                    (\i o l -> mute $ pipeioe i o (errf l)) 
+                                    (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ outbox) (errf l)) 
+                                    (\i l -> mute $ pipeie i (errf l))
+                                    pipeline
+                               `finally` atomically seal `finally` atomically eseal
+                    )
+          PPInput action -> do
+                (outbox, inbox, seal) <- spawn' Single
+                runConceit $  
+                    (Conceit $ action (toOutput outbox,atomically seal))
+                    <* 
+                    (Conceit $ executePipelineInternal 
+                                    (\o _ -> mute $ pipeio (fromProducer . fromInput $ inbox) o)
+                                    (\i o _ -> mute $ pipeio i o) 
+                                    (\i _ -> mute $ pipei i) 
+                                    (\i _ -> mute $ pipei i) 
+                                    pipeline
+                               `finally` atomically seal
+                    )
+          PPInputOutput action -> do
+                (ioutbox, iinbox, iseal) <- spawn' Single
+                (ooutbox, oinbox, oseal) <- spawn' Single
+                runConceit $  
+                    (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox))
+                    <* 
+                    (Conceit $ executePipelineInternal 
+                                    (\o _ -> mute $ pipeio (fromProducer . fromInput $ iinbox) o)
+                                    (\i o _ -> mute $ pipeio i o) 
+                                    (\i _ -> mute $ pipeio i (fromConsumer . toOutput $ ooutbox)) 
+                                    (\i _ -> mute $ pipei i) 
+                                    pipeline
+                               `finally` atomically iseal `finally` atomically oseal
+                    )
+          PPInputError action -> do
+                (outbox, inbox, seal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))
+                    <* 
+                    (Conceit $ executePipelineInternal 
+                                    (\o l -> mute $ pipeioe (fromProducer . fromInput $ inbox) o (errf l))
+                                    (\i o l -> mute $ pipeioe i o (errf l)) 
+                                    (\i l -> mute $ pipeie i (errf l)) 
+                                    (\i l -> mute $ pipeie i (errf l)) 
+                                    pipeline
+                               `finally` atomically seal `finally` atomically eseal
+                    )
+          PPInputOutputError action -> do
+                (ioutbox, iinbox, iseal) <- spawn' Single
+                (ooutbox, oinbox, oseal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' Single
+                errf <- errorSiphonUTF8 <$> newMVar eoutbox
+                runConceit $  
+                    (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))
+                    <* 
+                    (Conceit $ executePipelineInternal 
+                                    (\o l -> mute $ pipeioe (fromProducer . fromInput $ iinbox) o (errf l))
+                                    (\i o l -> mute $ pipeioe i o (errf l)) 
+                                    (\i l -> mute $ pipeioe i (fromConsumer . toOutput $ ooutbox) (errf l)) 
+                                    (\i l -> mute $ pipeie i (errf l))  
+                                    pipeline
+                               `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal
+                    )
+
+errorSiphonUTF8 :: MVar (Output ByteString) -> LinePolicy e -> Siphon ByteString e ()
+errorSiphonUTF8 mvar (LinePolicy fun twk) = Halting $ fun twk iterTLines 
+  where     
+    iterTLines = iterT $ \textProducer -> do
+        -- the P.drain bit was difficult to figure out!!!
+        join $ withMVar mvar $ \output -> do
+            runEffect $     (textProducer <* P.yield (singleton '\n')) 
+                        >->  P.map Data.Text.Encoding.encodeUtf8 
+                        >-> (toOutput output >> P.drain)
+
+mute :: Functor f => f a -> f ()
+mute = fmap (const ())
+
+{-|
+   An individual stage in a process pipeline. 
+ -}
+data Stage e = Stage 
+           {
+             processDefinition' :: CreateProcess 
+           , stderrLinePolicy' :: LinePolicy e
+           , exitCodePolicy' :: ExitCode -> Either e ()
+           , inbound' :: forall r. Producer ByteString IO r -> Producer ByteString (ExceptT e IO) r 
+           } 
+
+instance Functor (Stage) where
+    fmap f (Stage a b c d) = Stage a (fmap f b) (bimap f id . c) (hoist (mapExceptT $ liftM (bimap f id)) . d)
+
+{-|
+    Builds a 'Stage' out of a 'LinePolicy' that specifies how to handle
+    @stderr@ when piped, a function that determines whether an
+    'ExitCode' represents an error (some programs return non-standard exit
+    codes) and a process definition. 
+-}
+stage :: LinePolicy e -> (ExitCode -> Either e ()) -> CreateProcess -> Stage e       
+stage lp ec cp = Stage cp lp ec (hoist lift) 
+
+{-|
+   Applies a transformation to the stream of bytes flowing into a stage from previous stages.
+
+   This function is ignored for first stages.
+-}
+inbound :: (forall r. Producer ByteString (ExceptT e IO) r -> Producer ByteString (ExceptT e IO) r)
+        -> Stage e -> Stage e 
+inbound f (Stage a b c d) = Stage a b c (f . d)
+
+data CreatePipeline e =  CreatePipeline (Stage e) (NonEmpty (Tree (Stage e))) deriving (Functor)
+
+executePipelineInternal :: (Show e,Typeable e) 
+                        => (Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ())
+                        -> (Pump ByteString e () -> Siphon ByteString e () -> LinePolicy e -> PipingPolicy e ())
+                        -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ())
+                        -> (Pump ByteString e () -> LinePolicy e -> PipingPolicy e ())
+                        -> CreatePipeline e 
+                        -> IO (Either e ())
+executePipelineInternal ppinitial ppmiddle ppend ppend' (CreatePipeline (Stage cp lpol ecpol _) a) =      
+    blende ecpol <$> executeFallibly (ppinitial (runNonEmpty ppend ppend' a) lpol) cp
+  where 
+    runTree ppend ppend' (Node (Stage cp lpol ecpol pipe) forest) = case forest of
+        [] -> Halting $ \producer ->
+            blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ pipe producer) lpol) cp
+        c1 : cs -> Halting $ \producer ->
+           blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ pipe producer) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp
+
+    runNonEmpty ppend ppend' (b :| bs) = 
+        runTree ppend ppend' b <* Prelude.foldr (<*) (pure ()) (runTree ppend' ppend' <$> bs) 
+    
+blende :: (ExitCode -> Either e ()) -> Either e (ExitCode,a) -> Either e a
+blende f r = r >>= \(ec,a) -> f ec *> pure a
+
+{-|
+  Converts any 'ExitFailure' to the left side of an 'Either'. 
+-}
+pipefail :: ExitCode -> Either Int ()
+pipefail ec = case ec of
+    ExitSuccess -> Right ()
+    ExitFailure i -> Left i
+
+{- $reexports
+ 
+"System.Process" is re-exported for convenience.
+
+-} 
+
diff --git a/tests/test.hs b/tests/test.hs
--- a/tests/test.hs
+++ b/tests/test.hs
@@ -1,255 +1,323 @@
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE ViewPatterns #-}
-
-module Main where 
-
-import Test.Tasty
-import Test.Tasty.HUnit
-
-import Data.Bifunctor
-import Data.Monoid
-import Data.Foldable
-import Data.List.NonEmpty
-import Data.ByteString
-import Data.ByteString.Lazy as BL
-import Data.Text.Lazy as TL
-import Data.Typeable
-import Data.Tree
-import qualified Data.Attoparsec.Text as A
-import Control.Applicative
-import Control.Monad
-import Control.Monad.Trans.Except
-import Control.Lens (view)
-import Control.Concurrent.Async
-import Pipes
-import qualified Pipes.ByteString as B
-import qualified Pipes.Prelude as P
-import qualified Pipes.Parse as P
-import qualified Pipes.Attoparsec as P
-import qualified Pipes.Text as T
-import qualified Pipes.Text.Encoding as T
-import qualified Pipes.Text.IO as T
-import qualified Pipes.Group as G
-import qualified Pipes.Safe as S
-import qualified Pipes.Safe.Prelude as S
-import System.IO
-import System.IO.Error
-import System.Exit
-import System.Directory
-import System.Process.Streaming
-
-main = defaultMain tests
-
-tests :: TestTree
-tests = testGroup "Tests" 
-            [ testCollectStdoutStderrAsByteString
-            , testFeedStdinCollectStdoutAsText  
-            , testCombinedStdoutStderr
-            , testInterruptExecution 
-            , testFailIfAnythingShowsInStderr 
-            , testTwoTextParsersInParallel  
-            , testCountWords 
-            , testBasicPipeline
-            , testBranchingPipeline 
-            ]
-
--------------------------------------------------------------------------------
-testCollectStdoutStderrAsByteString :: TestTree
-testCollectStdoutStderrAsByteString = testCase "collectStdoutStderrAsByteString" $ do
-    r <- collectStdoutStderrAsByteString
-    case r of
-        (ExitSuccess,("ooo\nppp\n","eee\nffff\n")) -> return ()
-        _ -> assertFailure "oops"
-
-collectStdoutStderrAsByteString :: IO (ExitCode,(BL.ByteString,BL.ByteString))
-collectStdoutStderrAsByteString = execute
-    (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM))
-    (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }")
-
-
--------------------------------------------------------------------------------
-testFeedStdinCollectStdoutAsText  :: TestTree
-testFeedStdinCollectStdoutAsText = testCase "feedStdinCollectStdoutAsText" $ do
-    r <- feedStdinCollectStdoutAsText
-    case r of
-        (ExitSuccess,((),"aaaaaa\naaaaa")) -> return ()
-        _ -> assertFailure "oops"
-
-feedStdinCollectStdoutAsText :: IO (ExitCode, ((), Text))
-feedStdinCollectStdoutAsText = execute
-    (pipeio (fromProducer $ yield "aaaaaa\naaaaa")
-            (encoded T.decodeIso8859_1 (pure id) $ fromFold T.toLazyM))
-    (shell "cat")
-
--------------------------------------------------------------------------------
-
-testCombinedStdoutStderr :: TestTree
-testCombinedStdoutStderr = testCase "testCombinedStdoutStderr"  $ do
-    r <- combinedStdoutStderr 
-    case r of 
-        (ExitSuccess,TL.lines -> ls) -> do
-            assertEqual "line count" (Prelude.length ls) 4
-            assertBool "expected lines" $ 
-                getAll $ foldMap (All . flip Prelude.elem ls) $
-                    [ "ooo"
-                    , "ppp"
-                    , "errprefix: eee"
-                    , "errprefix: ffff"
-                    ]
-        _ -> assertFailure "oops"
-
-combinedStdoutStderr :: IO (ExitCode,TL.Text)
-combinedStdoutStderr = execute
-    (pipeoec (linePolicy T.decodeIso8859_1 (pure ()) id)
-             (linePolicy T.decodeIso8859_1 (pure ()) annotate)    
-             (fromFold T.toLazyM))
-    (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }")
-  where
-    annotate x = P.yield "errprefix: " *> x  
-
--------------------------------------------------------------------------------
-
-testInterruptExecution :: TestTree
-testInterruptExecution = localOption (mkTimeout $ 5*(10^6)) $
-    testCase "interruptExecution" $ do
-        r <- interruptExecution
-        case r of
-            Left "interrupted" -> return ()
-            _ -> assertFailure "oops"
-
-interruptExecution :: IO (Either String (ExitCode,()))
-interruptExecution = executeFallibly
-    (pipeo . siphon $ \_ -> runExceptT . throwE $ "interrupted")
-    (shell "sleep 100s")
-
--------------------------------------------------------------------------------
-
-testFailIfAnythingShowsInStderr :: TestTree
-testFailIfAnythingShowsInStderr = localOption (mkTimeout $ 5*(10^6)) $
-    testCase "failIfAnythingShowsInStderr" $ do
-        r <- failIfAnythingShowsInStderr 
-        case r of
-            Left "morestuff\n" -> return ()
-            _ -> assertFailure "oops"
-
-failIfAnythingShowsInStderr :: IO (Either T.ByteString (ExitCode,()))
-failIfAnythingShowsInStderr = executeFallibly
-    (pipee (unwanted ()))
-    (shell "{ echo morestuff 1>&2 ; sleep 100s ; }")
-
--------------------------------------------------------------------------------
-
-testTwoTextParsersInParallel  :: TestTree
-testTwoTextParsersInParallel  = testCase "twoTextParsersInParallel" $ do
-    r <- twoTextParsersInParallel
-    case r of 
-        Right (ExitSuccess,("ooooooo","aaaaaa")) -> return ()
-        _ -> assertFailure "oops"
-
-parseChars :: Char -> A.Parser [Char] 
-parseChars c = fmap mconcat $ 
-    many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c))
-        
-parser1 = parseChars 'o'
-
-parser2 = parseChars 'a'
-
-twoTextParsersInParallel :: IO (Either String (ExitCode,([Char], [Char])))
-twoTextParsersInParallel = executeFallibly
-    (pipeo (encoded T.decodeIso8859_1 (pure id) $ 
-                (,) <$> adapt parser1 <*> adapt parser2))
-    (shell "{ echo ooaaoo ; echo aaooaoa; }")
-  where
-    adapt p = fromParser $ do
-        r <- P.parse p
-        return $ case r of
-            Just (Right r') -> Right r'
-            _ -> Left "parse error"
-
--------------------------------------------------------------------------------
-
-testCountWords :: TestTree
-testCountWords = testCase "testCountWords" $ do
-    r <- countWords 
-    case r of 
-        (ExitSuccess,3) -> return ()                   
-        _ -> assertFailure "oops"
-
-countWords :: IO (ExitCode,Int)
-countWords = execute
-    (pipeo (encoded T.decodeIso8859_1 (pure id) $
-                fromFold $ P.sum . G.folds const () (const 1) . view T.words))
-    (shell "{ echo aaa ; echo bbb ; echo ccc ; }")
-
--------------------------------------------------------------------------------
-testBasicPipeline :: TestTree
-testBasicPipeline = testCase "basicPipeline" $ do
-    r <- basicPipeline 
-    case r of 
-        Right ((),"aaaccc\n") -> return ()                   
-        _ -> assertFailure "oops"
-
-basicPipeline :: IO (Either String ((),BL.ByteString))
-basicPipeline =  executePipelineFallibly 
-    (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") 
-            (fromFold B.toLazyM)) 
-    (simplePipeline T.decodeUtf8 (shell "grep aaa") (pure . pure $ shell "grep ccc"))
-
--------------------------------------------------------------------------------
-
-testBranchingPipeline :: TestTree
-testBranchingPipeline = testCase "branchingPipeline" $ do
-    exists <- doesFileExist branchingPipelineFile
-    when exists $ removeFile branchingPipelineFile
-    r <- branchingPipeline 
-    case r of 
-        ("ppp\v","eee\nffff\n") -> return ()                   
-        _ -> assertFailure "oops"
-    fileContents <- withFile branchingPipelineFile ReadMode  $ \hIn -> do
-        B.toLazyM $ B.fromHandle hIn 
-    assertBool "file contexts" $ BL.isPrefixOf "yyy" fileContents 
-
-branchingPipelineFile :: String 
-branchingPipelineFile = "dist/test/process-streaming-pipeline-text.txt"
-
-branchingPipeline :: IO (BL.ByteString, BL.ByteString)
-branchingPipeline = executePipeline
-    (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) 
-    (CreatePipeline rootStage . fromList $ 
-        [ Node branch1 [pure terminalStage1] , Node branch2 [pure terminalStage2] ] )
-  where
-    succStage = SubsequentStage (P.map (Data.ByteString.map succ))
-
-    rootStage :: (Show e, Typeable e) => Stage e
-    rootStage = Stage (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ;  echo ffff 1>&2 ; }")
-                      (linePolicy T.decodeIso8859_1 (pure ()) id)                 
-                      (\_ -> Nothing)
-
-    branch1 :: (Show e, Typeable e) => SubsequentStage e
-    branch1 = SubsequentStage cat $
-        Stage (shell "grep ooo")
-              (linePolicy T.decodeIso8859_1 (pure ()) id)                 
-              (\_ -> Nothing)
-
-    branch2 :: (Show e, Typeable e) => SubsequentStage e
-    branch2 = SubsequentStage cat $
-        Stage (shell "grep xxx")
-              (linePolicy T.decodeIso8859_1 (pure ()) id)                 
-              (\_ -> Nothing)
-
-    terminalStage1 :: (Show e, Typeable e) => SubsequentStage e
-    terminalStage1 = succStage $
-        Stage (shell "tr -d b")
-              (linePolicy T.decodeIso8859_1 (pure ()) id)                 
-              (\_ -> Nothing)
-
-    terminalStage2 :: (Show e, Typeable e) => SubsequentStage e
-    terminalStage2 = succStage $
-        Stage (shell $ "cat > " ++ branchingPipelineFile)
-              (linePolicy T.decodeIso8859_1 (pure ()) id)                 
-              (\_ -> Nothing)
-
-
-
-
-
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ViewPatterns #-}
+
+module Main where 
+
+import Test.Tasty
+import Test.Tasty.HUnit
+
+import Data.Bifunctor
+import Data.Monoid
+import Data.Foldable
+import Data.List.NonEmpty
+import Data.ByteString
+import Data.ByteString.Lazy as BL
+import Data.Text.Lazy as TL
+import Data.Typeable
+import Data.Tree
+import qualified Data.Attoparsec.Text as A
+import Control.Applicative
+import Control.Monad
+import Control.Monad.Trans.Except
+import Control.Lens (view)
+import Control.Concurrent.Async
+import Pipes
+import qualified Pipes.ByteString as B
+import qualified Pipes.Prelude as P
+import qualified Pipes.Parse as P
+import qualified Pipes.Attoparsec as P
+import qualified Pipes.Text as T
+import qualified Pipes.Text.Encoding as T
+import qualified Pipes.Text.IO as T
+import qualified Pipes.Group as G
+import qualified Pipes.Safe as S
+import qualified Pipes.Safe.Prelude as S
+import System.IO
+import System.IO.Error
+import System.Exit
+import System.Directory
+import System.Process.Streaming
+
+main = defaultMain tests
+
+tests :: TestTree
+tests = testGroup "Tests" 
+            [ testCollectStdoutStderrAsByteString
+            , testFeedStdinCollectStdoutAsText  
+            , testCombinedStdoutStderr
+            , testInterruptExecution 
+            , testFailIfAnythingShowsInStderr 
+            , testTwoTextParsersInParallel  
+            , testCountWords 
+            , testSingletonPipeline 
+            , testBasicPipeline
+            , testBranchingPipeline 
+            , testDrainageDeadlock
+            , testAlternatingWithCombined 
+            ]
+
+-------------------------------------------------------------------------------
+testCollectStdoutStderrAsByteString :: TestTree
+testCollectStdoutStderrAsByteString = testCase "collectStdoutStderrAsByteString" $ do
+    r <- collectStdoutStderrAsByteString
+    case r of
+        (ExitSuccess,("ooo\nppp\n","eee\nffff\n")) -> return ()
+        _ -> assertFailure "oops"
+
+collectStdoutStderrAsByteString :: IO (ExitCode,(BL.ByteString,BL.ByteString))
+collectStdoutStderrAsByteString = execute
+    (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM))
+    (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }")
+
+
+-------------------------------------------------------------------------------
+testFeedStdinCollectStdoutAsText  :: TestTree
+testFeedStdinCollectStdoutAsText = testCase "feedStdinCollectStdoutAsText" $ do
+    r <- feedStdinCollectStdoutAsText
+    case r of
+        (ExitSuccess,((),"aaaaaa\naaaaa")) -> return ()
+        _ -> assertFailure "oops"
+
+feedStdinCollectStdoutAsText :: IO (ExitCode, ((), Text))
+feedStdinCollectStdoutAsText = execute
+    (pipeio (fromProducer $ yield "aaaaaa\naaaaa")
+            (encoded T.decodeIso8859_1 (pure id) $ fromFold T.toLazyM))
+    (shell "cat")
+
+-------------------------------------------------------------------------------
+
+testCombinedStdoutStderr :: TestTree
+testCombinedStdoutStderr = testCase "testCombinedStdoutStderr"  $ do
+    r <- combinedStdoutStderr 
+    case r of 
+        (ExitSuccess,TL.lines -> ls) -> do
+            assertEqual "line count" (Prelude.length ls) 4
+            assertBool "expected lines" $ 
+                getAll $ foldMap (All . flip Prelude.elem ls) $
+                    [ "ooo"
+                    , "ppp"
+                    , "errprefix: eee"
+                    , "errprefix: ffff"
+                    ]
+        _ -> assertFailure "oops"
+
+combinedStdoutStderr :: IO (ExitCode,TL.Text)
+combinedStdoutStderr = execute
+    (pipeoec (linePolicy T.decodeIso8859_1 (pure ()))
+             (tweakLines annotate $ linePolicy T.decodeIso8859_1 (pure ()))    
+             (fromFold T.toLazyM))
+    (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }")
+  where
+    annotate x = P.yield "errprefix: " *> x  
+
+-------------------------------------------------------------------------------
+
+testInterruptExecution :: TestTree
+testInterruptExecution = localOption (mkTimeout $ 5*(10^6)) $
+    testCase "interruptExecution" $ do
+        r <- interruptExecution
+        case r of
+            Left "interrupted" -> return ()
+            _ -> assertFailure "oops"
+
+interruptExecution :: IO (Either String (ExitCode,()))
+interruptExecution = executeFallibly
+    (pipeo . siphon $ \_ -> runExceptT . throwE $ "interrupted")
+    (shell "sleep 100s")
+
+-------------------------------------------------------------------------------
+
+testFailIfAnythingShowsInStderr :: TestTree
+testFailIfAnythingShowsInStderr = localOption (mkTimeout $ 5*(10^6)) $
+    testCase "failIfAnythingShowsInStderr" $ do
+        r <- failIfAnythingShowsInStderr 
+        case r of
+            Left "morestuff\n" -> return ()
+            _ -> assertFailure "oops"
+
+failIfAnythingShowsInStderr :: IO (Either T.ByteString (ExitCode,()))
+failIfAnythingShowsInStderr = executeFallibly
+    (pipee (unwanted ()))
+    (shell "{ echo morestuff 1>&2 ; sleep 100s ; }")
+
+-------------------------------------------------------------------------------
+
+testTwoTextParsersInParallel  :: TestTree
+testTwoTextParsersInParallel  = testCase "twoTextParsersInParallel" $ do
+    r <- twoTextParsersInParallel
+    case r of 
+        Right (ExitSuccess,("ooooooo","aaaaaa")) -> return ()
+        _ -> assertFailure "oops"
+
+parseChars :: Char -> A.Parser [Char] 
+parseChars c = fmap mconcat $ 
+    many (A.notChar c) *> A.many1 (some (A.char c) <* many (A.notChar c))
+        
+parser1 = parseChars 'o'
+
+parser2 = parseChars 'a'
+
+twoTextParsersInParallel :: IO (Either String (ExitCode,([Char], [Char])))
+twoTextParsersInParallel = executeFallibly
+    (pipeo (encoded T.decodeIso8859_1 (pure id) $ 
+                (,) <$> adapt parser1 <*> adapt parser2))
+    (shell "{ echo ooaaoo ; echo aaooaoa; }")
+  where
+    adapt p = fromParser $ do
+        r <- P.parse p
+        return $ case r of
+            Just (Right r') -> Right r'
+            _ -> Left "parse error"
+
+-------------------------------------------------------------------------------
+
+testCountWords :: TestTree
+testCountWords = testCase "testCountWords" $ do
+    r <- countWords 
+    case r of 
+        (ExitSuccess,3) -> return ()                   
+        _ -> assertFailure "oops"
+
+countWords :: IO (ExitCode,Int)
+countWords = execute
+    (pipeo (encoded T.decodeIso8859_1 (pure id) $
+                fromFold $ P.sum . G.folds const () (const 1) . view T.words))
+    (shell "{ echo aaa ; echo bbb ; echo ccc ; }")
+
+-------------------------------------------------------------------------------
+testSingletonPipeline :: TestTree
+testSingletonPipeline = testCase "singletonPipeline" $ do
+    r <- singletonPipeline 
+    case r of
+        (Right ("ooo\nppp\n","eee\nffff\n")) -> return ()
+        _ -> assertFailure "oops"
+
+singletonPipeline :: IO (Either Int (BL.ByteString,BL.ByteString))
+singletonPipeline =  executePipelineFallibly 
+    (pipeoe (fromFold B.toLazyM) 
+            (fromFold B.toLazyM)) 
+    (pure $ stage (linePolicy T.decodeUtf8 (pure ())) pipefail $ 
+         shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }"
+    )     
+
+-------------------------------------------------------------------------------
+testBasicPipeline :: TestTree
+testBasicPipeline = testCase "basicPipeline" $ do
+    r <- basicPipeline 
+    case r of 
+        Right ((),"aaaccc\n") -> return ()                   
+        _ -> assertFailure "oops"
+
+basicPipeline :: IO (Either Int ((),BL.ByteString))
+basicPipeline =  executePipelineFallibly 
+    (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") 
+            (fromFold B.toLazyM)) 
+    (fmap (stage (linePolicy T.decodeUtf8 (pure ())) pipefail) $   
+        Node (shell "grep aaa") [Node (shell "grep ccc") []] )
+
+-------------------------------------------------------------------------------
+
+testBranchingPipeline :: TestTree
+testBranchingPipeline = testCase "branchingPipeline" $ do
+    exists <- doesFileExist branchingPipelineFile
+    when exists $ removeFile branchingPipelineFile
+    r <- branchingPipeline 
+    case r of 
+        (Right ("ppp\v","eee\nffff\n")) -> return ()                   
+        _ -> assertFailure "oops"
+    fileContents <- withFile branchingPipelineFile ReadMode  $ \hIn -> do
+        B.toLazyM $ B.fromHandle hIn 
+    assertBool "file contexts" $ BL.isPrefixOf "yyy" fileContents 
+
+branchingPipelineFile :: String 
+branchingPipelineFile = "dist/test/process-streaming-pipeline-text.txt"
+
+branchingPipeline :: IO (Either Int (BL.ByteString, BL.ByteString))
+branchingPipeline = executePipelineFallibly
+    (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) 
+    (Node rootStage 
+        [ Node branch1 [Node terminalStage1 []]
+        , Node branch2 [Node terminalStage2 []]
+        ] 
+    )
+  where
+    succStage = P.map (Data.ByteString.map succ)
+
+    rootStage :: Stage Int 
+    rootStage = stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+                      pipefail
+                      (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ;  echo ffff 1>&2 ; }")
+
+    branch1 :: Stage Int 
+    branch1 = stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+                    pipefail
+                    (shell "grep ooo")
+    branch2 :: Stage Int 
+    branch2 = stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+                    pipefail
+                    (shell "grep xxx")
+
+    terminalStage1 :: Stage Int 
+    terminalStage1 = inbound (\p -> p >-> succStage) $
+        stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+              pipefail
+              (shell "tr -d b")
+
+    terminalStage2 :: Stage Int
+    terminalStage2 = inbound (\p -> p >-> succStage) $
+        stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+              pipefail
+              (shell $ "cat > " ++ branchingPipelineFile)
+
+-------------------------------------------------------------------------------
+
+testDrainageDeadlock :: TestTree
+testDrainageDeadlock = localOption (mkTimeout $ 20*(10^6)) $
+    testCase "drainageDeadlock" $ do
+        execute nopiping $ shell "chmod u+x tests/alternating.sh"
+        r <- drainageDeadlock
+        case r of
+            (ExitSuccess,((),())) -> return ()
+            _ -> assertFailure "oops"
+
+-- A bug caused some streams not to be drained, and this caused problems
+-- due to full output buffers.
+drainageDeadlock :: IO (ExitCode,((),()))
+drainageDeadlock = execute
+    (pipeoe (pure ()) (fromFold $ \producer -> next producer >> pure ()))
+    (proc "tests/alternating.sh" [])
+
+
+-------------------------------------------------------------------------------
+
+testAlternatingWithCombined :: TestTree
+testAlternatingWithCombined = localOption (mkTimeout $ 20*(10^6)) $
+    testCase "testAlternatingWithCombined" $ do
+        execute nopiping $ shell "chmod u+x tests/alternating.sh"
+        r <- alternatingWithCombined  
+        case r of 
+            (ExitSuccess,80000) -> return ()
+            _ -> assertFailure "oops"
+        r <- alternatingWithCombined2  
+        case r of 
+            (ExitSuccess,(80000,80000)) -> return ()
+            _ -> assertFailure "oops"
+
+alternatingWithCombined :: IO (ExitCode,Integer)
+alternatingWithCombined = execute
+    (pipeoec lp lp countLines)
+    (proc "tests/alternating.sh" [])
+  where
+    lp = linePolicy T.decodeIso8859_1 (pure ()) 
+    countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines
+
+
+alternatingWithCombined2 :: IO (ExitCode,(Integer,Integer))
+alternatingWithCombined2 = execute
+    (pipeoec lp lp $ (,) <$> countLines <*> countLines)
+    (proc "tests/alternating.sh" [])
+  where
+    lp = linePolicy T.decodeIso8859_1 (pure ()) 
+    countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines
+
+