diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -1,3 +1,7 @@
+## 1.3.0.3
+
+* Improve fusion framework rewrite rules
+
 ## 1.3.0.2
 
 * Replace `ReadMode` with `WriteMode` in `withSinkFile`
diff --git a/conduit.cabal b/conduit.cabal
--- a/conduit.cabal
+++ b/conduit.cabal
@@ -1,5 +1,5 @@
 Name:                conduit
-Version:             1.3.0.2
+Version:             1.3.0.3
 Synopsis:            Streaming data processing library.
 description:
     `conduit` is a solution to the streaming data problem, allowing for production,
@@ -58,7 +58,6 @@
 
   if os(windows)
     build-depends:     Win32
-                     , filepath
     other-modules:     System.Win32File
     cpp-options:       -DWINDOWS
   else
@@ -96,6 +95,9 @@
                    , filepath
                    , unliftio >= 0.2.4.0
     ghc-options:     -Wall
+
+  if os(windows)
+    cpp-options:     -DWINDOWS
 
 --test-suite doctests
 --    hs-source-dirs: test
diff --git a/src/Data/Conduit.hs b/src/Data/Conduit.hs
--- a/src/Data/Conduit.hs
+++ b/src/Data/Conduit.hs
@@ -104,23 +104,3 @@
 import Data.Functor.Identity (Identity, runIdentity)
 import Control.Monad.Trans.Resource (ResourceT, runResourceT)
 import Control.Monad.IO.Unlift (MonadUnliftIO)
