diff --git a/CHANGELOG b/CHANGELOG
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,3 +1,17 @@
+0.7.0.0
+-------
+- Changed signature of toLines to make it more consistent with that of encoded
+  (breaking change).
+- Stricter signatures of some fromConsumer... functions (breaking change).
+- Pump constructor now hidden (breaking change).
+- Added new fromConsumer... functions for Consumers that don't stop.
+- Functions to throw exceptions for encoding errors (unwantedX, leftoverX,
+  _leftoverX). Also a new LeftoverException.
+- Re-exported more decoding functions from pipes-text.
+- New module System.Process.Streaming.Extended with some experimental
+  functions.
+- New module System.Process.Streaming.Internal. 
+
 0.6.9.0
 -------
 - fromLazyBytes, intoLazyBytes, intoLazyText.
diff --git a/process-streaming.cabal b/process-streaming.cabal
--- a/process-streaming.cabal
+++ b/process-streaming.cabal
@@ -1,5 +1,5 @@
 name:          process-streaming
-version:       0.6.9.0
+version:       0.7.0.0
 license:       BSD3
 license-file:  LICENSE
 data-files:    
@@ -19,9 +19,11 @@
     default-language: Haskell2010
     hs-source-dirs: src
     exposed-modules: 
-        System.Process.Lens
-        System.Process.Streaming
+        System.Process.Streaming.Extended
+        System.Process.Streaming.Internal
         System.Process.Streaming.Tutorial
+        System.Process.Streaming
+        System.Process.Lens
     other-modules: 
     ghc-options: -Wall -threaded -O2
     build-depends:         
@@ -67,7 +69,7 @@
       , pipes-bytestring >= 2.1.0 && < 2.2
       , pipes-text >= 0.0.0.10 && < 0.0.2
       , text >= 0.11.2 && < 1.2.1
-      , pipes-concurrency >= 2.0.2 && < 3
+      , pipes-concurrency >= 2.0.3 && < 3
       , pipes-safe >= 2.2.0 && < 3
       , pipes-parse >=3.0.1 && <3.1
       , exceptions >= 0.6.0 && < 1.0
@@ -101,7 +103,7 @@
       , 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-concurrency >= 2.0.3 && < 3
       , pipes-safe >= 2.2.0 && < 3
       , pipes-parse >=3.0.1 && <3.1
       , exceptions >= 0.6.0 && < 1
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
@@ -16,6 +16,7 @@
 
 {-# LANGUAGE DeriveFunctor #-}
 {-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE DeriveDataTypeable #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE ViewPatterns #-}
@@ -41,7 +42,7 @@
         , pipeioec
 
         -- * Pumping bytes into stdin
-        , Pump (..)
+        , Pump
         , fromProducer
         , fromProducerM
         , fromSafeProducer
@@ -56,26 +57,33 @@
         , fromFold
         , fromFold'
         , fromFold'_
-        , fromFoldl
-        , fromFoldlIO
-        , fromFoldlM
         , fromConsumer
+        , fromConsumer'
         , fromConsumerM
+        , fromConsumerM'
         , fromSafeConsumer
         , fromFallibleConsumer
         , fromParser
         , fromParserM 
-        , unwanted
+        , fromFoldl
+        , fromFoldlIO
+        , fromFoldlM
         , intoLazyBytes
         , intoLazyText 
+        , unwanted
         , DecodingFunction
         , encoded
         , SiphonOp (..)
-        -- * Line handling
+        -- * Handling lines
         , Lines
         , toLines
         , tweakLines
         , prefixLines
+        -- * Throwing exceptions
+        , unwantedX
+        , LeftoverException (..)
+        , leftoverX
+        , _leftoverX
         -- * Pipelines
         , executePipelineFallibly
         , executePipeline
@@ -88,38 +96,27 @@
         -- $reexports
         , module System.Process
         , T.decodeUtf8 
-        -- * Deprecated
-        -- $deprecated
-        , PipingPolicy
-        , LinePolicy
-        , linePolicy 
+        , T.decodeAscii 
+        , T.decodeIso8859_1
     ) where
 
-import Data.Maybe
 import qualified Data.ByteString.Lazy as BL
-import Data.Bifunctor
-import Data.Functor.Identity
 import Data.Functor.Contravariant
 import Data.Functor.Contravariant.Divisible
-import Data.Either
 import Data.Monoid
 import Data.Foldable
-import Data.Traversable
+import Data.Typeable
 import Data.Tree
-import Data.String
 import qualified Data.Text.Lazy as TL
 import Data.Text 
-import Data.Text.Encoding 
+import Data.Text.Encoding hiding (decodeUtf8)
 import Data.Void
 import Data.List.NonEmpty
-import qualified Data.List.NonEmpty as N
 import Control.Applicative
 import Control.Applicative.Lift
 import Control.Monad
 import Control.Monad.Trans.Free hiding (Pure)
 import Control.Monad.Trans.Except
-import Control.Monad.Trans.State
-import Control.Monad.Trans.Writer.Strict
 import qualified Control.Foldl as L
 import Control.Exception
 import Control.Concurrent
@@ -127,7 +124,6 @@
 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
@@ -135,14 +131,15 @@
 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
 
+import System.Process.Streaming.Internal
+
 {-|
-  A simplified version of 'executeFallibly' for when the error type is `Void`.
-  Note however that this function may still throw exceptions.
+  A simplified version of 'executeFallibly' for when the error type unifies
+  with `Void`.  Note however that this function may still throw exceptions.
  -}
 execute :: Piping Void a -> CreateProcess -> IO (ExitCode,a)
 execute pp cprocess = either absurd id <$> executeFallibly pp cprocess
@@ -216,8 +213,8 @@
                 -> (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
+    (mi,mout,merr,phandle) <- createProcess record
+    case getFirst . getConst . somePrism (Const . First . Just) $ (mi,mout,merr) of
         Nothing -> 
             throwIO (userError "stdin/stdout/stderr handle unwantedly null")
             `finally`
@@ -228,10 +225,6 @@
             -- 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
@@ -254,83 +247,6 @@
             return $ Right (exitCode,r)  
 
 {-|
-    A 'Piping' 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 'Piping' 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 Piping 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)
-
-
-{-| 
-    'first' is useful to massage errors.
--}
-instance Bifunctor Piping 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 :: Piping e ()
@@ -340,27 +256,27 @@
     Pipe @stdout@.
 -}
 pipeo :: Siphon ByteString e a -> Piping e a
-pipeo (runSiphon -> siphonout) = PPOutput $ siphonout
+pipeo (runSiphonDumb -> siphonout) = PPOutput $ siphonout
 
 {-|
     Pipe @stderr@.
 -}
 pipee :: Siphon ByteString e a -> Piping e a
-pipee (runSiphon -> siphonout) = PPError $ siphonout
+pipee (runSiphonDumb -> siphonout) = PPError $ siphonout
 
 {-|
     Pipe @stdout@ and @stderr@.
 -}
 pipeoe :: Siphon ByteString e a -> Siphon ByteString e b -> Piping e (a,b)
-pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = 
+pipeoe (runSiphonDumb -> siphonout) (runSiphonDumb -> siphonerr) = 
     PPOutputError $ uncurry $ separated siphonout siphonerr  
 
 {-|
     Pipe @stdout@ and @stderr@ and consume them combined as 'Text'.  
 -}
 pipeoec :: Lines e -> Lines e -> Siphon Text e a -> Piping e a
-pipeoec policy1 policy2 (runSiphon -> siphon) = 
-    PPOutputError $ uncurry $ combined policy1 policy2 siphon  
+pipeoec policy1 policy2 (runSiphonDumb -> s) = 
+    PPOutputError $ uncurry $ combined policy1 policy2 s
 
 {-|
     Pipe @stdin@.
@@ -372,21 +288,21 @@
     Pipe @stdin@ and @stdout@.
 -}
 pipeio :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a)
-pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) ->
+pipeio (Pump feeder) (runSiphonDumb -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) ->
         (conceit (feeder consumer `finally` cleanup) (siphonout producer))
 
 {-|
     Pipe @stdin@ and @stderr@.
 -}
 pipeie :: Pump ByteString e i -> Siphon ByteString e a -> Piping e (i,a)
-pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) ->
+pipeie (Pump feeder) (runSiphonDumb -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) ->
         (conceit (feeder consumer `finally` cleanup) (siphonerr producer))
 
 {-|
     Pipe @stdin@, @stdout@ and @stderr@.
 -}
 pipeioe :: Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> Piping e (i,a,b)
-pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $
+pipeioe (Pump feeder) (runSiphonDumb -> siphonout) (runSiphonDumb -> siphonerr) = fmap flattenTuple $ PPInputOutputError $
     \(consumer,cleanup,outprod,errprod) -> 
              (conceit (feeder consumer `finally` cleanup) 
                       (separated siphonout siphonerr outprod errprod))
@@ -397,10 +313,10 @@
     Pipe @stdin@, @stdout@ and @stderr@, consuming the last two combined as 'Text'.
 -}
 pipeioec :: Pump ByteString e i -> Lines e -> Lines e -> Siphon Text e a -> Piping e (i,a)
-pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $
+pipeioec (Pump feeder) policy1 policy2 (runSiphonDumb -> s) = PPInputOutputError $
     \(consumer,cleanup,outprod,errprod) -> 
              (conceit (feeder consumer `finally` cleanup) 
-                      (combined policy1 policy2 siphon outprod errprod))
+                      (combined policy1 policy2 s outprod errprod))
 
 separated :: (Producer ByteString IO () -> IO (Either e a))
           -> (Producer ByteString IO () -> IO (Either e b))
@@ -408,70 +324,6 @@
 separated outfunc errfunc outprod errprod = 
     conceit (outfunc outprod) (errfunc errprod)
 
--- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here
-combined :: Lines e 
-         -> Lines e 
-         -> (Producer T.Text IO () -> IO (Either e a))
-         -> Producer ByteString IO () -> Producer ByteString IO () -> IO (Either e a)
-combined (Lines fun1 twk1) (Lines fun2 twk2) combinedConsumer prod1 prod2 = 
-    manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer 
-  where     
-    manyCombined :: [(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)
-
-
-newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor
-
-
-{-| 
-    'first' is useful to massage errors.
--}
-instance Bifunctor (Pump b) where
-  bimap f g (Pump x) = Pump $ fmap (liftM  (bimap f g)) x
-
-instance 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 (Monoid a) => Monoid (Pump b e a) where
-   mempty = Pump . pure . pure . pure $ mempty
-   mappend s1 s2 = (<>) <$> s1 <*> s2
-
-instance IsString b => IsString (Pump b e ()) where 
-   fromString = fromProducer . P.yield . fromString 
-
 fromProducer :: Producer b IO r -> Pump b e ()
 fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) 
 
@@ -493,170 +345,7 @@
 fromLazyBytes :: BL.ByteString -> Pump ByteString e () 
 fromLazyBytes = fromProducer . fromLazy 
 
-{-| 
-    A 'Siphon' represents a computation that completely drains a producer, but
-may fail early with an error of type @e@. 
-
- -}
-newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor)
-
-
-{-| 
-    '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.
--}
-instance Applicative (Siphon b e) where
-    pure a = Siphon (pure a)
-    (Siphon fa) <*> (Siphon a) = Siphon (fa <*> a)
-
-data Siphon_ b e a = 
-         Exhaustive (forall r. Producer b IO r -> IO (Either e (a,r)))
-       | Nonexhaustive (Producer b IO () -> IO (Either e a))
-       deriving (Functor)
-
-
-instance Applicative (Siphon_ b e) where
-    pure a = Exhaustive $ \producer -> do
-        r <- runEffect (producer >-> P.drain)
-        pure (pure (a,r))
-    s1 <*> s2 = bifurcate (nonexhaustive s1) (nonexhaustive s2)  
-      where 
-        bifurcate fs as = Exhaustive $ \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 ((runEffect $ 
-                              producer >-> P.tee (toOutput outbox1 >> P.drain) 
-                                       >->       (toOutput outbox2 >> P.drain))   
-                          `finally` atomically seal1 `finally` atomically seal2
-                         ) 
-
-nonexhaustive :: Siphon_ b e a -> Producer b IO () -> IO (Either e a)
-nonexhaustive (Exhaustive e) = \producer -> liftM (fmap fst) (e producer)
-nonexhaustive (Nonexhaustive u) = u
-
-exhaustive :: Siphon_ b e a -> Producer b IO r -> IO (Either e (a,r))
-exhaustive s = case s of 
-    Exhaustive e -> e
-    Nonexhaustive activity -> \producer -> do 
-        (outbox,inbox,seal) <- spawn' Single
-        runConceit $ 
-            (,) 
-            <$>
-            Conceit (activity (fromInput inbox) `finally` atomically seal)
-            <*>
-            _Conceit (runEffect (producer >-> (toOutput outbox >> P.drain)) 
-                      `finally` atomically seal
-                     )
-
-runSiphon :: Siphon b e a -> Producer b IO () -> IO (Either e a)
-runSiphon (Siphon (unLift -> s)) = nonexhaustive $ case s of 
-    Exhaustive _ -> s
-    Nonexhaustive _ -> Exhaustive (exhaustive s)
-
-
-instance Bifunctor (Siphon_ b) where
-  bimap f g s = case s of
-      Exhaustive u -> Exhaustive $ fmap (liftM  (bimap f (bimap g id))) u
-      Nonexhaustive h -> Nonexhaustive $ fmap (liftM  (bimap f g)) h
-
-{-| 
-    'first' is useful to massage errors.
--}
-instance Bifunctor (Siphon b) where
-  bimap f g (Siphon s) = Siphon $ case s of
-      Pure a -> Pure (g a)
-      Other o -> Other (bimap f g o)
-
-instance (Monoid a) => Monoid (Siphon b e a) where
-   mempty = pure mempty
-   mappend s1 s2 = (<>) <$> s1 <*> s2
-
-newtype SiphonOp e a b = SiphonOp { getSiphonOp :: Siphon b e a } 
-
--- | 'contramap' carn turn a 'SiphonOp' for bytes into a 'SiphonOp' for text.
-instance Contravariant (SiphonOp e a) where
-    contramap f (SiphonOp (Siphon s)) = SiphonOp . Siphon $ case s of
-        Pure p -> Pure p
-        Other o -> Other $ case o of
-            Exhaustive e -> Exhaustive $ \producer ->
-                e $ producer >-> P.map f
-            Nonexhaustive ne -> Nonexhaustive $ \producer ->
-                ne $ producer >-> P.map f
-
--- | 'divide' builds a 'SiphonOp' for a composite out of the 'SiphonOp's
--- for the parts.
-instance Monoid a => Divisible (SiphonOp e a) where
-    divide divider siphonOp1 siphonOp2 = contramap divider . SiphonOp $ 
-        (getSiphonOp (contramap fst siphonOp1)) 
-        `mappend`
-        (getSiphonOp (contramap snd siphonOp2))
-    conquer = SiphonOp (pure mempty)
-
--- | 'choose' builds a 'SiphonOp' for a sum out of the 'SiphonOp's
--- for the branches.
-instance Monoid a => Decidable (SiphonOp e a) where
-    choose chooser (SiphonOp s1) (SiphonOp s2) = 
-        contramap chooser . SiphonOp $ 
-            (contraPipeMapL s1) 
-            `mappend`
-            (contraPipeMapR s2)
-      where
-        contraPipeMapL (Siphon s) = Siphon $ case s of
-            Pure p -> Pure p
-            Other o -> Other $ case o of
-                Exhaustive e -> Exhaustive $ \producer ->
-                    e $ producer >-> allowLefts
-                Nonexhaustive ne -> Nonexhaustive $ \producer ->
-                    ne $ producer >-> allowLefts
-        contraPipeMapR (Siphon s) = Siphon $ case s of
-            Pure p -> Pure p
-            Other o -> Other $ case o of
-                Exhaustive e -> Exhaustive $ \producer ->
-                    e $ producer >-> allowRights
-                Nonexhaustive ne -> Nonexhaustive $ \producer ->
-                    ne $ producer >-> allowRights
-        allowLefts = do
-            e <- await
-            case e of 
-                Left l -> Pipes.yield l >> allowLefts
-                Right _ -> allowLefts
-        allowRights = do
-            e <- await
-            case e of 
-                Right r -> Pipes.yield r >> allowRights
-                Left _ -> allowRights
-    lose f = SiphonOp . Siphon . Other . Nonexhaustive $ \producer -> do
-        n <- next producer  
-        return $ case n of 
-            Left () -> Right mempty
-            Right (b,_) -> Right (absurd (f b))
-
-allowLefts :: Monad m => Pipe (Either b a) b m r
-allowLefts = do
-    e <- await
-    case e of 
-        Left l -> Pipes.yield l >> allowLefts
-        Right _ -> allowLefts
                                            
