diff --git a/src/Protocols/Df.hs b/src/Protocols/Df.hs
index eea702ea..adf07314 100644
--- a/src/Protocols/Df.hs
+++ b/src/Protocols/Df.hs
@@ -33,6 +33,9 @@ module Protocols.Df (
   mapMaybe,
   catMaybes,
   coerce,
+  compressor,
+  expander,
+  compander,
   filter,
   filterS,
   either,
@@ -238,6 +241,84 @@ forceResetSanity = forceResetSanityGeneric
 coerce :: (Coerce.Coercible a b) => Circuit (Df dom a) (Df dom b)
 coerce = fromSignals $ \(fwdA, bwdB) -> (Coerce.coerce bwdB, Coerce.coerce fwdA)
 
+{- | Takes one or more values from the left and "compresses" it into a single
+value that is occasionally sent to the right. Useful for taking small high-speed
+inputs (like bits from a serial line) and turning them into slower wide outputs
+(like 32-bit integers).
+
+Example:
+
+>>> accumulate xs x = let xs' = x:xs in if length xs' == 3 then ([], Just xs') else (xs', Nothing)
+>>> circuit = C.exposeClockResetEnable (compressor @C.System [] accumulate)
+>>> take 2 (simulateCSE circuit [(1::Int),2,3,4,5,6,7])
+[[3,2,1],[6,5,4]]
+-}
+compressor ::
+  forall dom s i o.
+  (C.HiddenClockResetEnable dom, C.NFDataX s) =>
+  s ->
+  -- | Return `Just` when the compressed value is complete.
+  (s -> i -> (s, Maybe o)) ->
+  Circuit (Df dom i) (Df dom o)
+compressor s0 f = compander s0 $
+  \s i ->
+    let (s', o) = f s i
+     in (s', o, True)
+
+{- | Takes a value from the left and "expands" it into one or more values that
+are sent off to the right. Useful for taking wide, slow inputs (like a stream of
+32-bit integers) and turning them into a fast, narrow output (like a stream of bits).
+
+Example:
+
+>>> step index = if index == maxBound then (0, True) else (index + 1, False)
+>>> expandVector index vec = let (index', done) = step index in (index', vec C.!! index, done)
+>>> circuit = C.exposeClockResetEnable (expander @C.System (0 :: C.Index 3) expandVector)
+>>> take 6 (simulateCSE circuit [1 :> 2 :> 3 :> Nil, 4 :> 5 :> 6 :> Nil])
+[1,2,3,4,5,6]
+-}
+expander ::
+  forall dom i o s.
+  (C.HiddenClockResetEnable dom, C.NFDataX s) =>
+  s ->
+  -- | Return `True` when you're finished with the current input value
+  -- and are ready for the next one.
+  (s -> i -> (s, o, Bool)) ->
+  Circuit (Df dom i) (Df dom o)
+expander s0 f = compander s0 $
+  \s i ->
+    let (s', o, done) = f s i
+     in (s', Just o, done)
+
+{- | Takes values from the left,
+possibly holding them there for a while while working on them,
+and occasionally sends values off to the right.
+Used to implement both `expander` and `compressor`, so you can use it
+when there's not a straightforward one-to-many or many-to-one relationship
+between the input and output streams.
+-}
+compander ::
+  forall dom i o s.
+  (C.HiddenClockResetEnable dom, C.NFDataX s) =>
+  s ->
+  -- | Return `True` when you're finished with the current input value
+  -- and are ready for the next one.
+  -- Return `Just` to send the produced value off to the right.
+  (s -> i -> (s, Maybe o, Bool)) ->
+  Circuit (Df dom i) (Df dom o)
+compander s0 f = forceResetSanity |> Circuit (C.unbundle . go . C.bundle)
+ where
+  go :: Signal dom (Data i, Ack) -> Signal dom (Ack, Data o)
+  go = C.mealy f' s0
+  f' :: s -> (Data i, Ack) -> (s, (Ack, Data o))
+  f' s (NoData, _) = (s, (Ack False, NoData))
+  f' s (Data i, Ack ack) = (s'', (Ack ackBack, maybe NoData Data o))
+   where
+    (s', o, doneWithInput) = f s i
+    -- We only care about the downstream ack if we're sending them something
+    mustWaitForAck = Maybe.isJust o
+    (s'', ackBack) = if mustWaitForAck && not ack then (s, False) else (s', doneWithInput)
+
 -- | Like 'P.map', but over payload (/a/) of a Df stream.
 map :: (a -> b) -> Circuit (Df dom a) (Df dom b)
 map f = mapS (C.pure f)
diff --git a/tests/Tests/Protocols/Df.hs b/tests/Tests/Protocols/Df.hs
index 30b3a387..8e994514 100644
--- a/tests/Tests/Protocols/Df.hs
+++ b/tests/Tests/Protocols/Df.hs
@@ -133,6 +133,100 @@ prop_catMaybes =
     catMaybes
     Df.catMaybes
 
+-- A parameterized test definition validating that an expander which
+-- simply releases a value downstream once every N cycles
+-- does not otherwise change the contents of the stream.
+testExpanderPassThrough :: forall n. (C.KnownNat n) => C.SNat n -> Property
+testExpanderPassThrough _periodicity =
+  idWithModelSingleDomain @C.System
+    defExpectOptions
+    (genData genSmallInt)
+    (C.exposeClockResetEnable id)
+    ( C.exposeClockResetEnable $
+        passThroughExpander |> Df.catMaybes
+    )
+ where
+  -- Just stares at a value for a few cycles and then forwards it
+  passThroughExpander ::
+    forall dom a.
+    (C.HiddenClockResetEnable dom) =>
+    Circuit (Df dom a) (Df dom (Maybe a))
+  passThroughExpander = Df.expander (0 :: C.Index n) $ \count input ->
+    let done = count == maxBound
+     in ( if done then 0 else count + 1
+        , if done then Just input else Nothing
+        , done
+        )
+
+prop_expander_passthrough_linerate :: Property
+prop_expander_passthrough_linerate = testExpanderPassThrough C.d1
+
+prop_expander_passthrough_slow :: Property
+prop_expander_passthrough_slow = testExpanderPassThrough C.d4
+
+-- A parameterized test definition validating that an expander duplicates
+-- input values N times and sends them downstream.
+testExpanderDuplicate :: forall n. (C.KnownNat n) => C.SNat n -> Property
+testExpanderDuplicate duplication =
+  idWithModelSingleDomain @C.System
+    defExpectOptions
+    (genData genSmallInt)
+    (C.exposeClockResetEnable (concatMap (replicate (C.snatToNum duplication))))
+    ( C.exposeClockResetEnable
+        duplicator
+    )
+ where
+  -- Creates n copies of a value
+  duplicator ::
+    forall dom a.
+    (C.HiddenClockResetEnable dom) =>
+    Circuit (Df dom a) (Df dom a)
+  duplicator = Df.expander (0 :: C.Index n) $ \count input ->
+    let done = count == maxBound
+     in ( if done then 0 else count + 1
+        , input
+        , done
+        )
+
+prop_expander_duplicate_linerate :: Property
+prop_expander_duplicate_linerate = testExpanderDuplicate C.d1
+
+prop_expander_duplicate_slow :: Property
+prop_expander_duplicate_slow = testExpanderDuplicate C.d4
+
+-- A paremterized test definition validating that a compressor correctly
+-- sums up batches of N values.
+testCompressorSum :: forall n. (C.KnownNat n) => C.SNat n -> Property
+testCompressorSum batchSize =
+  idWithModelSingleDomain @C.System
+    defExpectOptions
+    (genData genSmallInt)
+    (C.exposeClockResetEnable referenceImpl)
+    ( C.exposeClockResetEnable
+        passThroughExpander
+    )
+ where
+  chunk = C.snatToNum batchSize
+  -- Given [a,b,c,d,e] and chunk = 2, yield [a+b,c+d]
+  referenceImpl = map sum . takeWhile ((== chunk) . length) . map (take chunk) . iterate (drop chunk)
+  -- Sum groups of `n` samples together
+  passThroughExpander ::
+    forall dom.
+    (C.HiddenClockResetEnable dom) =>
+    Circuit (Df dom Int) (Df dom Int)
+  passThroughExpander = Df.compressor (0 :: C.Index n, 0 :: Int) $ \(count, total) input ->
+    let done = count == maxBound
+        total' = total + input
+     in ( if done then (0, 0) else (count + 1, total')
+        , if done then Just total' else Nothing
+        )
+
+prop_compressor_sum_linerate :: Property
+prop_compressor_sum_linerate = testCompressorSum C.d1
+
+prop_compressor_sum_slow :: Property
+prop_compressor_sum_slow = testCompressorSum C.d4
+
 prop_registerFwd :: Property
 prop_registerFwd =
   idWithModelSingleDomain
@@ -335,9 +429,8 @@ prop_selectN =
     n <- genSmallInt
     ixs <- Gen.list (Range.singleton n) Gen.enumBounded
     lenghts <- Gen.list (Range.singleton n) Gen.enumBounded
-    let
-      tallied = tallyOn fst (fromIntegral . snd) (zip ixs lenghts)
-      tall i = fromMaybe 0 (HashMap.lookup i tallied)
+    let tallied = tallyOn fst (fromIntegral . snd) (zip ixs lenghts)
+        tall i = fromMaybe 0 (HashMap.lookup i tallied)
     dats <- mapM (\i -> Gen.list (Range.singleton (tall i)) genSmallInt) C.indicesI
     pure (dats, zip ixs lenghts)