-
--- | Run a pure pipeline until processing completes, i.e. a pipeline
--- with @Identity@ as the base monad. This is equivalient to
--- @runIdentity . runConduit@.
---
--- @since 1.2.8
-runConduitPure :: ConduitT () Void Identity r -> r
-runConduitPure = runIdentity . runConduit
-{-# INLINE runConduitPure #-}
-
--- | Run a pipeline which acquires resources with @ResourceT@, and
--- then run the @ResourceT@ transformer. This is equivalent to
--- @runResourceT . runConduit@.
---
--- @since 1.2.8
-runConduitRes :: MonadUnliftIO m
-              => ConduitT () Void (ResourceT m) r
-              -> m r
-runConduitRes = runResourceT . runConduit
-{-# INLINE runConduitRes #-}
diff --git a/src/Data/Conduit/Internal/Conduit.hs b/src/Data/Conduit/Internal/Conduit.hs
--- a/src/Data/Conduit/Internal/Conduit.hs
+++ b/src/Data/Conduit/Internal/Conduit.hs
@@ -34,6 +34,8 @@
     , yieldM
     , leftover
     , runConduit
+    , runConduitPure
+    , runConduitRes
     , fuse
     , connect
       -- ** Composition
@@ -97,6 +99,7 @@
 import Control.Monad.Trans.Class (MonadTrans (lift))
 import Control.Monad.IO.Unlift (MonadIO (liftIO), MonadUnliftIO, withRunInIO)
 import Control.Monad.Primitive (PrimMonad, PrimState, primitive)
+import Data.Functor.Identity (Identity, runIdentity)
 import Data.Void (Void, absurd)
 import Data.Monoid (Monoid (mappend, mempty))
 import Data.Semigroup (Semigroup ((<>)))
@@ -730,6 +733,22 @@
 -- Equivalent to 'fuse' and '=$=', however the latter is deprecated and will
 -- be removed in a future version.
 --
+-- Note that, while this operator looks like categorical composition
+-- (from "Control.Category"), there are a few reasons it's different:
+--
+-- * The position of the type parameters to 'ConduitT' do not
+--   match. We would need to change @ConduitT i o m r@ to @ConduitT r
+--   m i o@, which would preclude a 'Monad' or 'MonadTrans' instance.
+--
+-- * The result value from upstream and downstream are allowed to
+--   differ between upstream and downstream. In other words, we would
+--   need the type signature here to look like @ConduitT a b m r ->
+--   ConduitT b c m r -> ConduitT a c m r@.
+--
+-- * Due to leftovers, we do not have a left identity in Conduit. This
+--   can be achieved with the underlying @Pipe@ datatype, but this is
+--   not generally recommended. See <https://stackoverflow.com/a/15263700>.
+--
 -- @since 1.2.8
 (.|) :: Monad m
      => ConduitM a b m () -- ^ upstream
@@ -1231,3 +1250,23 @@
 {-# RULES "conduit: leftover l >> p" forall l (p :: ConduitT i o m r). leftover l >> p =
     ConduitT (Leftover (unConduitT p) l) #-}
     -}
+
+-- | Run a pure pipeline until processing completes, i.e. a pipeline
+-- with @Identity@ as the base monad. This is equivalient to
+-- @runIdentity . runConduit@.
+--
+-- @since 1.2.8
+runConduitPure :: ConduitT () Void Identity r -> r
+runConduitPure = runIdentity . runConduit
+{-# INLINE runConduitPure #-}
+
+-- | Run a pipeline which acquires resources with @ResourceT@, and
+-- then run the @ResourceT@ transformer. This is equivalent to
+-- @runResourceT . runConduit@.
+--
+-- @since 1.2.8
+runConduitRes :: MonadUnliftIO m
+              => ConduitT () Void (ResourceT m) r
+              -> m r
+runConduitRes = runResourceT . runConduit
+{-# INLINE runConduitRes #-}
diff --git a/src/Data/Conduit/Internal/Fusion.hs b/src/Data/Conduit/Internal/Fusion.hs
--- a/src/Data/Conduit/Internal/Fusion.hs
+++ b/src/Data/Conduit/Internal/Fusion.hs
@@ -3,6 +3,7 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DeriveFunctor #-}
 {-# LANGUAGE Trustworthy #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 module Data.Conduit.Internal.Fusion
     ( -- ** Types
       Step (..)
@@ -25,6 +26,7 @@
 import Data.Conduit.Internal.Pipe (Pipe (..))
 import Data.Functor.Identity (Identity (runIdentity))
 import Data.Void (Void, absurd)
+import Control.Monad.Trans.Resource (runResourceT)
 
 -- | This is the same as stream fusion\'s Step. Constructors are renamed to
 -- avoid confusion with conduit names.
@@ -66,9 +68,15 @@
   ConduitWithStream (a .| b) (y . x)
 {-# INLINE fuseStream #-}
 
-{-# RULES "conduit: fuseStream" forall left right.
+{-# RULES "conduit: fuseStream (.|)" forall left right.
         unstream left .| unstream right = unstream (fuseStream left right)
   #-}
+{-# RULES "conduit: fuseStream (fuse)" forall left right.
+        fuse (unstream left) (unstream right) = unstream (fuseStream left right)
+  #-}
+{-# RULES "conduit: fuseStream (=$=)" forall left right.
+        unstream left =$= unstream right = unstream (fuseStream left right)
+  #-}
 
 runStream :: Monad m
           => ConduitWithStream () Void m r
@@ -91,8 +99,13 @@
 {-# RULES "conduit: runStream" forall stream.
         runConduit (unstream stream) = runStream stream
   #-}
+{-# RULES "conduit: runStream (pure)" forall stream.
+        runConduitPure (unstream stream) = runIdentity (runStream stream)
+  #-}
+{-# RULES "conduit: runStream (ResourceT)" forall stream.
+        runConduitRes (unstream stream) = runResourceT (runStream stream)
+  #-}
 
-{-
 connectStream :: Monad m
               => ConduitWithStream () i    m ()
               -> ConduitWithStream i  Void m r
@@ -111,10 +124,8 @@
                 Skip s' -> loop s'
                 Emit _ o -> absurd o
 {-# INLINE connectStream #-}
--}
 
-{- Deprecated
-{-# RULES "conduit: connectStream" forall left right.
+{-# RULES "conduit: connectStream ($$)" forall left right.
         unstream left $$ unstream right = connectStream left right
   #-}
 
@@ -138,28 +149,54 @@
                         Emit s' i -> loop [] (p i) s'
              in ms0 >>= loop [] (sink0 Done)
 {-# INLINE connectStream1 #-}
--}
 
-{- Deprecated
-{-# RULES "conduit: connectStream1" forall left right.
+{-# RULES "conduit: connectStream1 ($$)" forall left right.
         unstream left $$ right = connectStream1 left right
   #-}
--}
 
-{-
+{-# RULES "conduit: connectStream1 (runConduit/.|)" forall left right.
+        runConduit (unstream left .| right) = connectStream1 left right
+  #-}
+{-# RULES "conduit: connectStream1 (runConduit/=$=)" forall left right.
+        runConduit (unstream left =$= right) = connectStream1 left right
+  #-}
+{-# RULES "conduit: connectStream1 (runConduit/fuse)" forall left right.
+        runConduit (fuse (unstream left) right) = connectStream1 left right
+  #-}
 
-Not only will this rule not fire reliably, but due to finalizers, it can change
-behavior unless implemented very carefully. Odds are that the careful
-implementation won't be any faster, so leaving this commented out for now.
+{-# RULES "conduit: connectStream1 (runConduitPure/.|)" forall left right.
+        runConduitPure (unstream left .| right) = runIdentity (connectStream1 left right)
+  #-}
+{-# RULES "conduit: connectStream1 (runConduitPure/=$=)" forall left right.
+        runConduitPure (unstream left =$= right) = runIdentity (connectStream1 left right)
+  #-}
+{-# RULES "conduit: connectStream1 (runConduitPure/fuse)" forall left right.
+        runConduitPure (fuse (unstream left) right) = runIdentity (connectStream1 left right)
+  #-}
 
-connectStream2 :: Monad m
-               => ConduitT      () i    m ()
+{-# RULES "conduit: connectStream1 (runConduitRes/.|)" forall left right.
+        runConduitRes (unstream left .| right) = runResourceT (connectStream1 left right)
+  #-}
+{-# RULES "conduit: connectStream1 (runConduitRes/=$=)" forall left right.
+        runConduitRes (unstream left =$= right) = runResourceT (connectStream1 left right)
+  #-}
+{-# RULES "conduit: connectStream1 (runConduitRes/fuse)" forall left right.
+        runConduitRes (fuse (unstream left) right) = runResourceT (connectStream1 left right)
+  #-}
+
+connectStream2 :: forall i m r. Monad m
+               => ConduitT          () i    m ()
                -> ConduitWithStream i  Void m r
                -> m r
 connectStream2 (ConduitT src0) (ConduitWithStream _ fstream) =
-    run $ fstream $ Stream step' $ return (return (), src0 Done)
+    run $ fstream $ Stream step' $ return (src0 Done)
   where
-    step' (_, Done ()) = return $ Stop ()
+    step' :: Pipe () () i () m () -> m (Step (Pipe () () i () m ()) i ())
+    step' (Done ()) = return $ Stop ()
+    step' (HaveOutput pipe o) = return $ Emit pipe o
+    step' (NeedInput _ c) = return $ Skip $ c ()
+    step' (PipeM mp) = Skip <$> mp
+    step' (Leftover p ()) = return $ Skip p
     {-# INLINE step' #-}
 
     run (Stream step ms0) =
@@ -173,10 +210,39 @@
                 Skip s' -> loop s'
 {-# INLINE connectStream2 #-}
 
-{-# RULES "conduit: connectStream2" forall left right.
+{-# RULES "conduit: connectStream2 ($$)" forall left right.
         left $$ unstream right = connectStream2 left right
   #-}
--}
+
+{-# RULES "conduit: connectStream2 (runConduit/.|)" forall left right.
+        runConduit (left .| unstream right) = connectStream2 left right
+  #-}
+{-# RULES "conduit: connectStream2 (runConduit/fuse)" forall left right.
+        runConduit (fuse left (unstream right)) = connectStream2 left right
+  #-}
+{-# RULES "conduit: connectStream2 (runConduit/=$=)" forall left right.
+        runConduit (left =$= unstream right) = connectStream2 left right
+  #-}
+
+{-# RULES "conduit: connectStream2 (runConduitPure/.|)" forall left right.
+        runConduitPure (left .| unstream right) = runIdentity (connectStream2 left right)
+  #-}
+{-# RULES "conduit: connectStream2 (runConduitPure/fuse)" forall left right.
+        runConduitPure (fuse left (unstream right)) = runIdentity (connectStream2 left right)
+  #-}
+{-# RULES "conduit: connectStream2 (runConduitPure/=$=)" forall left right.
+        runConduitPure (left =$= unstream right) = runIdentity (connectStream2 left right)
+  #-}
+
+{-# RULES "conduit: connectStream2 (runConduitRes/.|)" forall left right.
+        runConduitRes (left .| unstream right) = runResourceT (connectStream2 left right)
+  #-}
+{-# RULES "conduit: connectStream2 (runConduitRes/fuse)" forall left right.
+        runConduitRes (fuse left (unstream right)) = runResourceT (connectStream2 left right)
+  #-}
+{-# RULES "conduit: connectStream2 (runConduitRes/=$=)" forall left right.
+        runConduitRes (left =$= unstream right) = runResourceT (connectStream2 left right)
+  #-}
 
 streamConduit :: ConduitT i o m r
               -> (Stream m i () -> Stream m o r)
diff --git a/test/Spec.hs b/test/Spec.hs
--- a/test/Spec.hs
+++ b/test/Spec.hs
@@ -11,7 +11,7 @@
 import Data.Conduit.Combinators (slidingWindow, chunksOfE, chunksOfExactlyE)
 import Data.List (intersperse, sort, find, mapAccumL)
 import Safe (tailSafe)
-import System.FilePath (takeExtension)
+import System.FilePath (takeExtension, (</>))
 import Test.Hspec
 import Test.Hspec.QuickCheck
 import qualified Data.Text as T
@@ -134,12 +134,12 @@
         res <- runConduitRes
              $ sourceDirectory "test" .| filterC (not . hasExtension' ".swp") .| sinkList
         sort res `shouldBe`
-          [ "test/Data"
-          , "test/Spec.hs"
-          , "test/StreamSpec.hs"
-          , "test/doctests.hs"
-          , "test/main.hs"
-          , "test/subdir"
+          [ "test" </> "Data"
+          , "test" </> "Spec.hs"
+          , "test" </> "StreamSpec.hs"
+          , "test" </> "doctests.hs"
+          , "test" </> "main.hs"
+          , "test" </> "subdir"
           ]
     it "sourceDirectoryDeep" $ do
         res1 <- runConduitRes
@@ -147,13 +147,13 @@
         res2 <- runConduitRes
               $ sourceDirectoryDeep True "test" .| filterC (not . hasExtension' ".swp") .| sinkList
         sort res1 `shouldBe`
-          [ "test/Data/Conduit/Extra/ZipConduitSpec.hs"
-          , "test/Data/Conduit/StreamSpec.hs"
-          , "test/Spec.hs"
-          , "test/StreamSpec.hs"
-          , "test/doctests.hs"
-          , "test/main.hs"
-          , "test/subdir/dummyfile.txt"
+          [ "test" </> "Data" </> "Conduit" </> "Extra" </> "ZipConduitSpec.hs"
+          , "test" </> "Data" </> "Conduit" </> "StreamSpec.hs"
+          , "test" </> "Spec.hs"
+          , "test" </> "StreamSpec.hs"
+          , "test" </> "doctests.hs"
+          , "test" </> "main.hs"
+          , "test" </> "subdir" </> "dummyfile.txt"
           ]
         sort res1 `shouldBe` sort res2
     prop "drop" $ \(T.pack -> input) count ->