-allowRights :: Monad m => Pipe (Either b a) a m r
-allowRights = do
-    e <- await
-    case e of 
-        Right r -> Pipes.yield r >> allowRights
-        Left _ -> allowRights
-                                           
 intoLazyBytes :: Siphon ByteString e BL.ByteString 
 intoLazyBytes = fromFold toLazyM  
 
@@ -724,27 +413,42 @@
 fromFoldlM whittle aFoldM = siphon' $ \producer -> 
     whittle $ L.impurely P.foldM' aFoldM (hoist liftIO producer)
 
-fromConsumer :: Consumer b IO r -> Siphon b e ()
-fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer 
+fromConsumer :: Consumer b IO () -> Siphon b e ()
+fromConsumer consumer = fromFold $ \producer -> runEffect $ producer >-> consumer 
 
+fromConsumer' :: Consumer b IO Void -> Siphon b e ()
+fromConsumer' consumer = fromFold'_$ \producer -> runEffect $ producer >-> fmap absurd consumer 
+
 fromConsumerM :: MonadIO m 
               => (m () -> IO (Either e a)) 
-              -> Consumer b m r 
+              -> Consumer b m () 
               -> Siphon b e a
-fromConsumerM whittle consumer = siphon $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> mute consumer 
+fromConsumerM whittle consumer = siphon $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> consumer 
 
-fromSafeConsumer :: Consumer b (SafeT IO) r -> Siphon b e ()
-fromSafeConsumer = fromConsumerM (fmap pure . runSafeT)
+fromConsumerM' :: MonadIO m 
+               => (forall r. m r -> IO (Either e (a,r))) 
+               -> Consumer b m Void
+               -> Siphon b e a
+fromConsumerM' whittle consumer = siphon' $ \producer -> whittle $ runEffect $ (hoist liftIO producer) >-> fmap absurd consumer 
 
-fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e ()
-fromFallibleConsumer = fromConsumerM runExceptT
+fromSafeConsumer :: Consumer b (SafeT IO) Void -> Siphon b e ()
+fromSafeConsumer = fromConsumerM' (fmap (\r -> Right ((),r)) . runSafeT)
 
