diff --git a/format/context.go b/format/context.go
deleted file mode 100644
index d1fb703..0000000
--- a/format/context.go
+++ /dev/null
@@ -1,21 +0,0 @@
-package format
-
-import (
-	"context"
-)
-
-const (
-	completedChKey = "completedCh"
-)
-
-// SetCompletedChannel is used to set a channel for indication processing completion in the provided context.
-func SetCompletedChannel(ctx context.Context, completedCh chan string) context.Context {
-	return context.WithValue(ctx, completedChKey, completedCh)
-}
-
-// MarkPathComplete is used to indicate that all processing has finished for the provided path.
-// This is done by adding the path to the completion channel which should have already been set using
-// SetCompletedChannel.
-func MarkPathComplete(ctx context.Context, path string) {
-	ctx.Value(completedChKey).(chan string) <- path
-}
diff --git a/format/formatter.go b/format/formatter.go
index 532a206..7e2353a 100644
--- a/format/formatter.go
+++ b/format/formatter.go
@@ -27,6 +27,8 @@ type Formatter struct {
 	// internal compiled versions of Includes and Excludes.
 	includes []glob.Glob
 	excludes []glob.Glob
+
+	batch []string
 }
 
 // Executable returns the path to the executable defined by Command
@@ -34,30 +36,53 @@ func (f *Formatter) Executable() string {
 	return f.executable
 }
 
-func (f *Formatter) Apply(ctx context.Context, paths []string) error {
-	// only apply if the resultant batch is not empty
-	if len(paths) > 0 {
-		// construct args, starting with config
-		args := f.config.Options
+func (f *Formatter) Apply(ctx context.Context, paths []string, filter bool) error {
+	// construct args, starting with config
+	args := f.config.Options
 
-		// append each file path
+	// If filter is true it indicates we are executing as part of a pipeline.
+	// In such a scenario each formatter must sub filter the paths provided as different formatters might want different
+	// files in a pipeline.
+	if filter {
+		// reset the batch
+		f.batch = f.batch[:]
+
+		// filter paths
 		for _, path := range paths {
-			args = append(args, path)
+			if f.Wants(path) {
+				f.batch = append(f.batch, path)
+			}
 		}
 
-		// execute
-		start := time.Now()
-		cmd := exec.CommandContext(ctx, f.config.Command, args...)
-
-		if out, err := cmd.CombinedOutput(); err != nil {
-			f.log.Debugf("\n%v", string(out))
-			// todo log output
-			return err
+		// exit early if nothing to process
+		if len(f.batch) == 0 {
+			return nil
 		}
 
-		f.log.Infof("%v files processed in %v", len(paths), time.Now().Sub(start))
+		// append paths to the args
+		args = append(args, f.batch...)
+	} else {
+		// exit early if nothing to process
+		if len(paths) == 0 {
+			return nil
+		}
+
+		// append paths to the args
+		args = append(args, paths...)
 	}
 
+	// execute the command
+	start := time.Now()
+	cmd := exec.CommandContext(ctx, f.config.Command, args...)
+
+	if out, err := cmd.CombinedOutput(); err != nil {
+		f.log.Debugf("\n%v", string(out))
+		// todo log output
+		return err
+	}
+
+	f.log.Infof("%v files processed in %v", len(paths), time.Now().Sub(start))
+
 	return nil
 }
 
@@ -95,7 +120,11 @@ func NewFormatter(
 	f.executable = executable
 
 	// initialise internal state
-	f.log = log.WithPrefix("format | " + name)
+	if config.Pipeline == "" {
+		f.log = log.WithPrefix(fmt.Sprintf("format | %s", name))
+	} else {
+		f.log = log.WithPrefix(fmt.Sprintf("format | %s[%s]", config.Pipeline, name))
+	}
 
 	f.includes, err = CompileGlobs(config.Includes)
 	if err != nil {
diff --git a/format/pipeline.go b/format/pipeline.go
index 29b94fb..84dbefc 100644
--- a/format/pipeline.go
+++ b/format/pipeline.go
@@ -23,7 +23,7 @@ func (p *Pipeline) Wants(path string) bool {
 
 func (p *Pipeline) Apply(ctx context.Context, paths []string) error {
 	for _, f := range p.sequence {
-		if err := f.Apply(ctx, paths); err != nil {
+		if err := f.Apply(ctx, paths, len(p.sequence) > 1); err != nil {
 			return err
 		}
 	}