+fromFallibleConsumer :: Consumer b (ExceptT e IO) Void -> Siphon b e ()
+fromFallibleConsumer = fromConsumerM' (fmap (fmap (\r -> ((), r))) . runExceptT)
+
 {-| 
   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 
-
+fromParser parser = siphon' $ \producer -> drainage $ Pipes.Parse.runStateT parser producer
+  where
+    drainage m = do 
+        (a,leftovers) <- m
+        r <- runEffect (leftovers >-> P.drain)
+        case a of
+            Left e -> return (Left e)
+            Right a' -> return (Right (a',r)) 
 
 {-| 
   Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'.
@@ -760,8 +464,8 @@
         return (a,r)
 
 {-|
-  Constructs a 'Siphon' that aborts the computation if the underlying
-'Producer' produces anything.
+  Constructs a 'Siphon' that aborts the computation with an explicit error
+  if the underlying 'Producer' produces anything.
  -}
 unwanted :: a -> Siphon b b a
 unwanted a = siphon' $ \producer -> do
@@ -771,6 +475,54 @@
         Right (b,_) -> Left b
 
 {-|
+  Like 'unwanted', but throws an exception instead of using the explicit
+  error type.
+-}
+unwantedX :: Exception ex => (b -> ex) -> a -> Siphon b e a
+unwantedX f a = siphon' $ \producer -> do
+    n <- next producer  
+    case n of 
+        Left r -> return $ Right (a,r)
+        Right (b,_) -> throwIO (f b)
+
+{-|
+  Exception that carries a message and a sample of the leftover data.  
+-}
+data LeftoverException b = LeftoverException String b deriving (Typeable)
+
+instance (Typeable b) => Exception (LeftoverException b)
+
+instance (Typeable b) => Show (LeftoverException b) where
+    show (LeftoverException msg _) = 
+        "[Leftovers of type " ++ typeName (Proxy::Data.Typeable.Proxy b) ++ "]" ++ msg'
+      where
+        typeName p = showsTypeRep (typeRep p) []
+        msg' = case msg of
+                   [] -> []
+                   _ -> " " ++ msg
+
+{-|
+    Throws 'LeftoverException' if any data comes out of the underlying
+    producer, and returns 'id' otherwise.
+-}
+leftoverX :: String 
+          -- ^ Error message
+          -> Siphon ByteString e (a -> a)
+leftoverX msg = unwantedX (LeftoverException msg') id
+    where 
+      msg' = "leftoverX." ++ case msg of
+         "" -> ""
+         _ -> " " ++ msg
+
+{-|
+    Like 'leftoverX', but doesn't take an error message.
+-}
+_leftoverX :: Siphon ByteString e (a -> a)
+_leftoverX = unwantedX (LeftoverException msg) id
+    where 
+      msg = "_leftoverX."
+
+{-|
     See the section /Non-lens decoding functions/ in the documentation for the
 @pipes-text@ package.  
 -}
@@ -783,9 +535,10 @@
 encoded :: DecodingFunction bytes text
         -- ^ A decoding function.
         -> Siphon bytes e (a -> b)
-        -- ^ A 'Siphon' that determines how to handle leftovers. 
-        -- Pass @pure id@ to ignore leftovers. Pass
-        -- @unwanted id@ to abort the computation if leftovers remain.
+        -- ^ A 'Siphon' that determines how to handle decoding leftovers.
+        -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort
+        -- the computation with an explicit error if leftovers remain. Pass
+        -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain.
         -> Siphon text  e a 
         -> Siphon bytes e b
 encoded decoder (Siphon (unLift -> policy)) (Siphon (unLift -> activity)) = 
@@ -797,23 +550,65 @@
         pure (f a,r)
 
 
-{-|
-    A configuration parameter used in functions that combine lines of text from
-    multiple streams.
- -}
+newtype SiphonOp e a b = SiphonOp { getSiphonOp :: Siphon b e a } 
 
-data Lines e = Lines 
-    {
-        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
-    } 
+-- | 'contramap' carn turn a 'SiphonOp' for bytes into a 'SiphonOp' for text.
+instance Contravariant (SiphonOp e a) where
+    contramap f (SiphonOp (Siphon s)) = SiphonOp . Siphon $ case s of
+        Pure p -> Pure p
+        Other o -> Other $ case o of
+            Exhaustive e -> Exhaustive $ \producer ->
+                e $ producer >-> P.map f
+            Nonexhaustive ne -> Nonexhaustive $ \producer ->
+                ne $ producer >-> P.map f
 
--- | 'fmap' maps over the encoding error. 
-instance Functor Lines where
-  fmap f (Lines func lt) = Lines (\x y z -> fmap (bimap f id) $ func x y z) lt
+-- | 'divide' builds a 'SiphonOp' for a composite out of the 'SiphonOp's
+-- for the parts.
+instance Monoid a => Divisible (SiphonOp e a) where
+    divide divider siphonOp1 siphonOp2 = contramap divider . SiphonOp $ 
+        (getSiphonOp (contramap fst siphonOp1)) 
+        `mappend`
+        (getSiphonOp (contramap snd siphonOp2))
+    conquer = SiphonOp (pure mempty)
 
+-- | 'choose' builds a 'SiphonOp' for a sum out of the 'SiphonOp's
+-- for the branches.
+instance Monoid a => Decidable (SiphonOp e a) where
+    choose chooser (SiphonOp s1) (SiphonOp s2) = 
+        contramap chooser . SiphonOp $ 
+            (contraPipeMapL s1) 
+            `mappend`
+            (contraPipeMapR s2)
+      where
+        contraPipeMapL (Siphon s) = Siphon $ case s of
+            Pure p -> Pure p
+            Other o -> Other $ case o of
+                Exhaustive e -> Exhaustive $ \producer ->
+                    e $ producer >-> allowLefts
+                Nonexhaustive ne -> Nonexhaustive $ \producer ->
+                    ne $ producer >-> allowLefts
+        contraPipeMapR (Siphon s) = Siphon $ case s of
+            Pure p -> Pure p
+            Other o -> Other $ case o of
+                Exhaustive e -> Exhaustive $ \producer ->
+                    e $ producer >-> allowRights
+                Nonexhaustive ne -> Nonexhaustive $ \producer ->
+                    ne $ producer >-> allowRights
+        allowLefts = do
+            e <- await
+            case e of 
+                Left l -> Pipes.yield l >> allowLefts
+                Right _ -> allowLefts
+        allowRights = do
+            e <- await
+            case e of 
+                Right r -> Pipes.yield r >> allowRights
+                Left _ -> allowRights
+    lose f = SiphonOp . Siphon . Other . Nonexhaustive $ \producer -> do
+        n <- next producer  
+        return $ case n of 
+            Left () -> Right mempty
+            Right (b,_) -> Right (absurd (f b))
 
 {-|
     Specifies a transformation that will be applied to each line of text,
@@ -836,26 +631,27 @@
  -}
 toLines :: DecodingFunction ByteString Text 
         -- ^ A decoding function for lines of text.
-        -> Siphon ByteString e ()
-        -- ^ 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.
+        -> Siphon ByteString e (() -> ())
+        -- ^ A 'Siphon' that determines how to handle decoding leftovers.
+        -- Pass @pure id@ to ignore leftovers. Pass @unwanted id@ to abort
+        -- the computation with an explicit error if leftovers remain. Pass
+        -- '_leftoverX' to throw a 'LeftoverException' if leftovers remain.
         -> Lines e 
 toLines decoder lopo = Lines
-    (\tweaker teardown producer -> do
+    (\tweaker tear producer -> do
         let freeLines = transFreeT tweaker 
                       . viewLines 
                       . decoder
                       $ producer
             viewLines = getConst . T.lines Const
-        teardown freeLines >>= runSiphon lopo)
+        tear freeLines >>= runSiphonDumb (fmap ($()) lopo))
     id 
 
 
 {-|
-  A simplified version of 'executePipelineFallibly' for when the error type is `Void`.
-  Note however that this function may still throw exceptions.
+  A simplified version of 'executePipelineFallibly' for when the error type
+  unified with `Void`.  Note however that this function may still throw
+  exceptions.
  -}
 executePipeline :: Piping Void a -> Tree (Stage Void) -> IO a 
 executePipeline pp pipeline = either absurd id <$> executePipelineFallibly pp pipeline
@@ -883,18 +679,18 @@
                         -- its parent in the tree.
                         -> 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 
+          PPNone _ -> blende ecpol <$> executeFallibly policy cp 
+          PPOutput _ -> blende ecpol <$> executeFallibly policy cp 
           PPError action -> do
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 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
+                (outbox, inbox, seal) <- spawn' (bounded 1)
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action $ (fromInput inbox,fromInput einbox))
@@ -903,11 +699,11 @@
                                     (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
+          PPInput _ -> blende ecpol <$> executeFallibly policy cp
+          PPInputOutput _ -> blende ecpol <$> executeFallibly policy cp
           PPInputError action -> do
-                (outbox, inbox, seal) <- spawn' Single
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (outbox, inbox, seal) <- spawn' (bounded 1)
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))
@@ -917,9 +713,9 @@
                                `finally` atomically seal `finally` atomically eseal
                     )
           PPInputOutputError action -> do
-                (ioutbox, iinbox, iseal) <- spawn' Single
-                (ooutbox, oinbox, oseal) <- spawn' Single
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (ioutbox, iinbox, iseal) <- spawn' (bounded 1)
+                (ooutbox, oinbox, oseal) <- spawn' (bounded 1)
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))
@@ -943,7 +739,7 @@
                     (\i _ -> mute $ pipei i) 
                     pipeline
           PPOutput action -> do
-                (outbox, inbox, seal) <- spawn' Single
+                (outbox, inbox, seal) <- spawn' (bounded 1)
                 runConceit $  
                     (Conceit $ action $ fromInput inbox)
                     <* 
@@ -956,7 +752,7 @@
                                `finally` atomically seal
                     ) 
           PPError action -> do
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action $ fromInput einbox)
@@ -969,8 +765,8 @@
                                 pipeline
                                 `finally` atomically eseal)
           PPOutputError action -> do
-                (outbox, inbox, seal) <- spawn' Single
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (outbox, inbox, seal) <- spawn' (bounded 1)
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action $ (fromInput inbox,fromInput einbox))
@@ -984,7 +780,7 @@
                                `finally` atomically seal `finally` atomically eseal
                     )
           PPInput action -> do
-                (outbox, inbox, seal) <- spawn' Single
+                (outbox, inbox, seal) <- spawn' (bounded 1)
                 runConceit $  
                     (Conceit $ action (toOutput outbox,atomically seal))
                     <* 
@@ -997,8 +793,8 @@
                                `finally` atomically seal
                     )
           PPInputOutput action -> do
-                (ioutbox, iinbox, iseal) <- spawn' Single
-                (ooutbox, oinbox, oseal) <- spawn' Single
+                (ioutbox, iinbox, iseal) <- spawn' (bounded 1)
+                (ooutbox, oinbox, oseal) <- spawn' (bounded 1)
                 runConceit $  
                     (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox))
                     <* 
@@ -1011,8 +807,8 @@
                                `finally` atomically iseal `finally` atomically oseal
                     )
           PPInputError action -> do
-                (outbox, inbox, seal) <- spawn' Single
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (outbox, inbox, seal) <- spawn' (bounded 1)
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action (toOutput outbox,atomically seal,fromInput einbox))
@@ -1026,9 +822,9 @@
                                `finally` atomically seal `finally` atomically eseal
                     )
           PPInputOutputError action -> do
-                (ioutbox, iinbox, iseal) <- spawn' Single
-                (ooutbox, oinbox, oseal) <- spawn' Single
-                (eoutbox, einbox, eseal) <- spawn' Single
+                (ioutbox, iinbox, iseal) <- spawn' (bounded 1)
+                (ooutbox, oinbox, oseal) <- spawn' (bounded 1)
+                (eoutbox, einbox, eseal) <- spawn' (bounded 1)
                 errf <- errorSiphonUTF8 <$> newMVar eoutbox
                 runConceit $  
                     (Conceit $ action (toOutput ioutbox,atomically iseal,fromInput oinbox,fromInput einbox))
@@ -1055,21 +851,7 @@
 mute :: Functor f => f a -> f ()
 mute = fmap (const ())
 
-{-|
-   An individual stage in a process pipeline. 
- -}
-data Stage e = Stage 
-           {
-             processDefinition' :: CreateProcess 
-           , stderrLines' :: Lines 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'.
 -}
@@ -1103,14 +885,14 @@
 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
+    runTree _ppend _ppend' (Node (Stage _cp _lpol _ecpol pipe) forest) = case forest of
         [] -> siphon $ \producer ->
-            blende ecpol <$> executeFallibly (ppend (fromFallibleProducer $ pipe producer) lpol) cp
+            blende _ecpol <$> executeFallibly (_ppend (fromFallibleProducer $ pipe producer) _lpol) _cp
         c1 : cs -> siphon $ \producer ->
-           blende ecpol <$> executeFallibly (ppmiddle (fromFallibleProducer $ pipe producer) (runNonEmpty ppend ppend' (c1 :| cs)) lpol) cp
+           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) 
+    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
@@ -1129,19 +911,3 @@
 "System.Process" is re-exported for convenience.
 
 -} 
-
-{- $deprecated
- 
-
--} 
-{-# DEPRECATED PipingPolicy "Use Piping instead" #-} 
-type PipingPolicy e a = Piping e a  
-
-{-# DEPRECATED LinePolicy "Use Lines instead" #-} 
-type LinePolicy e = Lines e   
-
-{-# DEPRECATED linePolicy "Use toLines instead" #-} 
-linePolicy :: DecodingFunction ByteString Text 
-           -> Siphon ByteString e ()
-           -> Lines e 
-linePolicy = toLines 
diff --git a/src/System/Process/Streaming/Extended.hs b/src/System/Process/Streaming/Extended.hs
new file mode 100644
--- /dev/null
+++ b/src/System/Process/Streaming/Extended.hs
@@ -0,0 +1,131 @@
+
+{-|
+-}
+
+{-# LANGUAGE DeriveFunctor #-}
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE ViewPatterns #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+
+module System.Process.Streaming.Extended ( 
+       Piap
+    ,  piapi
+    ,  piapo
+    ,  piape
+    ,  piapoe
+    ,  samei
+    ,  sameo
+    ,  samee
+    ,  sameioe
+    ,  toPiping
+    ,  pumpFromHandle 
+    ,  siphonToHandle 
+    ,  module System.Process.Streaming
+    ) where
+
+import Data.Text 
+import Control.Applicative
+import Control.Exception
+import Control.Concurrent.Conceit
+import Pipes.ByteString
+import System.IO
+
+import System.Process.Streaming
+import System.Process.Streaming.Internal
+
+toPiping :: Piap e a -> Piping e a  
+toPiping (Piap f) = PPInputOutputError f
+
+{-|
+    Do stuff with @stdout@.
+-}
+piapo :: Siphon ByteString e a -> Piap e a
+piapo s = Piap $ \(consumer, cleanup, producer1, producer2) -> do
+    let nullInput = runPump (pure ()) consumer `finally` cleanup
+        drainOutput = runSiphonDumb s producer1 
+        drainError = runSiphonDumb (pure ()) producer2
+    runConceit $ 
+        (\_ r _ -> r)
+        <$>
+        Conceit nullInput
+        <*>
+        Conceit drainOutput
+        <*>
+        Conceit drainError
+
+{-|
+    Do stuff with @stderr@.
+-}
+piape :: Siphon ByteString e a -> Piap e a
+piape s = Piap $ \(consumer, cleanup, producer1, producer2) -> do
+    let nullInput = runPump (pure ()) consumer `finally` cleanup
+        drainOutput = runSiphonDumb (pure ()) producer1 
+        drainError = runSiphonDumb s producer2
+    runConceit $ 
+        (\_ _ r -> r)
+        <$>
+        Conceit nullInput
+        <*>
+        Conceit drainOutput
+        <*>
+        Conceit drainError
+
+{-|
+    Do stuff with @stdin@.
+-}
+piapi :: Pump ByteString e a -> Piap e a
+piapi p = Piap $ \(consumer, cleanup, producer1, producer2) -> do
+    let nullInput = runPump p consumer `finally` cleanup
+        drainOutput = runSiphonDumb (pure ()) producer1 
+        drainError = runSiphonDumb (pure ()) producer2
+    runConceit $ 
+        (\r _ _ -> r)
+        <$>
+        Conceit nullInput
+        <*>
+        Conceit drainOutput
+        <*>
+        Conceit drainError
+
+{-|
+    Do stuff with @stdout@ and @stderr@ combined as 'Text'.  
+-}
+piapoe :: Lines e -> Lines e -> Siphon Text e a -> Piap e a
+piapoe policy1 policy2 s = Piap $ \(consumer, cleanup, producer1, producer2) -> do
+    let nullInput = runPump (pure ()) consumer `finally` cleanup
+        combination = combined policy1 policy2 (runSiphonDumb s) producer1 producer2 
+    runConceit $ 
+        (\_ r -> r)
+        <$>
+        Conceit nullInput
+        <*>
+        Conceit combination
+
+{-|
+    Pipe @stdin@ to the created process' @stdin@.
+-}
+samei :: Piap e ()
+samei = piapi $ pumpFromHandle System.IO.stdin
+
+{-|
+    Pipe the created process' @stdout@ to @stdout@.
+-}
+sameo :: Piap e ()
+sameo = piapo $ siphonToHandle System.IO.stdout
+
+{-|
+    Pipe the created process' @stderr@ to @stderr@.
+-}
+samee :: Piap e ()
+samee = piape $ siphonToHandle System.IO.stderr
+
+sameioe :: Piap e ()
+sameioe = samei *> sameo *> samee
+
+pumpFromHandle :: Handle -> Pump ByteString e ()
+pumpFromHandle = fromProducer . fromHandle
+
+siphonToHandle :: Handle -> Siphon ByteString e ()
+siphonToHandle = fromConsumer . toHandle
diff --git a/src/System/Process/Streaming/Internal.hs b/src/System/Process/Streaming/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/System/Process/Streaming/Internal.hs
@@ -0,0 +1,397 @@
+{-# LANGUAGE DeriveFunctor #-}
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE ViewPatterns #-}
+
+module System.Process.Streaming.Internal ( 
+        Piping(..), 
+        Piap(..), 
+        Pump(..),
+        Siphon(..),
+        runSiphon,
+        runSiphonDumb,
+        Siphon_(..),
+        exhaustive,
+        Lines(..),
+        combined,
+        manyCombined,
+        Stage(..)
+    ) where
+
+import Data.Bifunctor
+import Data.Monoid
+import Data.Text 
+import Control.Applicative
+import Control.Applicative.Lift
+import Control.Monad
+import Control.Monad.Trans.Free hiding (Pure)
+import Control.Monad.Trans.Except
+import Control.Exception
+import Control.Concurrent
+import Control.Concurrent.Conceit
+import Pipes
+import qualified Pipes as P
+import qualified Pipes.Prelude as P
+import Pipes.ByteString
+import qualified Pipes.Text as T
+import Pipes.Concurrent
+import System.Process
+import System.Exit
+
+{-|
+    A 'Piping' 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 'Piping' 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 Piping 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)
+
+
+{-| 
+    'first' is useful to massage errors.
+-}
+instance Bifunctor Piping 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
+
+
+newtype Pump b e a = Pump { runPump :: Consumer b IO () -> IO (Either e a) } deriving Functor
+
+{-| 
+    'first' is useful to massage errors.
+-}
+instance Bifunctor (Pump b) where
+  bimap f g (Pump x) = Pump $ fmap (liftM  (bimap f g)) x
+
+instance Applicative (Pump b e) where
+  pure = Pump . pure . pure . pure
+  Pump fs <*> Pump as = 
+      Pump $ \consumer -> do
+          (outbox1,inbox1,seal1) <- spawn' (bounded 1)
+          (outbox2,inbox2,seal2) <- spawn' (bounded 1)
+          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 (Monoid a) => Monoid (Pump b e a) where
+   mempty = Pump . pure . pure . pure $ mempty
+   mappend s1 s2 = (<>) <$> s1 <*> s2
+
+-- instance IsString b => IsString (Pump b e ()) where 
+--    fromString = fromProducer . P.yield . fromString 
+
+{-|
+    An alternative to `Piping` for defining what to do with the
+    standard streams. 'Piap' is an instance of 'Applicative', unlike
+    'Piping'. 
+
+    With 'Piap', the standard streams are always piped. The values of
+    @std_in@, @std_out@ and @std_err@ in the 'CreateProcess' record are
+    ignored.
+ -}
+newtype Piap e a = Piap { runPiap :: (Consumer ByteString IO (), IO (), Producer ByteString IO (), Producer ByteString IO ()) -> IO (Either e a) } deriving (Functor)
+
+instance Bifunctor Piap where
+  bimap f g (Piap action) = Piap $ fmap (fmap (bimap f g)) action
+
+
+{-| 
+    'pure' creates a 'Piap' that writes nothing to @stdin@ and drains
+    @stdout@ and @stderr@, discarding the data.
+
+    '<*>' schedules the writes to @stdin@ sequentially, and the reads from
+    @stdout@ and @stderr@ concurrently.
+-}
+instance Applicative (Piap e) where
+  pure a = Piap $ \(consumer, cleanup, producer1, producer2) -> do
+      let nullInput = runPump (pure ()) consumer `finally` cleanup
+          drainOutput = runSiphonDumb (pure ()) producer1
+          drainError = runSiphonDumb (pure ()) producer2
+      runConceit $ 
+          (\_ _ _ -> a)
+          <$>
+          Conceit nullInput
+          <*>
+          Conceit drainOutput
+          <*>
+          Conceit drainError
+
+  (Piap fa) <*> (Piap fb) = Piap $ \(consumer, cleanup, producer1, producer2) -> do
+        latch <- newEmptyMVar :: IO (MVar ())
+        (ioutbox, iinbox, iseal) <- spawn' (bounded 1)
+        (ooutbox, oinbox, oseal) <- spawn' (bounded 1)
+        (eoutbox, einbox, eseal) <- spawn' (bounded 1)
+        (ioutbox2, iinbox2, iseal2) <- spawn' (bounded 1)
+        (ooutbox2, oinbox2, oseal2) <- spawn' (bounded 1)
+        (eoutbox2, einbox2, eseal2) <- spawn' (bounded 1)
+        let reroutei = runEffect (fromInput (iinbox <> iinbox2) >-> consumer)
+                       `finally` atomically iseal 
+                       `finally` atomically iseal2
+                       `finally` cleanup
+            rerouteo = runEffect (producer1 >-> toOutput (ooutbox <> ooutbox2))
+                       `finally` atomically oseal 
+                       `finally` atomically oseal2
+            reroutee = runEffect (producer2 >-> toOutput (eoutbox <> eoutbox2))
+                       `finally` atomically eseal 
+                       `finally` atomically eseal2
+            deceivedf = fa 
+                (toOutput ioutbox, 
+                 atomically iseal `finally` putMVar latch (), 
+                 fromInput oinbox, 
+                 fromInput einbox)
+                 `finally` atomically iseal 
+                 `finally` atomically oseal 
+                 `finally` atomically eseal 
+            deceivedx = fb 
+                (liftIO (takeMVar latch) *> toOutput ioutbox2, 
+                 atomically iseal2, 
+                 fromInput oinbox2, 
+                 fromInput einbox2)
+                 `finally` atomically iseal2 
+                 `finally` atomically oseal2 
+                 `finally` atomically eseal2 
+        runConceit $
+            _Conceit reroutei 
+            *>
+            _Conceit rerouteo 
+            *> 
+            _Conceit reroutee 
+            *> 
+            (Conceit deceivedf <*> Conceit deceivedx)
+
+
+{-| 
+    A 'Siphon' represents a computation that completely drains a producer, but
+may fail early with an error of type @e@. 
+
+ -}
+newtype Siphon b e a = Siphon (Lift (Siphon_ b e) a) deriving (Functor)
+
+
+{-| 
+    '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.
+-}
+instance Applicative (Siphon b e) where
+    pure a = Siphon (pure a)
+    (Siphon fa) <*> (Siphon a) = Siphon (fa <*> a)
+
+data Siphon_ b e a = 
+         Exhaustive (forall r. Producer b IO r -> IO (Either e (a,r)))
+       | Nonexhaustive (Producer b IO () -> IO (Either e a))
+       deriving (Functor)
+
+
+instance Applicative (Siphon_ b e) where
+    pure a = Exhaustive $ \producer -> do
+        r <- runEffect (producer >-> P.drain)
+        pure (pure (a,r))
+    s1 <*> s2 = bifurcate (nonexhaustive s1) (nonexhaustive s2)  
+      where 
+        bifurcate fs as = Exhaustive $ \producer -> do
+            (outbox1,inbox1,seal1) <- spawn' (bounded 1)
+            (outbox2,inbox2,seal2) <- spawn' (bounded 1)
+            runConceit $
+                (,)
+                <$>
+                Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) 
+                                                        `finally` atomically seal1) 
+                                                        ((as $ fromInput inbox2) 
+                                                        `finally` atomically seal2) 
+                        )
+                <*>
+                _Conceit ((runEffect $ 
+                              producer >-> P.tee (toOutput outbox1 >> P.drain) 
+                                       >->       (toOutput outbox2 >> P.drain))   
+                          `finally` atomically seal1 `finally` atomically seal2
+                         ) 
+
+nonexhaustive :: Siphon_ b e a -> Producer b IO () -> IO (Either e a)
+nonexhaustive (Exhaustive e) = \producer -> liftM (fmap fst) (e producer)
+nonexhaustive (Nonexhaustive u) = u
+
+exhaustive :: Siphon_ b e a -> Producer b IO r -> IO (Either e (a,r))
+exhaustive s = case s of 
+    Exhaustive e -> e
+    Nonexhaustive activity -> \producer -> do 
+        (outbox,inbox,seal) <- spawn' (bounded 1)
+        runConceit $ 
+            (,) 
+            <$>
+            Conceit (activity (fromInput inbox) `finally` atomically seal)
+            <*>
+            _Conceit (runEffect (producer >-> (toOutput outbox >> P.drain)) 
+                      `finally` atomically seal
+                     )
+
+runSiphon :: Siphon b e a -> Producer b IO r -> IO (Either e (a,r))
+runSiphon (Siphon (unLift -> s)) = exhaustive s
+
+runSiphonDumb :: Siphon b e a -> Producer b IO () -> IO (Either e a)
+runSiphonDumb (Siphon (unLift -> s)) = nonexhaustive $ case s of 
+    Exhaustive _ -> s
+    Nonexhaustive _ -> Exhaustive (exhaustive s)
+
+
+instance Bifunctor (Siphon_ b) where
+  bimap f g s = case s of
+      Exhaustive u -> Exhaustive $ fmap (liftM  (bimap f (bimap g id))) u
+      Nonexhaustive h -> Nonexhaustive $ fmap (liftM  (bimap f g)) h
+
+{-| 
+    'first' is useful to massage errors.
+-}
+instance Bifunctor (Siphon b) where
+  bimap f g (Siphon s) = Siphon $ case s of
+      Pure a -> Pure (g a)
+      Other o -> Other (bimap f g o)
+
+instance (Monoid a) => Monoid (Siphon b e a) where
+   mempty = pure mempty
+   mappend s1 s2 = (<>) <$> s1 <*> s2
+
+
+
+-- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here
+combined :: Lines e 
+         -> Lines e 
+         -> (Producer T.Text IO () -> IO (Either e a))
+         -> Producer ByteString IO () 
+         -> Producer ByteString IO () 
+         -> IO (Either e a)
+combined (Lines fun1 twk1) (Lines fun2 twk2) combinedConsumer prod1 prod2 = 
+    manyCombined [fmap ($prod1) (fun1 twk1), fmap ($prod2) (fun2 twk2)] combinedConsumer 
+
+manyCombined :: [(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' (bounded 1)
+    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)
+
+{-|
+    A configuration parameter used in functions that combine lines of text from
+    multiple streams.
+ -}
+
+data Lines e = Lines 
+    {
+        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
+    } 
+
+-- | 'fmap' maps over the encoding error. 
+instance Functor Lines where
+  fmap f (Lines func lt) = Lines (\x y z -> fmap (bimap f id) $ func x y z) lt
+
+
+
+{-|
+   An individual stage in a process pipeline. 
+ -}
+data Stage e = Stage 
+           {
+             _processDefinition :: CreateProcess 
+           , _stderrLines :: Lines 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)
+
+
+
+
diff --git a/src/System/Process/Streaming/Tutorial.hs b/src/System/Process/Streaming/Tutorial.hs
--- a/src/System/Process/Streaming/Tutorial.hs
+++ b/src/System/Process/Streaming/Tutorial.hs
@@ -85,7 +85,6 @@
 >>> 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
@@ -99,15 +98,20 @@
 
 {- $collstdout  
 
-This example uses the 'toLazyM' fold from @pipes-bytestring@.
+We can do this with the 'toLazyM' fold from @pipes-bytestring@:
 
->>> execute (pipeo (fromFold B.toLazyM)) (shell "echo ooo")
+>>> execute (pipeo intoLazyBytes) (shell "echo ooo")
 (ExitSuccess,"ooo\n")
 
+But the 'intoLazyBytes' function is easier to use:
+
+>>> execute (pipeo intoLazyBytes) (shell "echo ooo")
+(ExitSuccess,"ooo\n")
+
 'Siphon's are functors, so if we wanted to collect the output as a strict
 'ByteString', we could do
 
->>> execute (pipeo (BL.toStrict <$> fromFold B.toLazyM)) (shell "echo ooo")
+>>> execute (pipeo (BL.toStrict <$> intoLazyBytes)) (shell "echo ooo")
 (ExitSuccess,"ooo\n")
 
 Of course, collecting the output in this way breaks streaming. But this is OK
@@ -120,7 +124,7 @@
 
 We can use 'pipeoe' collect @stdout@ and @stderr@ concurrently:
 
->>> execute (pipeoe (fromFold B.toLazyM) (fromFold B.toLazyM)) (shell "{ echo ooo ; echo eee 1>&2 ; }")
+>>> execute (pipeoe intoLazyBytes intoLazyBytes) (shell "{ echo ooo ; echo eee 1>&2 ; }")
 (ExitSuccess,("ooo\n","eee\n"))
 
 -}
@@ -129,17 +133,17 @@
 {- $collstdouttext  
 
 If we want to consume @stdout@ as text, we need to use the 'encoded'
-function. 'encoded' takes as parameters a decoding function (the example
-uses one from @pipes-text@) and a 'Siphon' that specifies how to handle the
-leftovers. It returns a function that converts a 'Siphon' for text
-into a 'Siphon' for bytes.
+function. 'encoded' takes as parameters a decoding function (here
+'decodeUtf8') and a 'Siphon' that specifies how to handle the leftovers. It
+returns a function that converts a 'Siphon' for text into a 'Siphon' for
+bytes.
 
 In the example we pass @pure id@ as the leftover-handling 'Siphon'. This
 means "drain all the undecoded data remaining in the stream and return
-unchanged the result of @(fromFold T.toLazyM)@". In other words: ignore any
+unchanged the result of 'intoLazyText'". In other words: ignore any
 leftovers.
 
->>> execute (pipeo (encoded T.decodeUtf8 (pure id) (fromFold T.toLazyM))) (shell "echo ooo")
+>>> execute (pipeo (encoded decodeUtf8 (pure id) intoLazyText)) (shell "echo ooo")
 (ExitSuccess,"ooo\n")
 
 But suppose we want to interrupt the execution of the program when we
@@ -150,12 +154,12 @@
 stream as the error value. So, in this example the error type will be
 'ByteString':
 
->>> executeFallibly (pipeo (encoded T.decodeUtf8 (unwanted id) (fromFold T.toLazyM))) (shell "echo ooo")
+>>> executeFallibly (pipeo (encoded decodeUtf8 (unwanted id) intoLazyText)) (shell "echo ooo")
 Right (ExitSuccess,"ooo\n")
 
 Notice also that we had to switch from 'execute' to 'executeFallibly'. This
 is because, for the first time in the tutorial, we actually have a need for
-the error type. 'execute' only works when the error type is 'Void'.
+the error type. 'execute' only works when the error type unified with 'Void'.
 
 Beware: even if the error type is 'Void', exceptions can still be thrown.
 -}
@@ -172,9 +176,9 @@
 
 >>> :{ 
    let 
-      lin = toLines T.decodeUtf8 (pure ()) 
+      lin = toLines decodeUtf8 (pure id) 
       program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }"
-   in execute (pipeoec lin lin (fromFold T.toLazyM)) program
+   in execute (pipeoec lin lin intoLazyText) program
    :}
 (ExitSuccess,"ooo\neee\n")
 
@@ -182,11 +186,11 @@
 
 >>> :{ 
    let 
-      lin = toLines T.decodeUtf8 (pure ()) 
+      lin = toLines decodeUtf8 (pure id) 
       lin_stdout = tweakLines (\p -> P.yield "O" *> p) lin 
       lin_stderr = tweakLines (\p -> P.yield "E" *> p) lin 
       program = shell "{ echo ooo ; sleep 1 ; echo eee 1>&2 ; }"
-   in execute (pipeoec lin_stdout lin_stderr (fromFold T.toLazyM)) program
+   in execute (pipeoec lin_stdout lin_stderr intoLazyText) program
    :}
 (ExitSuccess,"Oooo\nEeee\n")
 
@@ -197,7 +201,7 @@
 We can feed bytes to @stdin@ while we read @stdout@ or @stderr@. We use the
 'Pump' datatype for that.
 
->>> execute (pipeio (fromProducer (yield "iii")) (fromFold B.toLazyM)) (shell "cat")
+>>> execute (pipeio (fromLazyBytes "iii") intoLazyBytes) (shell "cat")
 (ExitSuccess,((),"iii"))
 -}
 
diff --git a/tests/test.hs b/tests/test.hs
--- a/tests/test.hs
+++ b/tests/test.hs
@@ -102,8 +102,8 @@
 
 combinedStdoutStderr :: IO (ExitCode,TL.Text)
 combinedStdoutStderr = execute
-    (pipeoec (linePolicy T.decodeIso8859_1 (pure ()))
-             (tweakLines annotate $ linePolicy T.decodeIso8859_1 (pure ()))    
+    (pipeoec (toLines T.decodeIso8859_1 (pure id))
+             (tweakLines annotate $ toLines T.decodeIso8859_1 (pure id))    
              (fromFold T.toLazyM))
     (shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }")
   where
@@ -195,7 +195,7 @@
 singletonPipeline =  executePipelineFallibly 
     (pipeoe (fromFold B.toLazyM) 
             (fromFold B.toLazyM)) 
-    (pure $ stage (linePolicy T.decodeUtf8 (pure ())) pipefail $ 
+    (pure $ stage (toLines T.decodeUtf8 (pure id)) pipefail $ 
          shell "{ echo ooo ; echo eee 1>&2 ; echo ppp ;  echo ffff 1>&2 ; }"
     )     
 
@@ -211,7 +211,7 @@
 basicPipeline =  executePipelineFallibly 
     (pipeio (fromProducer $ yield "aaabbb\naaaccc\nxxxccc") 
             (fromFold B.toLazyM)) 
-    (fmap (stage (linePolicy T.decodeUtf8 (pure ())) pipefail) $   
+    (fmap (stage (toLines T.decodeUtf8 (pure id)) pipefail) $   
         Node (shell "grep aaa") [Node (shell "grep ccc") []] )
 
 -------------------------------------------------------------------------------
@@ -243,28 +243,28 @@
     succStage = P.map (Data.ByteString.map succ)
 
     rootStage :: Stage Int 
-    rootStage = stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+    rootStage = stage (toLines T.decodeIso8859_1 (pure id))                 
                       pipefail
                       (shell "{ echo oooaaa ; echo eee 1>&2 ; echo xxx ;  echo ffff 1>&2 ; }")
 
     branch1 :: Stage Int 
-    branch1 = stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+    branch1 = stage (toLines T.decodeIso8859_1 (pure id))                 
                     pipefail
                     (shell "grep ooo")
     branch2 :: Stage Int 
-    branch2 = stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+    branch2 = stage (toLines T.decodeIso8859_1 (pure id))                 
                     pipefail
                     (shell "grep xxx")
 
     terminalStage1 :: Stage Int 
     terminalStage1 = inbound (\p -> p >-> succStage) $
-        stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+        stage (toLines T.decodeIso8859_1 (pure id))                 
               pipefail
               (shell "tr -d b")
 
     terminalStage2 :: Stage Int
     terminalStage2 = inbound (\p -> p >-> succStage) $
-        stage (linePolicy T.decodeIso8859_1 (pure ()))                 
+        stage (toLines T.decodeIso8859_1 (pure id))                 
               pipefail
               (shell $ "cat > " ++ branchingPipelineFile)
 
@@ -307,7 +307,7 @@
     (pipeoec lp lp countLines)
     (proc "tests/alternating.sh" [])
   where
-    lp = linePolicy T.decodeIso8859_1 (pure ()) 
+    lp = toLines T.decodeIso8859_1 (pure id) 
     countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines
 
 
@@ -316,7 +316,7 @@
     (pipeoec lp lp $ (,) <$> countLines <*> countLines)
     (proc "tests/alternating.sh" [])
   where
-    lp = linePolicy T.decodeIso8859_1 (pure ()) 
+    lp = toLines T.decodeIso8859_1 (pure id) 
     countLines = fromFold $ P.sum . G.folds const () (const 1) . view T.lines