Repository: puppetlabs/wash
Pull request #232 (2019-05-06 18:14:53)

Generalize `wash find`'s expression parser
This will make it easier to implement expression support for the `meta` primary.


Patch:
diff --git a/api/exec.go b/api/exec.go
index 9fc32ad..044074c 100644
--- a/api/exec.go
+++ b/api/exec.go
@@ -29,15 +29,19 @@ func sendPacket(ctx context.Context, w *json.Encoder, p *apitypes.ExecPacket) {
 	}
 }
 
+var outputStreamNames = [2]string{apitypes.Stdout, apitypes.Stderr}
+
 func streamOutput(ctx context.Context, w *json.Encoder, outputCh <-chan plugin.ExecOutputChunk) {
 	if outputCh == nil {
 		return
 	}
 
 	for chunk := range outputCh {
-		packet := apitypes.ExecPacket{TypeField: chunk.StreamID, Timestamp: chunk.Timestamp}
+		stream := outputStreamNames[chunk.StreamID]
+
+		packet := apitypes.ExecPacket{TypeField: stream, Timestamp: chunk.Timestamp}
 		if err := chunk.Err; err != nil {
-			packet.Err = newStreamingErrorObj(chunk.StreamID, err.Error())
+			packet.Err = newStreamingErrorObj(stream, err.Error())
 		} else {
 			packet.Data = chunk.Data
 		}
diff --git a/api/stream.go b/api/stream.go
index 9ccb6b1..79d0596 100644
--- a/api/stream.go
+++ b/api/stream.go
@@ -57,12 +57,13 @@ var streamHandler handler = func(w http.ResponseWriter, r *http.Request) *errorR
 	wf := &streamableResponseWriter{f}
 	f.Flush()
 
-	// Ensure it's closed when the context is cancelled.
-	go func() {
-		<-r.Context().Done()
-		activity.Record(ctx, "API: Stream %v closed by completed context: %v", path, rdr.Close())
-	}()
-
+	if closer, ok := rdr.(io.Closer); ok {
+		// If a ReadCloser, ensure it's closed when the context is cancelled.
+		go func() {
+			<-r.Context().Done()
+			activity.Record(ctx, "API: Stream %v closed by completed context: %v", path, closer.Close())
+		}()
+	}
 	if _, err := io.Copy(wf, rdr); err != nil {
 		// Common for copy to error when the caller closes the connection.
 		log.Debugf("Errored streaming response for entry %v: %v", path, err)
diff --git a/api/types/exec.go b/api/types/exec.go
index 58e44c0..06d69ae 100644
--- a/api/types/exec.go
+++ b/api/types/exec.go
@@ -1,7 +1,6 @@
 package apitypes
 
 import (
-	"github.com/puppetlabs/wash/plugin"
 	"time"
 )
 
@@ -31,11 +30,14 @@ type ExecBody struct {
 	Opts ExecOptions `json:"opts"`
 }
 
-// Enumerates packet types used by the API.
+// ExecPacketType identifies the packet type.
+type ExecPacketType = string
+
+// Enumerates packet types.
 const (
-	Stdout   plugin.ExecPacketType = plugin.Stdout
-	Stderr   plugin.ExecPacketType = plugin.Stderr
-	Exitcode plugin.ExecPacketType = "exitcode"
+	Stdout   ExecPacketType = "stdout"
+	Stderr   ExecPacketType = "stderr"
+	Exitcode ExecPacketType = "exitcode"
 )
 
 // ExecPacket is a single packet of results from an exec.
@@ -44,8 +46,8 @@ const (
 //
 // swagger:response
 type ExecPacket struct {
-	TypeField plugin.ExecPacketType `json:"type"`
-	Timestamp time.Time             `json:"timestamp"`
-	Data      interface{}           `json:"data"`
-	Err       *ErrorObj             `json:"error"`
+	TypeField ExecPacketType `json:"type"`
+	Timestamp time.Time      `json:"timestamp"`
+	Data      interface{}    `json:"data"`
+	Err       *ErrorObj      `json:"error"`
 }
diff --git a/cmd/internal/find/grammar/atom.go b/cmd/internal/find/grammar/atom.go
deleted file mode 100644
index 6cf80c8..0000000
--- a/cmd/internal/find/grammar/atom.go
+++ /dev/null
@@ -1,29 +0,0 @@
-package grammar
-
-import "github.com/puppetlabs/wash/cmd/internal/find/types"
-
-// Atom represents an atom in `wash find`'s expression grammar.
-type Atom struct {
-	Tokens []string
-	// tokens[0] will always include the atom's token that the user
-	// passed-in
-	Parse func(tokens []string) (types.Predicate, []string, error)
-}
-
-// Atoms is a map of <token> => <atom>. This is populated by newAtom.
-var Atoms = make(map[string]*Atom)
-
-// NewAtom creates a new atom. When creating a new atom with this function
-// via a package variable assignment, be sure to comment nolint above the
-// variable so that CI does not mark it as unused. See notOp in find/operators.go
-// for an example.
-func NewAtom(tokens []string, parse func(tokens []string) (types.Predicate, []string, error)) *Atom {
-	a := &Atom{
-		Tokens: tokens,
-		Parse:  parse,
-	}
-	for _, t := range tokens {
-		Atoms[t] = a
-	}
-	return a
-}
diff --git a/cmd/internal/find/grammar/binaryOp.go b/cmd/internal/find/grammar/binaryOp.go
deleted file mode 100644
index 515cfba..0000000
--- a/cmd/internal/find/grammar/binaryOp.go
+++ /dev/null
@@ -1,29 +0,0 @@
-package grammar
-
-import "github.com/puppetlabs/wash/cmd/internal/find/types"
-
-// BinaryOp represents a binary operator in a `wash find` expression
-type BinaryOp struct {
-	Tokens     []string
-	Precedence int
-	Combine    func(p1 types.Predicate, p2 types.Predicate) types.Predicate
-}
-
-// BinaryOps is a map of <token> => <binaryOp>. This is populated by NewBinaryOp.
-var BinaryOps = make(map[string]*BinaryOp)
-
-// NewBinaryOp creates a new binary op. When creating a new binary op with this function
-// via a package variable assignment, be sure to comment nolint above the
-// variable so that CI does not mark it as unused. See andOp in find/operators.go
-// for an example.
-func NewBinaryOp(tokens []string, precedence int, combine func(p1 types.Predicate, p2 types.Predicate) types.Predicate) *BinaryOp {
-	b := &BinaryOp{
-		Tokens:     tokens,
-		Precedence: precedence,
-		Combine:    combine,
-	}
-	for _, t := range tokens {
-		BinaryOps[t] = b
-	}
-	return b
-}
diff --git a/cmd/internal/find/primary/errz/match.go b/cmd/internal/find/parser/errz/match.go
similarity index 94%
rename from cmd/internal/find/primary/errz/match.go
rename to cmd/internal/find/parser/errz/match.go
index af11483..bee5d0a 100644
--- a/cmd/internal/find/primary/errz/match.go
+++ b/cmd/internal/find/parser/errz/match.go
@@ -1,7 +1,7 @@
 package errz
 
 // MatchError represents the case when the input tokens did not
-// match the given parser.
+// match a given parser.
 type MatchError struct {
 	reason string
 }
diff --git a/cmd/internal/find/parser/expression/atom.go b/cmd/internal/find/parser/expression/atom.go
new file mode 100644
index 0000000..8b69b97
--- /dev/null
+++ b/cmd/internal/find/parser/expression/atom.go
@@ -0,0 +1,50 @@
+package expression
+
+import "github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+import "github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+import "fmt"
+
+func notOpParser(parser *Parser) predicate.Parser {
+	return predicate.ToParser(func(tokens []string) (predicate.Predicate, []string, error) {
+		notToken := tokens[0]
+		if notToken != "!" && notToken != "-not" {
+			return nil, nil, errz.NewMatchError("expected ! or -not")
+		}
+		tokens = tokens[1:]
+		if len(tokens) == 0 {
+			return nil, nil, fmt.Errorf("%v: no following expression", notToken)
+		}
+		if tokens[0] == ")" {
+			if parser.numOpenParens <= 0 {
+				return nil, nil, fmt.Errorf("): no beginning '('")
+			}
+			return nil, nil, fmt.Errorf("%v: no following expression", notToken)
+		}
+		p, tokens, err := parser.atom.Parse(tokens)
+		if err != nil {
+			if errz.IsMatchError(err) {
+				err = fmt.Errorf("%v: no following expression", notToken)
+			}
+			return nil, nil, err
+		}
+		return p.Negate(), tokens, err
+	})
+}
+
+func parensOpParser(parser *Parser) predicate.Parser {
+	return predicate.ToParser(func(tokens []string) (predicate.Predicate, []string, error) {
+		if tokens[0] != "(" {
+			return nil, nil, errz.NewMatchError("expected an '('")
+		}
+		tokens = tokens[1:]
+		parser.numOpenParens++
+		// Save the current evaluation stack. We will restore it after parsing
+		// the parenthesized expression
+		stack := parser.stack
+		defer func() {
+			parser.stack = stack
+			parser.numOpenParens--
+		}()
+		return parser.Parse(tokens)
+	})
+}
\ No newline at end of file
diff --git a/cmd/internal/find/parser/expression/binaryOp.go b/cmd/internal/find/parser/expression/binaryOp.go
new file mode 100644
index 0000000..db47e6e
--- /dev/null
+++ b/cmd/internal/find/parser/expression/binaryOp.go
@@ -0,0 +1,26 @@
+package expression
+
+import "github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+
+// BinaryOp represents a binary operator in a predicate expression
+type BinaryOp struct {
+	tokens     []string
+	precedence int
+	combine    func(p1 predicate.Predicate, p2 predicate.Predicate) predicate.Predicate
+}
+
+func newBinaryOp(tokens []string, precedence int, combine func(p1 predicate.Predicate, p2 predicate.Predicate) predicate.Predicate) *BinaryOp {
+	return &BinaryOp{
+		tokens:     tokens,
+		precedence: precedence,
+		combine:    combine,
+	}
+}
+
+var andOp = newBinaryOp([]string{"-a", "-and"}, 1, func(p1 predicate.Predicate, p2 predicate.Predicate) predicate.Predicate {
+		return p1.And(p2)
+})
+
+var orOp = newBinaryOp([]string{"-o", "-or"}, 0, func(p1 predicate.Predicate, p2 predicate.Predicate) predicate.Predicate {
+		return p1.Or(p2)
+})
diff --git a/cmd/internal/find/parser/expression/error.go b/cmd/internal/find/parser/expression/error.go
new file mode 100644
index 0000000..21fe21f
--- /dev/null
+++ b/cmd/internal/find/parser/expression/error.go
@@ -0,0 +1,15 @@
+package expression
+
+import (
+	"fmt"
+)
+
+// UnknownTokenError represents an unknown token that was found
+// when parsing the expression
+type UnknownTokenError struct {
+	Token string
+}
+
+func (e UnknownTokenError) Error() string {
+	return fmt.Sprintf("unknown token %v", e.Token)
+}
\ No newline at end of file
diff --git a/cmd/internal/find/parser/expression/parser.go b/cmd/internal/find/parser/expression/parser.go
new file mode 100644
index 0000000..aae39dd
--- /dev/null
+++ b/cmd/internal/find/parser/expression/parser.go
@@ -0,0 +1,243 @@
+package expression
+
+import (
+	"fmt"
+
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/golang-collections/collections/stack"
+)
+
+/*
+Parser is a predicate parser that parses predicate expressions. Expressions
+have the following grammar:
+  Expression => Expression (-a|-and) Atom |
+                Expression Atom           | 
+                Expression (-o|-or)  Atom |
+                Atom
+
+  Atom       => (!|-not) Atom             |
+                '(' Expression ')'        |
+                Predicate
+
+where 'Expression Atom' is semantically equivalent to 'Expression -a Atom'.
+The grammar for Predicate is caller-specific.
+
+Operator precedence is (from highest to lowest):
+  ()
+  -not
+  -and
+  -or
+
+The precedence of the () and -not operators is already enforced by the grammar.
+Precedence of the binary operators -and and -or is enforced by maintaining an
+evaluation stack.
+*/
+type Parser struct {
+	// Storing the binary ops this way makes it easier for us to add the capability
+	// for callers to extend the parser if they'd like to support additional binary
+	// ops. We will likely need this capability in the future if/when we add the ","
+	// operator to `wash find`.
+	binaryOps map[string]*BinaryOp
+	atom *predicate.CompositeParser
+	stack *evalStack
+	numOpenParens int
+	opTokens map[string]struct{}
+}
+
+// NewParser returns a new predicate expression parser. The passed-in
+// predicateParser must be able to parse the "Predicate" nonterminal
+// in the expression grammar.
+func NewParser(predicateParser predicate.Parser) *Parser {
+	p := &Parser{}
+	p.binaryOps = make(map[string]*BinaryOp)
+	p.opTokens = map[string]struct{}{
+		"!": struct{}{},
+		"-not": struct{}{},
+		"(": struct{}{},
+		")": struct{}{},
+	}
+	for _, op := range []*BinaryOp{andOp, orOp} {
+		for _, token := range op.tokens {
+			p.binaryOps[token] = op
+			p.opTokens[token] = struct{}{}
+		}
+	}
+	p.atom = &predicate.CompositeParser{
+		ErrMsg: "expected an atom",
+		Parsers: []predicate.Parser{
+			notOpParser(p),
+			parensOpParser(p),
+			predicateParser,
+		},
+	}
+	return p
+}
+
+// IsOp returns true if the given token represents the parentheses operator,
+// the not operator, or a binary operator.
+func (parser *Parser) IsOp(token string) bool {
+	_, ok := parser.opTokens[token]
+	return ok
+}
+
+/*
+Parse parses a predicate expression captured by the given tokens. It will process
+the tokens until it either (1) exhausts the input tokens, (2) stumbles upon a
+a token that it cannot parse, or (3) finds a syntax error. For Cases (1) and (2),
+Parse will return a syntax error if it did not parse a predicate. Otherwise, it will
+return the parsed predicate + any remaining tokens. Case (2) will also return an
+UnknownTokenError containing the offending token.
+
+Case 2's useful if we're parsing an expression inside an expression. It lets the caller
+decide if they've finished parsing the inner expression. We will take advantage of Case 2
+when parsing `meta` primary expressions.
+*/
+func (parser *Parser) Parse(tokens []string) (predicate.Predicate, []string, error) {
+	parser.stack = newEvalStack()
+
+	// Declare these as variables so that we can cleanly update
+	// err for each iteration without having to worry about the
+	// := operator's scoping rules. tks is used to avoid accidentally
+	// overwriting tokens.
+	//
+	// POST-LOOP INVARIANT: err == nil or err is an UnknownTokenError
+	var p predicate.Predicate
+	var tks []string
+	var err error
+	for {
+		// Reset err in each iteration to maintain the post-loop invariant
+		err = nil
+		if len(tokens) == 0 {
+			if parser.numOpenParens > 0 {
+				return nil, nil, fmt.Errorf("(: missing closing ')'")
+			}
+			break
+		}
+		token := tokens[0]
+		if token == ")" {
+			if parser.numOpenParens <= 0 {
+				return nil, nil, fmt.Errorf("): no beginning '('")
+			}
+			// We're reached the end of a parenthesized expression, so shift tokens
+			// and break out of the loop
+			tokens = tokens[1:]
+			break
+		}
+		// Try parsing an atom first.
+		p, tks, err = parser.atom.Parse(tokens)
+		if err == nil {
+			// Successfully parsed an atom, so push the parsed predicate onto the stack.
+			parser.stack.pushPredicate(p)
+			tokens = tks
+			continue
+		}
+		if !errz.IsMatchError(err) {
+			// Syntax error when parsing the atom, so return the error
+			return nil, nil, err
+		}
+		// Parsing an atom didn't work, so try parsing a binaryOp
+		b, ok := parser.binaryOps[token]
+		if !ok {
+			// Found an unknown token. Break out of the loop to evaluate
+			// the final predicate.
+			err = UnknownTokenError{token}
+			break	
+		}
+		// Parsed a binaryOp, so shift tokens and push the op on the evaluation stack.
+		tokens = tokens[1:]
+		if parser.stack.mostRecentOp == nil {
+			if _, ok := parser.stack.Peek().(predicate.Predicate); !ok {
+				return nil, nil, fmt.Errorf("%v: no expression before %v", token, token)
+			}
+			parser.stack.pushBinaryOp(token, b)
+			continue
+		}
+		if _, ok := parser.stack.Peek().(*BinaryOp); ok {
+			// mostRecentOp's on the stack, and the parser's asking us to
+			// push b. This means that mostRecentOp did not have an expression
+			// after it, so report the syntax error.
+			return nil, nil, fmt.Errorf(
+				"%v: no expression after %v",
+				parser.stack.mostRecentOpToken,
+				parser.stack.mostRecentOpToken,
+			)
+		}
+		parser.stack.pushBinaryOp(token, b)
+	}
+	// Parsing's finished.
+	if parser.stack.Len() <= 0 {
+		// We didn't parse anything. Either we have an empty expression, or
+		// err is an UnknownTokenError
+		if err == nil {
+			// We have an empty expression
+			if parser.numOpenParens > 0 {
+				err = fmt.Errorf("(): empty inner expression")
+			} else {
+				err = fmt.Errorf("empty expression")
+			}
+			return nil, nil, err
+		}
+		// err is an UnknownTokenError
+		return nil, tokens, err
+	}
+	if _, ok := parser.stack.Peek().(*BinaryOp); ok {
+		// This codepath is possible via something like "p1 -and" or "p1 -and <unknown_token>"
+		if err != nil {
+			// We have "p1 -and <unknown_token>"
+			return nil, nil, err
+		}
+		return nil, nil, fmt.Errorf(
+			"%v: no expression after %v",
+			parser.stack.mostRecentOpToken,
+			parser.stack.mostRecentOpToken,
+		)
+	}
+	// Call s.evaluate() to handle cases like "p1 -and p2"
+	parser.stack.evaluate()
+	return parser.stack.Pop().(predicate.Predicate), tokens, err
+}
+
+type evalStack struct {
+	*stack.Stack
+	mostRecentOp *BinaryOp
+	mostRecentOpToken string
+}
+
+func newEvalStack() *evalStack {
+	return &evalStack{
+		Stack: stack.New(),
+	}
+}
+
+func (s *evalStack) pushBinaryOp(token string, b *BinaryOp) {
+	// Invariant: s.Peek() returns a predicate.Predicate type.
+	if s.mostRecentOp != nil {
+		if b.precedence <= s.mostRecentOp.precedence {
+			s.evaluate()
+		}
+	}
+	s.mostRecentOp = b
+	s.mostRecentOpToken = token
+	s.Push(b)
+}
+
+func (s *evalStack) pushPredicate(p predicate.Predicate) {
+	if _, ok := s.Peek().(predicate.Predicate); ok {
+		// We have p1 p2, where p1 == s.Peek() and p2 = p. Since p1 p2 == p1 -and p2,
+		// push andOp before pushing p2.
+		s.pushBinaryOp(andOp.tokens[0], andOp)
+	}
+	s.Push(p)
+}
+
+func (s *evalStack) evaluate() {
+	// Invariant: s's layout is something like "p (<op> p)*"
+	for s.Len() > 1 {
+		p2 := s.Pop().(predicate.Predicate)
+		op := s.Pop().(*BinaryOp)
+		p1 := s.Pop().(predicate.Predicate)
+		s.Push(op.combine(p1, p2))
+	}
+}
+
diff --git a/cmd/internal/find/parser/operators.go b/cmd/internal/find/parser/operators.go
deleted file mode 100644
index 3f7f200..0000000
--- a/cmd/internal/find/parser/operators.go
+++ /dev/null
@@ -1,73 +0,0 @@
-package parser
-
-import (
-	"fmt"
-
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
-	"github.com/puppetlabs/wash/cmd/internal/find/types"
-)
-
-//nolint
-var notOp = grammar.NewAtom([]string{"!", "-not"}, func(tokens []string) (types.Predicate, []string, error) {
-	notToken := tokens[0]
-	tokens = tokens[1:]
-	if len(tokens) == 0 {
-		return nil, nil, fmt.Errorf("%v: no following expression", notToken)
-	}
-	token := tokens[0]
-	if atom, ok := grammar.Atoms[token]; ok {
-		p, tokens, err := atom.Parse(tokens)
-		if err != nil {
-			return nil, nil, err
-		}
-		return func(e types.Entry) bool {
-			return !p(e)
-		}, tokens, err
-	}
-	return nil, nil, fmt.Errorf("%v: no following expression", notToken)
-})
-
-//nolint
-var parensOp = grammar.NewAtom([]string{"("}, func(tokens []string) (types.Predicate, []string, error) {
-	// Find the ")" that's paired with our "(". Use the algorithm
-	// described in https://stackoverflow.com/questions/12752225/how-do-i-find-the-position-of-matching-parentheses-or-braces-in-a-given-piece-of
-	// Note that we do not have to check for balanced parentheses here because
-	// that check was already done in the top-level parse method.
-	//
-	// TODO: Could optimize this by moving parens handling over to the evaluation
-	// stack. Not important right now because expressions to `wash find` will likely
-	// be simple.
-	tokens = tokens[1:]
-	ix := 0
-	counter := 1
-	for i, token := range tokens {
-		if token == "(" {
-			counter++
-		} else if token == ")" {
-			counter--
-			ix = i
-		}
-		if counter == 0 {
-			break
-		}
-	}
-	if ix == 0 {
-		return nil, nil, fmt.Errorf("(): empty inner expression")
-	}
-	p, err := parseExpressionHelper(tokens[:ix])
-	return p, tokens[ix+1:], err
-})
-
-//nolint
-var andOp = grammar.NewBinaryOp([]string{"-a", "-and"}, 1, func(p1 types.Predicate, p2 types.Predicate) types.Predicate {
-	return func(e types.Entry) bool {
-		return p1(e) && p2(e)
-	}
-})
-
-//nolint
-var orOp = grammar.NewBinaryOp([]string{"-o", "-or"}, 0, func(p1 types.Predicate, p2 types.Predicate) types.Predicate {
-	return func(e types.Entry) bool {
-		return p1(e) || p2(e)
-	}
-})
diff --git a/cmd/internal/find/parser/parse.go b/cmd/internal/find/parser/parse.go
index 6feff31..5985707 100644
--- a/cmd/internal/find/parser/parse.go
+++ b/cmd/internal/find/parser/parse.go
@@ -2,6 +2,7 @@ package parser
 
 import (
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 )
 
 // Result represents the result of parsing `wash find`'s
@@ -9,7 +10,7 @@ import (
 type Result struct {
 	Path      string
 	Options   types.Options
-	Predicate types.Predicate
+	Predicate predicate.Entry
 }
 
 /*
diff --git a/cmd/internal/find/parser/parseExpression.go b/cmd/internal/find/parser/parseExpression.go
index 78d11a7..44ee9c2 100644
--- a/cmd/internal/find/parser/parseExpression.go
+++ b/cmd/internal/find/parser/parseExpression.go
@@ -3,45 +3,17 @@ package parser
 import (
 	"fmt"
 
-	"github.com/golang-collections/collections/stack"
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/expression"
 	"github.com/puppetlabs/wash/cmd/internal/find/primary"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
 
 /*
-An expression is described by the following grammar
-	Expression => Expression (-a|-and) Atom |
-	              Expression Atom           |
-		          Expression (-o|-or) Atom  |
-		          Atom                      |
-
-	      Atom => (!|-not) Atom             |
-	              '(' Expression ')'        |
-		          Primary
-
-where 'Expression Atom' is semantically equivalent to 'Expression -a Atom'.
-Primaries have their own grammar. See the corresponding primary/*.go files
-for more details.
-
-Operator precedence is (from highest to lowest):
-	()
-	-not
-	-and
-	-or
-
-The precedence of the () and -not operators is already enforced by the grammar.
-Precedence of the binary operators -and and -or is enforced by maintaining an
-evaluation stack.
+See the comments of expression.Parser#Parse for the grammar. Substitute
+Predicate with Primary.
 */
-func parseExpression(tokens []string) (types.Predicate, error) {
-	// Each primary's declared as a package variable, so they'll
-	// be loaded when the primary package is loaded. The primary package
-	// is only loaded if it is consumed by another package, so we
-	// force it to load here by calling primary.LoadAll(). primary.LoadAll()
-	// noops, so it is safe to call it multiple times.
-	primary.LoadAll()
-
+func parseExpression(tokens []string) (predicate.Entry, error) {
 	if len(tokens) == 0 {
 		// tokens is empty, meaning the user did not provide an expression
 		// to `wash find`. Thus, we default to a predicate that always returns
@@ -50,108 +22,18 @@ func parseExpression(tokens []string) (types.Predicate, error) {
 			return true
 		}, nil
 	}
-
-	// Validate that the parentheses are correctly balanced.
-	// We do this outside of parseExpressionHelper to avoid
-	// redundant validation when parensOp recurses into it.
-	s := stack.New()
-	for _, token := range tokens {
-		if token == "(" {
-			s.Push(token)
-		} else if token == ")" {
-			if s.Len() == 0 {
-				return nil, fmt.Errorf("): no beginning '('")
-			}
-			s.Pop()
-		}
-	}
-	if s.Len() > 0 {
-		return nil, fmt.Errorf("(: missing closing ')'")
-	}
-	return parseExpressionHelper(tokens)
-}
-
-func parseExpressionHelper(tokens []string) (types.Predicate, error) {
-	if len(tokens) == 0 {
-		panic("parseExpressionHelper: called with len(tokens) == 0")
-	}
-
-	s := newEvalStack()
-	var mostRecentOp *grammar.BinaryOp
-	var mostRecentOpToken string
-	pushBinaryOp := func(token string, b *grammar.BinaryOp) {
-		// Invariant: s.Peek() returns a types.Predicate
-		if mostRecentOp != nil {
-			if b.Precedence <= mostRecentOp.Precedence {
-				s.evaluate()
-			}
-		}
-		mostRecentOp = b
-		mostRecentOpToken = token
-		s.Push(b)
-	}
-	for len(tokens) > 0 {
-		token := tokens[0]
-
-		// Declare these as variables so that we can cleanly update the
-		// tokens parameter for each iteration. Otherwise, := will create a
-		// new tokens variable within the if statement's scope.
-		var p types.Predicate
-		var err error
-		if atom, ok := grammar.Atoms[token]; ok {
-			p, tokens, err = atom.Parse(tokens)
-			if err != nil {
-				return nil, err
-			}
-			if _, ok := s.Peek().(types.Predicate); ok {
-				// We have p1 p2, where p1 == s.Peek() and p2 = p. Since p1 p2 == p1 -and p2,
-				// push andOp before pushing p2.
-				pushBinaryOp("-a", andOp)
-			}
-			s.Push(p)
-		} else if b, ok := grammar.BinaryOps[token]; ok {
-			tokens = tokens[1:]
-			if mostRecentOp == nil {
-				if _, ok := s.Peek().(types.Predicate); !ok {
-					return nil, fmt.Errorf("%v: no expression before %v", token, token)
-				}
-				pushBinaryOp(token, b)
-				continue
-			}
-			if _, ok := s.Peek().(*grammar.BinaryOp); ok {
-				// mostRecentOp's on the stack, and the parser's asking us to
-				// push b. This means that mostRecentOp did not have an expression
-				// after it, so report the error.
-				return nil, fmt.Errorf("%v: no expression after %v", mostRecentOpToken, mostRecentOpToken)
-			}
-			pushBinaryOp(token, b)
-		} else {
-			return nil, fmt.Errorf("%v: unknown primary or operator", token)
+	parser := expression.NewParser(primary.Parser)
+	p, tks, err := parser.Parse(tokens)
+	if err != nil {
+		if tkErr, ok := err.(expression.UnknownTokenError); ok {
+			err = fmt.Errorf("%v: unknown primary or operator", tkErr.Token)
 		}
+		return nil, err
 	}
-	if _, ok := s.Peek().(*grammar.BinaryOp); ok {
-		// This codepath is possible via something like "p1 -and"
-		return nil, fmt.Errorf("%v: no expression after %v", mostRecentOpToken, mostRecentOpToken)
-	}
-	// Call s.evaluate() to handle cases like "p1 -and p2"
-	s.evaluate()
-	return s.Pop().(types.Predicate), nil
-}
-
-type evalStack struct {
-	*stack.Stack
-}
-
-func newEvalStack() *evalStack {
-	return &evalStack{&stack.Stack{}}
-}
-
-func (s *evalStack) evaluate() {
-	// Invariant: s's layout is something like "p (<op> p)*"
-	for s.Len() > 1 {
-		p2 := s.Pop().(types.Predicate)
-		op := s.Pop().(*grammar.BinaryOp)
-		p1 := s.Pop().(types.Predicate)
-		s.Push(op.Combine(p1, p2))
+	if len(tks) != 0 {
+		// This should never happen, but better safe than sorry
+		msg := fmt.Sprintf("parser.parseExpression(): returned a non-empty set of tokens: %v", tks)
+		panic(msg)
 	}
-}
+	return p.(predicate.Entry), nil
+}
\ No newline at end of file
diff --git a/cmd/internal/find/parser/parseOptions.go b/cmd/internal/find/parser/parseOptions.go
index 69ad063..4acb5b2 100644
--- a/cmd/internal/find/parser/parseOptions.go
+++ b/cmd/internal/find/parser/parseOptions.go
@@ -1,7 +1,8 @@
 package parser
 
 import (
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/expression"
+	"github.com/puppetlabs/wash/cmd/internal/find/primary"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
 
@@ -11,14 +12,14 @@ func parseOptions(args []string) (types.Options, []string, error) {
 
 	// Find the index of the first non-option arg that FlagSet
 	// doesn't know about. This is either "--", the special flag
-	// termination symbol, or an atom/binary op, which indicates
+	// termination symbol, or a primary/operator, which indicates
 	// the beginning of a `wash find` expression. These non-option
 	// args will be handled/processed in parseExpression. Note that
 	// without this code, FlagSet would interpret "--" as the flag
 	// termination symbol, which is bad because "--" is invalid
-	// `wash find` syntax. It would also try to parse our atoms/binary
-	// ops (like -true, -false, -not, -and) as options, which is also
-	// bad.
+	// `wash find` syntax. It would also try to parse our primaries
+	// and operators (like -true, -false, -not, -and) as options, which is
+	// also bad.
 	//
 	// All other cases are properly handled by fs.Parse(). For example,
 	// if args is of the form ["-mindepth", "0", "foo", "bar", "baz"],
@@ -27,13 +28,7 @@ func parseOptions(args []string) (types.Options, []string, error) {
 	// arguments.
 	var endIx int
 	for _, arg := range args {
-		if arg == "--" {
-			break
-		}
-		if _, ok := grammar.Atoms[arg]; ok {
-			break
-		}
-		if _, ok := grammar.BinaryOps[arg]; ok {
+		if nonOptionArg(arg) {
 			break
 		}
 		endIx++
@@ -58,3 +53,8 @@ func parseOptions(args []string) (types.Options, []string, error) {
 
 	return o, args, nil
 }
+
+func nonOptionArg(arg string) bool {
+	parser := expression.NewParser(primary.Parser)
+	return arg == "--" || parser.IsOp(arg) || primary.Parser.IsPrimary(arg)
+}
diff --git a/cmd/internal/find/parser/predicate/entry.go b/cmd/internal/find/parser/predicate/entry.go
new file mode 100644
index 0000000..cd1440c
--- /dev/null
+++ b/cmd/internal/find/parser/predicate/entry.go
@@ -0,0 +1,35 @@
+package predicate
+
+import "github.com/puppetlabs/wash/cmd/internal/find/types"
+
+// Entry represents a predicate on a Wash entry.
+type Entry func(types.Entry) bool
+
+// And returns p1 && p2
+func (p1 Entry) And(p2 Predicate) Predicate {
+	return Entry(func(e types.Entry) bool {
+		return p1(e) && (p2.(Entry))(e)
+	})
+}
+
+// Or returns p1 || p2
+func (p1 Entry) Or(p2 Predicate) Predicate {
+	return Entry(func(e types.Entry) bool {
+		return p1(e) || (p2.(Entry))(e)
+	})
+}
+
+// Negate returns Not(p1)
+func (p1 Entry) Negate() Predicate {
+	return Entry(func(e types.Entry) bool {
+		return !p1(e)
+	})
+}
+
+// EntryParser is a type that parses Entry predicates
+type EntryParser func(tokens []string) (Entry, []string, error)
+
+// Parse parses an Entry predicate from the given input.
+func (parser EntryParser) Parse(tokens []string) (Predicate, []string, error) {
+	return parser(tokens)
+}
\ No newline at end of file
diff --git a/cmd/internal/find/parser/predicate/generic.go b/cmd/internal/find/parser/predicate/generic.go
new file mode 100644
index 0000000..b2e6b6b
--- /dev/null
+++ b/cmd/internal/find/parser/predicate/generic.go
@@ -0,0 +1,33 @@
+package predicate
+
+// Generic represents a predicate on an interface{}
+type Generic func(interface{}) bool
+
+// And returns p1 && p2
+func (p1 Generic) And(p2 Predicate) Predicate {
+	return Generic(func(v interface{}) bool {
+		return p1(v) && (p2.(Generic))(v)
+	})
+}
+
+// Or returns p1 || p2
+func (p1 Generic) Or(p2 Predicate) Predicate {
+	return Generic(func(v interface{}) bool {
+		return p1(v) || (p2.(Generic))(v)
+	})
+}
+
+// Negate returns Not(p1)
+func (p1 Generic) Negate() Predicate {
+	return Generic(func(v interface{}) bool {
+		return !p1(v)
+	})
+}
+
+// GenericParser is a type that parses Generic predicates
+type GenericParser func(tokens []string) (Generic, []string, error)
+
+// Parse parses a Generic predicate from the given input.
+func (parser GenericParser) Parse(tokens []string) (Predicate, []string, error) {
+	return parser(tokens)
+}
\ No newline at end of file
diff --git a/cmd/internal/find/parser/predicate/int.go b/cmd/internal/find/parser/predicate/int.go
new file mode 100644
index 0000000..1bba7d8
--- /dev/null
+++ b/cmd/internal/find/parser/predicate/int.go
@@ -0,0 +1,33 @@
+package predicate
+
+// Int represents a predicate on an integer
+type Int func(int64) bool
+
+// And returns p1 && p2
+func (p1 Int) And(p2 Predicate) Predicate {
+	return Int(func(n int64) bool {
+		return p1(n) && (p2.(Int))(n)
+	})
+}
+
+// Or returns p1 || p2
+func (p1 Int) Or(p2 Predicate) Predicate {
+	return Int(func(n int64) bool {
+		return p1(n) || (p2.(Int))(n)
+	})
+}
+
+// Negate returns Not(p1)
+func (p1 Int) Negate() Predicate {
+	return Int(func(n int64) bool {
+		return !p1(n)
+	})
+}
+
+// IntParser is a type that parses Int predicates
+type IntParser func(tokens []string) (Int, []string, error)
+
+// Parse parses an Int predicate from the given input.
+func (parser IntParser) Parse(tokens []string) (Predicate, []string, error) {
+	return parser(tokens)
+}
\ No newline at end of file
diff --git a/cmd/internal/find/parser/predicate/parser.go b/cmd/internal/find/parser/predicate/parser.go
new file mode 100644
index 0000000..9589a66
--- /dev/null
+++ b/cmd/internal/find/parser/predicate/parser.go
@@ -0,0 +1,54 @@
+package predicate
+
+import "github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+
+// Parser represents a parser that parses predicates.
+type Parser interface {
+	Parse(tokens []string) (Predicate, []string, error)
+}
+
+// CompositeParser represents a parser composed of multiple predicate
+// parsers.
+type CompositeParser struct {
+	ErrMsg string
+	Parsers []Parser
+}
+
+// Parse is a wrapper to ParseAndReturnParserID. It implements the predicate.Parser
+// interface for a CompositeParser.
+func (cp CompositeParser) Parse(tokens []string) (Predicate, []string, error) {
+	p, tokens, _, err := cp.ParseAndReturnParserID(tokens)
+	return p, tokens, err
+}
+
+// ParseAndReturnParserID attempts to parse a predicate from the given tokens. It loops
+// through each of cp's parsers, returning the result of the first parser that matches
+// the input, and the matching parser's ID. If no parser matches the input, then Parse
+// returns a MatchError containing cp.ErrMsg
+func (cp CompositeParser) ParseAndReturnParserID(tokens []string) (Predicate, []string, int, error) {
+	for i, parser := range cp.Parsers {
+		p, tokens, err := parser.Parse(tokens)
+		if err == nil {
+			return p, tokens, i, nil
+		}
+		if !errz.IsMatchError(err) {
+			// Parser matched the input, but returned a syntax error. Return
+			// the error.
+			return nil, nil, i, err
+		}
+	}
+	// None of the parsers matched the input, so return a MatchError
+	return nil, nil, -1, errz.NewMatchError(cp.ErrMsg)		
+}
+
+// ToParser converts the given parse function to a predicate.Parser object
+func ToParser(parseFunc func(tokens []string) (Predicate, []string, error)) Parser {
+	return predicateParser(parseFunc)
+}
+
+type predicateParser func(tokens []string) (Predicate, []string, error)
+
+// Parse parses a predicate from the given tokens
+func (p predicateParser) Parse(tokens []string) (Predicate, []string, error) {
+	return p(tokens)
+}
diff --git a/cmd/internal/find/parser/predicate/predicate.go b/cmd/internal/find/parser/predicate/predicate.go
new file mode 100644
index 0000000..8ca53a6
--- /dev/null
+++ b/cmd/internal/find/parser/predicate/predicate.go
@@ -0,0 +1,15 @@
+package predicate
+
+// Predicate is an interface representing a predicate.
+// It is used to generalize the expression parsing logic
+// used by the top-level parser and the meta primary.
+//
+// TODO: This interface is here because Go 1 doesn't support
+// generics. Go 2 is planning to introduce some basic generic
+// support, so we should re-evaluate this approach once it is
+// released.
+type Predicate interface {
+	And(Predicate) Predicate
+	Or(Predicate) Predicate
+	Negate() Predicate
+}
\ No newline at end of file
diff --git a/cmd/internal/find/primary/action.go b/cmd/internal/find/primary/action.go
index ca1bade..f959734 100644
--- a/cmd/internal/find/primary/action.go
+++ b/cmd/internal/find/primary/action.go
@@ -5,16 +5,15 @@ import (
 	"strings"
 
 	"github.com/puppetlabs/wash/plugin"
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
 
 // actionPrimary => <action>
 //nolint
-var actionPrimary = grammar.NewAtom([]string{"-action"}, func(tokens []string) (types.Predicate, []string, error) {
-	tokens = tokens[1:]
+var actionPrimary = Parser.newPrimary([]string{"-action"}, func(tokens []string) (predicate.Entry, []string, error) {
 	if len(tokens) == 0 {
-		return nil, nil, fmt.Errorf("-action: requires additional arguments")
+		return nil, nil, fmt.Errorf("requires additional arguments")
 	}
 	validActions := plugin.Actions()
 	action, ok := validActions[tokens[0]]
@@ -25,7 +24,7 @@ var actionPrimary = grammar.NewAtom([]string{"-action"}, func(tokens []string) (
 			validActionsArray = append(validActionsArray, actionName)
 		}
 		validActionsStr := strings.Join(validActionsArray, ", ")
-		return nil, nil, fmt.Errorf("-action: %v is an invalid action. Valid actions are %v", tokens[0], validActionsStr)
+		return nil, nil, fmt.Errorf("%v is an invalid action. Valid actions are %v", tokens[0], validActionsStr)
 	}
 	p := func(e types.Entry) bool {
 		return e.Supports(action)
diff --git a/cmd/internal/find/primary/boolean.go b/cmd/internal/find/primary/boolean.go
index ba3d7bc..a0897ea 100644
--- a/cmd/internal/find/primary/boolean.go
+++ b/cmd/internal/find/primary/boolean.go
@@ -3,16 +3,16 @@ package primary
 import (
 	"fmt"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
 
 //nolint
-func newBooleanPrimary(val bool) *grammar.Atom {
-	return grammar.NewAtom([]string{fmt.Sprintf("-%v", val)}, func(tokens []string) (types.Predicate, []string, error) {
+func newBooleanPrimary(val bool) *primary {
+	return Parser.newPrimary([]string{fmt.Sprintf("-%v", val)}, func(tokens []string) (predicate.Entry, []string, error) {
 		return func(e types.Entry) bool {
 			return val
-		}, tokens[1:], nil
+		}, tokens, nil
 	})
 }
 
diff --git a/cmd/internal/find/primary/core.go b/cmd/internal/find/primary/core.go
deleted file mode 100644
index ff88a58..0000000
--- a/cmd/internal/find/primary/core.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package primary
-
-// LoadAll loads all of `wash find`'s primaries.
-func LoadAll() {
-}
diff --git a/cmd/internal/find/primary/meta.go b/cmd/internal/find/primary/meta.go
index b0e0011..679a02c 100644
--- a/cmd/internal/find/primary/meta.go
+++ b/cmd/internal/find/primary/meta.go
@@ -2,5 +2,41 @@ package primary
 
 import "github.com/puppetlabs/wash/cmd/internal/find/primary/meta"
 
+/*
+metaPrimary        => (-meta|-m) ObjectPredicate
+ObjectPredicate    => EmptyPredicate | ‘.’ Key Predicate
+EmptyPredicate     => -empty
+Key                => [ ^.[ ] ]+ (i.e. one or more cs that aren't ".", "[", or "]")
+
+Predicate          => ObjectPredicate     |
+                      ArrayPredicate      |
+                      PrimitivePredicate
+
+ArrayPredicate     => EmptyPredicate      |
+                      ‘[' ? ‘]’ Predicate |
+                      ‘[' * ‘]’ Predicate |
+                      ‘[' N ‘]’ Predicate |
+
+PrimitivePredicate => NullPredicate       |
+                      ExistsPredicate     |
+                      BooleanPredicate    |
+                      NumericPredicate    |
+                      TimePredicate       |
+                      StringPredicate
+
+NullPredicate      => -null
+ExistsPredicate    => -exists
+BooleanPredicate   => -true | -false
+
+NumericPredicate   => (+|-)? Number
+Number             => N | '{' N '}' | numeric.SizeRegex
+
+TimePredicate      => (+|-)? Duration
+Duration           => numeric.DurationRegex | '{' numeric.DurationRegex '}'
+
+StringPredicate    => [^-].*
+
+N                  => \d+ (i.e. some number > 0)
+*/
 //nolint
-var metaPrimary = meta.Primary
+var metaPrimary = Parser.newPrimary([]string{"-meta", "-m"}, meta.Parse)
diff --git a/cmd/internal/find/primary/meta/arrayPredicate.go b/cmd/internal/find/primary/meta/arrayPredicate.go
index df271c5..e406fe7 100644
--- a/cmd/internal/find/primary/meta/arrayPredicate.go
+++ b/cmd/internal/find/primary/meta/arrayPredicate.go
@@ -5,14 +5,15 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 )
 
 // ArrayPredicate => EmptyPredicate      |
 //                   ‘[' ? ‘]’ Predicate |
 //                   ‘[' * ‘]’ Predicate |
 //                   ‘[' N ‘]’ Predicate |
-func parseArrayPredicate(tokens []string) (predicate, []string, error) {
+func parseArrayPredicate(tokens []string) (predicate.Generic, []string, error) {
 	if p, tokens, err := parseEmptyPredicate(tokens); err == nil {
 		return p, tokens, err
 	}
@@ -101,7 +102,7 @@ func parseArrayPredicateType(token string) (arrayPredicateType, string, error) {
 	return ptype, token[endIx+1:], nil
 }
 
-func arrayP(ptype arrayPredicateType, p predicate) predicate {
+func arrayP(ptype arrayPredicateType, p predicate.Generic) predicate.Generic {
 	switch ptype.t {
 	case 's':
 		return toArrayP(func(vs []interface{}) bool {
@@ -140,7 +141,7 @@ func arrayP(ptype arrayPredicateType, p predicate) predicate {
 
 // toArrayP is a helper for arrayP that's meant to reduce
 // the boilerplate type validation.
-func toArrayP(p func([]interface{}) bool) predicate {
+func toArrayP(p func([]interface{}) bool) predicate.Generic {
 	return func(v interface{}) bool {
 		arrayV, ok := v.([]interface{})
 		if !ok {
diff --git a/cmd/internal/find/primary/meta/emptyPredicate.go b/cmd/internal/find/primary/meta/emptyPredicate.go
index 24a4f86..8e215f0 100644
--- a/cmd/internal/find/primary/meta/emptyPredicate.go
+++ b/cmd/internal/find/primary/meta/emptyPredicate.go
@@ -1,9 +1,12 @@
 package meta
 
-import "github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+import (
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+)
 
 // emptyPredicate => -empty
-func parseEmptyPredicate(tokens []string) (predicate, []string, error) {
+func parseEmptyPredicate(tokens []string) (predicate.Generic, []string, error) {
 	if len(tokens) == 0 || tokens[0] != "-empty" {
 		return nil, nil, errz.NewMatchError("expected '-empty'")
 	}
diff --git a/cmd/internal/find/primary/meta/numericPredicate.go b/cmd/internal/find/primary/meta/numericPredicate.go
index 11f30f9..908d67c 100644
--- a/cmd/internal/find/primary/meta/numericPredicate.go
+++ b/cmd/internal/find/primary/meta/numericPredicate.go
@@ -3,13 +3,14 @@ package meta
 import (
 	"fmt"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/primary/numeric"
 )
 
 // NumericPredicate => (+|-)? Number
 // Number           => N | '{' N '}' | numeric.SizeRegex
-func parseNumericPredicate(tokens []string) (predicate, []string, error) {
+func parseNumericPredicate(tokens []string) (predicate.Generic, []string, error) {
 	if len(tokens) == 0 {
 		return nil, nil, errz.NewMatchError("expected a +, -, or a digit")
 	}
diff --git a/cmd/internal/find/primary/meta/objectPredicate.go b/cmd/internal/find/primary/meta/objectPredicate.go
index 68086a2..2ed7f18 100644
--- a/cmd/internal/find/primary/meta/objectPredicate.go
+++ b/cmd/internal/find/primary/meta/objectPredicate.go
@@ -5,7 +5,8 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 )
 
 // A key consists of one or more characters that aren't
@@ -18,7 +19,7 @@ var keyRegex = regexp.MustCompile(`^([^\.\[\]]+)`)
 
 // ObjectPredicate => EmptyPredicate | ‘.’ Key Predicate
 // Key             => keyRegex
-func parseObjectPredicate(tokens []string) (predicate, []string, error) {
+func parseObjectPredicate(tokens []string) (predicate.Generic, []string, error) {
 	if p, tokens, err := parseEmptyPredicate(tokens); err == nil {
 		return p, tokens, err
 	}
@@ -57,7 +58,7 @@ func parseObjectPredicate(tokens []string) (predicate, []string, error) {
 	return objectP(key, p), tokens, nil
 }
 
-func objectP(key string, p predicate) predicate {
+func objectP(key string, p predicate.Generic) predicate.Generic {
 	return func(v interface{}) bool {
 		mp, ok := v.(map[string]interface{})
 		if !ok {
diff --git a/cmd/internal/find/primary/meta/parse.go b/cmd/internal/find/primary/meta/parse.go
new file mode 100644
index 0000000..d7c3538
--- /dev/null
+++ b/cmd/internal/find/primary/meta/parse.go
@@ -0,0 +1,19 @@
+// Package meta contains all the parsing logic for the `meta` primary
+package meta
+
+import (
+	"github.com/puppetlabs/wash/cmd/internal/find/types"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+)
+
+// Parse is the meta primary's parse function.
+func Parse(tokens []string) (predicate.Entry, []string, error) {
+	p, tokens, err := parseObjectPredicate(tokens)
+	if err != nil {
+		return nil, nil, err
+	}
+	return func(e types.Entry) bool {
+		mp := map[string]interface{}(e.Attributes.Meta())
+		return p(mp)
+	}, tokens, nil
+}
diff --git a/cmd/internal/find/primary/meta/predicate.go b/cmd/internal/find/primary/meta/predicate.go
index 6aec386..81c09c0 100644
--- a/cmd/internal/find/primary/meta/predicate.go
+++ b/cmd/internal/find/primary/meta/predicate.go
@@ -1,29 +1,26 @@
 package meta
 
 import (
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 )
 
-// predicate represents a predicate in the meta primary's grammar.
-type predicate func(v interface{}) bool
-
 /*
 Predicate => ObjectPredicate |
              ArrayPredicate  |
              PrimitivePredicate
 */
-func parsePredicate(tokens []string) (predicate, []string, error) {
-	if len(tokens) == 0 {
-		return nil, nil, errz.NewMatchError("expected either a primitive, object, or array predicate")
+func parsePredicate(tokens []string) (predicate.Generic, []string, error) {
+	cp := &predicate.CompositeParser{
+		ErrMsg: "expected either a primitive, object, or array predicate",
+		Parsers: []predicate.Parser{
+			predicate.GenericParser(parseObjectPredicate),
+			predicate.GenericParser(parseArrayPredicate),
+			predicate.GenericParser(parsePrimitivePredicate),
+		},
 	}
-	switch token := tokens[0]; token[0] {
-	case '.':
-		return parseObjectPredicate(tokens)
-	case '[':
-		fallthrough
-	case ']':
-		return parseArrayPredicate(tokens)
-	default:
-		return parsePrimitivePredicate(tokens)
+	p, tokens, err := cp.Parse(tokens)
+	if err != nil {
+		return nil, nil, err
 	}
+	return p.(predicate.Generic), tokens, err
 }
diff --git a/cmd/internal/find/primary/meta/primary.go b/cmd/internal/find/primary/meta/primary.go
deleted file mode 100644
index c07acfd..0000000
--- a/cmd/internal/find/primary/meta/primary.go
+++ /dev/null
@@ -1,60 +0,0 @@
-// Package meta contains all the parsing logic for the `meta` primary
-package meta
-
-import (
-	"fmt"
-
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
-	"github.com/puppetlabs/wash/cmd/internal/find/types"
-)
-
-/*
-Primary            => (-meta|-m) ObjectPredicate
-
-ObjectPredicate    => EmptyPredicate | ‘.’ Key Predicate
-EmptyPredicate     => -empty
-Key                => [ ^.[ ] ]+ (i.e. one or more cs that aren't ".", "[", or "]")
-
-Predicate          => ObjectPredicate     |
-                      ArrayPredicate      |
-                      PrimitivePredicate
-
-ArrayPredicate     => EmptyPredicate      |
-                      ‘[' ? ‘]’ Predicate |
-                      ‘[' * ‘]’ Predicate |
-                      ‘[' N ‘]’ Predicate |
-
-PrimitivePredicate => NullPredicate       |
-                      ExistsPredicate     |
-                      BooleanPredicate    |
-                      NumericPredicate    |
-                      TimePredicate       |
-                      StringPredicate
-
-NullPredicate      => -null
-ExistsPredicate    => -exists
-BooleanPredicate   => -true | -false
-
-NumericPredicate   => (+|-)? Number
-Number             => N | '{' N '}' | numeric.SizeRegex
-
-TimePredicate      => (+|-)? Duration
-Duration           => numeric.DurationRegex | '{' numeric.DurationRegex '}'
-
-StringPredicate    => [^-].*
-
-N                  => \d+ (i.e. some number > 0)
-*/
-//nolint
-var Primary = grammar.NewAtom([]string{"-meta", "-m"}, func(tokens []string) (types.Predicate, []string, error) {
-	// tokens[0] contains the (-meta|-m) part
-	token := tokens[0]
-	p, tokens, err := parseObjectPredicate(tokens[1:])
-	if err != nil {
-		return nil, nil, fmt.Errorf("%v: %v", token, err)
-	}
-	return func(e types.Entry) bool {
-		mp := map[string]interface{}(e.Attributes.Meta())
-		return p(mp)
-	}, tokens, nil
-})
diff --git a/cmd/internal/find/primary/meta/primitivePredicate.go b/cmd/internal/find/primary/meta/primitivePredicate.go
index 354c499..01c505a 100644
--- a/cmd/internal/find/primary/meta/primitivePredicate.go
+++ b/cmd/internal/find/primary/meta/primitivePredicate.go
@@ -1,7 +1,8 @@
 package meta
 
 import (
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 )
 
 /*
@@ -12,7 +13,7 @@ PrimitivePredicate => NullPredicate       |
                       TimePredicate       |
                       StringPredicate
 */
-func parsePrimitivePredicate(tokens []string) (predicate, []string, error) {
+func parsePrimitivePredicate(tokens []string) (predicate.Generic, []string, error) {
 	if len(tokens) == 0 {
 		return nil, nil, errz.NewMatchError("expected a primitive predicate")
 	}
@@ -79,7 +80,7 @@ func existsP(v interface{}) bool {
 var trueP = makeBooleanPredicate(true)
 var falseP = makeBooleanPredicate(false)
 
-func makeBooleanPredicate(expectedVal bool) predicate {
+func makeBooleanPredicate(expectedVal bool) predicate.Generic {
 	return func(v interface{}) bool {
 		bv, ok := v.(bool)
 		if !ok {
diff --git a/cmd/internal/find/primary/meta/stringPredicate.go b/cmd/internal/find/primary/meta/stringPredicate.go
index 116eaaf..91f896f 100644
--- a/cmd/internal/find/primary/meta/stringPredicate.go
+++ b/cmd/internal/find/primary/meta/stringPredicate.go
@@ -3,11 +3,12 @@ package meta
 import (
 	"fmt"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 )
 
 // StringPredicate => [^-].*
-func parseStringPredicate(tokens []string) (predicate, []string, error) {
+func parseStringPredicate(tokens []string) (predicate.Generic, []string, error) {
 	if len(tokens) == 0 || len(tokens[0]) == 0 {
 		return nil, nil, errz.NewMatchError("expected a nonempty string")
 	}
diff --git a/cmd/internal/find/primary/meta/timePredicate.go b/cmd/internal/find/primary/meta/timePredicate.go
index f8a237a..f52c2e8 100644
--- a/cmd/internal/find/primary/meta/timePredicate.go
+++ b/cmd/internal/find/primary/meta/timePredicate.go
@@ -4,14 +4,15 @@ import (
 	"fmt"
 
 	"github.com/puppetlabs/wash/cmd/internal/find/params"
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/primary/numeric"
 	"github.com/puppetlabs/wash/munge"
 )
 
 // TimePredicate => (+|-)? Duration
 // Duration      => numeric.DurationRegex | '{' numeric.DurationRegex '}'
-func parseTimePredicate(tokens []string) (predicate, []string, error) {
+func parseTimePredicate(tokens []string) (predicate.Generic, []string, error) {
 	if params.StartTime.IsZero() {
 		panic("meta.parseTimePredicate called without setting params.StartTime")
 	}
diff --git a/cmd/internal/find/primary/name.go b/cmd/internal/find/primary/name.go
index 719ad7b..7f6494e 100644
--- a/cmd/internal/find/primary/name.go
+++ b/cmd/internal/find/primary/name.go
@@ -4,16 +4,15 @@ import (
 	"fmt"
 
 	"github.com/gobwas/glob"
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
 
 // namePrimary => -name ShellGlob
 //nolint
-var namePrimary = grammar.NewAtom([]string{"-name"}, func(tokens []string) (types.Predicate, []string, error) {
-	tokens = tokens[1:]
+var namePrimary = Parser.newPrimary([]string{"-name"}, func(tokens []string) (predicate.Entry, []string, error) {
 	if len(tokens) == 0 {
-		return nil, nil, fmt.Errorf("-name: requires additional arguments")
+		return nil, nil, fmt.Errorf("requires additional arguments")
 	}
 
 	g, err := glob.Compile(tokens[0])
diff --git a/cmd/internal/find/primary/numeric/bracket.go b/cmd/internal/find/primary/numeric/bracket.go
index ebb293b..bf349e6 100644
--- a/cmd/internal/find/primary/numeric/bracket.go
+++ b/cmd/internal/find/primary/numeric/bracket.go
@@ -3,7 +3,7 @@ package numeric
 import (
 	"fmt"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 )
 
 // Bracket returns a new parser g that parses all numbers
diff --git a/cmd/internal/find/primary/numeric/duration.go b/cmd/internal/find/primary/numeric/duration.go
index d508090..ec7abe6 100644
--- a/cmd/internal/find/primary/numeric/duration.go
+++ b/cmd/internal/find/primary/numeric/duration.go
@@ -6,7 +6,7 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 )
 
 var durationsMap = map[byte]time.Duration{
diff --git a/cmd/internal/find/primary/numeric/predicate.go b/cmd/internal/find/primary/numeric/predicate.go
index 6c74721..31094c9 100644
--- a/cmd/internal/find/primary/numeric/predicate.go
+++ b/cmd/internal/find/primary/numeric/predicate.go
@@ -3,21 +3,19 @@ package numeric
 import (
 	"fmt"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 )
 
-// Predicate is a predicate on a numeric value.
-type Predicate func(int64) bool
-
 // Parser parses numeric values.
 type Parser func(string) (int64, error)
 
-// ParsePredicate parses a Predicate from str. Str should
+// ParsePredicate parses a numeric predicate from str. Str should
 // satisfy the regex `(\+|\-)?<number>`, where <number> is s.t.
 // that parser(<number>) does not return an error for at least one
 // parser in parsers. The returned value is the parsed predicate
 // and the id of the parser that parsed <number>.
-func ParsePredicate(str string, parsers ...Parser) (Predicate, int, error) {
+func ParsePredicate(str string, parsers ...Parser) (predicate.Int, int, error) {
 	if len(str) == 0 {
 		return nil, -1, errz.NewMatchError("empty input")
 	}
@@ -62,11 +60,4 @@ func ParsePredicate(str string, parsers ...Parser) (Predicate, int, error) {
 			return v == n
 		}
 	}, parserID, nil
-}
-
-// Not returns a predicate f s.t. f(v) = !p(v)
-func Not(p Predicate) Predicate {
-	return func(v int64) bool {
-		return !p(v)
-	}
-}
+}
\ No newline at end of file
diff --git a/cmd/internal/find/primary/numeric/sign.go b/cmd/internal/find/primary/numeric/sign.go
index fa5d855..477e34a 100644
--- a/cmd/internal/find/primary/numeric/sign.go
+++ b/cmd/internal/find/primary/numeric/sign.go
@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 )
 
 // ParsePositiveInt parses a positive integer.
diff --git a/cmd/internal/find/primary/numeric/size.go b/cmd/internal/find/primary/numeric/size.go
index b71b632..8e9e073 100644
--- a/cmd/internal/find/primary/numeric/size.go
+++ b/cmd/internal/find/primary/numeric/size.go
@@ -5,7 +5,7 @@ import (
 	"regexp"
 	"strconv"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 )
 
 var bytesMap = map[byte]int64{
diff --git a/cmd/internal/find/primary/parser.go b/cmd/internal/find/primary/parser.go
new file mode 100644
index 0000000..9467faa
--- /dev/null
+++ b/cmd/internal/find/primary/parser.go
@@ -0,0 +1,71 @@
+package primary
+
+import (
+	"fmt"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"strings"
+)
+
+// Parser parses `wash find` primaries.
+var Parser = &parser{
+	CompositeParser: &predicate.CompositeParser{
+		ErrMsg: "unknown primary",
+	},
+	primaries: make(map[string]*primary),
+}
+
+type parser struct {
+	*predicate.CompositeParser
+	primaries map[string]*primary
+}
+
+// IsPrimary returns true if the token is a `wash find`
+// primary
+func (parser *parser) IsPrimary(token string) bool {
+	_, ok := parser.primaries[token]
+	return ok
+}
+
+func (parser *parser) newPrimary(tokens []string, parse func(tokens []string) (predicate.Entry, []string, error)) *primary {
+	p := &primary{
+		tokens: tokens,
+		parseFunc: parse,
+	}
+	p.tokensMap = make(map[string]struct{})
+	for _, token := range tokens {
+		p.tokensMap[token] = struct{}{}
+		parser.primaries[token] = p
+		parser.Parsers = append(parser.Parsers, p)
+	}
+	return p
+}
+
+// primary represents a `wash find` primary.
+type primary struct {
+	tokens []string
+	tokensMap map[string]struct{}
+	parseFunc predicate.EntryParser
+}
+
+func (primary *primary) parse(tokens []string) (predicate.Entry, []string, error) {
+	tokensErrMsg := fmt.Sprintf("expected one of: %v", strings.Join(primary.tokens, ","))
+	if len(tokens) == 0 {
+		return nil, nil, errz.NewMatchError(tokensErrMsg)
+	}
+	token := tokens[0]
+	if _, ok := primary.tokensMap[token]; !ok {
+		return nil, nil, errz.NewMatchError(tokensErrMsg)
+	}
+	tokens = tokens[1:]
+	p, tokens, err := primary.parseFunc(tokens)
+	if err != nil {
+		return nil, nil, fmt.Errorf("%v: %v", token, err)
+	}
+	return p, tokens, nil
+}
+
+// Parse parses a predicate from the given primary.
+func (primary *primary) Parse(tokens []string) (predicate.Predicate, []string, error) {
+	return primary.parse(tokens)
+}
\ No newline at end of file
diff --git a/cmd/internal/find/primary/path.go b/cmd/internal/find/primary/path.go
index 9e0f3d7..bf01475 100644
--- a/cmd/internal/find/primary/path.go
+++ b/cmd/internal/find/primary/path.go
@@ -4,21 +4,20 @@ import (
 	"fmt"
 
 	"github.com/gobwas/glob"
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
 
 // pathPrimary => -path ShellGlob
 //nolint
-var pathPrimary = grammar.NewAtom([]string{"-path"}, func(tokens []string) (types.Predicate, []string, error) {
-	tokens = tokens[1:]
+var pathPrimary = Parser.newPrimary([]string{"-path"}, func(tokens []string) (predicate.Entry, []string, error) {
 	if len(tokens) == 0 {
-		return nil, nil, fmt.Errorf("-path: requires additional arguments")
+		return nil, nil, fmt.Errorf("requires additional arguments")
 	}
 
 	g, err := glob.Compile(tokens[0])
 	if err != nil {
-		return nil, nil, fmt.Errorf("-path: invalid glob: %v", err)
+		return nil, nil, fmt.Errorf("invalid glob: %v", err)
 	}
 
 	return func(e types.Entry) bool {
diff --git a/cmd/internal/find/primary/size.go b/cmd/internal/find/primary/size.go
index d53ed09..e305137 100644
--- a/cmd/internal/find/primary/size.go
+++ b/cmd/internal/find/primary/size.go
@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"math"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/primary/numeric"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
 )
@@ -18,10 +18,9 @@ import (
 //   -size +1k (true if the entry's size is greater than 1 kibibyte (1024 bytes))
 //
 //nolint
-var sizePrimary = grammar.NewAtom([]string{"-size"}, func(tokens []string) (types.Predicate, []string, error) {
-	tokens = tokens[1:]
+var sizePrimary = Parser.newPrimary([]string{"-size"}, func(tokens []string) (predicate.Entry, []string, error) {
 	if len(tokens) == 0 {
-		return nil, nil, fmt.Errorf("-size: requires additional arguments")
+		return nil, nil, fmt.Errorf("requires additional arguments")
 	}
 	numericP, parserID, err := numeric.ParsePredicate(
 		tokens[0],
@@ -29,7 +28,7 @@ var sizePrimary = grammar.NewAtom([]string{"-size"}, func(tokens []string) (type
 		numeric.ParseSize,
 	)
 	if err != nil {
-		return nil, nil, fmt.Errorf("-size: %v: illegal size value", tokens[0])
+		return nil, nil, fmt.Errorf("%v: illegal size value", tokens[0])
 	}
 
 	p := func(e types.Entry) bool {
diff --git a/cmd/internal/find/primary/timeAttr.go b/cmd/internal/find/primary/timeAttr.go
index 53b76df..1ab1a3d 100644
--- a/cmd/internal/find/primary/timeAttr.go
+++ b/cmd/internal/find/primary/timeAttr.go
@@ -5,7 +5,7 @@ import (
 	"math"
 	"time"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/grammar"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/puppetlabs/wash/cmd/internal/find/params"
 	"github.com/puppetlabs/wash/cmd/internal/find/primary/numeric"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
@@ -37,15 +37,14 @@ func getTimeAttrValue(name string, e types.Entry) (time.Time, bool) {
 //
 // NOTE: For non-unit time values (e.g. 1, +1, -1), the difference is rounded to the next 24-hour period. For
 // example, a difference of 1.5 days will be rounded to 2 days.
-func newTimeAttrPrimary(name string) *grammar.Atom {
+func newTimeAttrPrimary(name string) *primary {
 	tk := "-" + name
-	return grammar.NewAtom([]string{tk}, func(tokens []string) (types.Predicate, []string, error) {
+	return Parser.newPrimary([]string{tk}, func(tokens []string) (predicate.Entry, []string, error) {
 		if params.StartTime.IsZero() {
 			panic("Attempting to parse a time primary without setting params.StartTime")
 		}
-		tokens = tokens[1:]
 		if len(tokens) == 0 {
-			return nil, nil, fmt.Errorf("%v: requires additional arguments", tk)
+			return nil, nil, fmt.Errorf("requires additional arguments")
 		}
 		numericP, parserID, err := numeric.ParsePredicate(
 			tokens[0],
@@ -53,7 +52,7 @@ func newTimeAttrPrimary(name string) *grammar.Atom {
 			numeric.ParseDuration,
 		)
 		if err != nil {
-			return nil, nil, fmt.Errorf("-%v: %v: illegal time value", name, tokens[0])
+			return nil, nil, fmt.Errorf("%v: illegal time value", tokens[0])
 		}
 
 		p := func(e types.Entry) bool {
diff --git a/cmd/internal/find/types/predicate.go b/cmd/internal/find/types/predicate.go
deleted file mode 100644
index 167d095..0000000
--- a/cmd/internal/find/types/predicate.go
+++ /dev/null
@@ -1,4 +0,0 @@
-package types
-
-// Predicate represents a predicate used by `wash find`
-type Predicate func(e Entry) bool
diff --git a/cmd/internal/find/walker.go b/cmd/internal/find/walker.go
index 64ed46a..5341f84 100644
--- a/cmd/internal/find/walker.go
+++ b/cmd/internal/find/walker.go
@@ -6,12 +6,13 @@ import (
 	"github.com/puppetlabs/wash/api/client"
 	"github.com/puppetlabs/wash/cmd/internal/find/parser"
 	"github.com/puppetlabs/wash/cmd/internal/find/types"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	cmdutil "github.com/puppetlabs/wash/cmd/util"
 	"github.com/puppetlabs/wash/plugin"
 )
 
 type walker struct {
-	p    types.Predicate
+	p    predicate.Entry
 	opts types.Options
 	conn client.Client
 }
diff --git a/plugin/aws/s3Object.go b/plugin/aws/s3Object.go
index b960f14..8568c0d 100644
--- a/plugin/aws/s3Object.go
+++ b/plugin/aws/s3Object.go
@@ -95,7 +95,7 @@ func (o *s3Object) Open(ctx context.Context) (plugin.SizedReader, error) {
 	return &s3ObjectReader{o: o}, nil
 }
 
-func (o *s3Object) Stream(context.Context) (io.ReadCloser, error) {
+func (o *s3Object) Stream(context.Context) (io.Reader, error) {
 	return o.fetchContent(0)
 }
 
diff --git a/plugin/cleanupReader.go b/plugin/cleanupReader.go
deleted file mode 100644
index b38a78f..0000000
--- a/plugin/cleanupReader.go
+++ /dev/null
@@ -1,16 +0,0 @@
-package plugin
-
-import "io"
-
-// CleanupReader is a wrapper for an io.ReadCloser that performs cleanup when closed.
-type CleanupReader struct {
-	io.ReadCloser
-	Cleanup func()
-}
-
-// Close closes the reader it wraps, then calls the Cleanup function and returns any errors.
-func (c CleanupReader) Close() error {
-	err := c.ReadCloser.Close()
-	c.Cleanup()
-	return err
-}
diff --git a/plugin/docker/container-log.go b/plugin/docker/container-log.go
index ae436c9..6bc9ea4 100644
--- a/plugin/docker/container-log.go
+++ b/plugin/docker/container-log.go
@@ -52,7 +52,7 @@ func (clf *containerLogFile) Open(ctx context.Context) (plugin.SizedReader, erro
 	return bytes.NewReader(buf.Bytes()), nil
 }
 
-func (clf *containerLogFile) Stream(ctx context.Context) (io.ReadCloser, error) {
+func (clf *containerLogFile) Stream(ctx context.Context) (io.Reader, error) {
 	opts := types.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true, Tail: "10"}
 	rdr, err := clf.client.ContainerLogs(ctx, clf.containerName, opts)
 	if err != nil {
diff --git a/plugin/docker/container.go b/plugin/docker/container.go
index 6c5336b..0a1793a 100644
--- a/plugin/docker/container.go
+++ b/plugin/docker/container.go
@@ -60,10 +60,8 @@ func (c *container) Exec(ctx context.Context, cmd string, args []string, opts pl
 	execResult := plugin.ExecResult{}
 
 	command := append([]string{cmd}, args...)
-	activity.Record(ctx, "Exec %v on %v", command, c.Name())
-
-	cfg := types.ExecConfig{Cmd: command, AttachStdout: true, AttachStderr: true, Tty: opts.Tty}
-	if opts.Stdin != nil || opts.Tty {
+	cfg := types.ExecConfig{Cmd: command, AttachStdout: true, AttachStderr: true}
+	if opts.Stdin != nil {
 		cfg.AttachStdin = true
 	}
 	created, err := c.client.ContainerExecCreate(ctx, c.id, cfg)
@@ -78,20 +76,7 @@ func (c *container) Exec(ctx context.Context, cmd string, args []string, opts pl
 
 	outputCh, stdout, stderr := plugin.CreateExecOutputStreams(ctx)
 	go func() {
-		go func() {
-			// Close the response on context cancellation. Copying will block until there's more to read
-			// from the exec output. For an action with no more output it may never return.
-			// Asynchronously watch for context cancellation, and when it happens close the response so
-			// the parent goroutine can complete. We expect all contexts to complete eventually.
-			<-ctx.Done()
-
-			if opts.Tty {
-				// If resp.Conn is still open, Send Ctrl-C over resp.Conn before closing it.
-				_, err := resp.Conn.Write([]byte{0x03})
-				activity.Record(ctx, "Sent ETX on context termination: %v", err)
-			}
-			resp.Close()
-		}()
+		defer resp.Close()
 
 		_, err := stdcopy.StdCopy(stdout, stderr, resp.Reader)
 		activity.Record(ctx, "Exec on %v complete: %v", c.Name(), err)
@@ -103,13 +88,8 @@ func (c *container) Exec(ctx context.Context, cmd string, args []string, opts pl
 	if opts.Stdin != nil {
 		go func() {
 			_, writeErr = io.Copy(resp.Conn, opts.Stdin)
-			// If using a Tty, wait until done reading in case we need to send Ctrl-C; when a Tty is
-			// allocated commands expect user input and will respond to control signals. Otherwise close
-			// input now to ensure commands that depend on EOF execute correctly.
-			if !opts.Tty {
-				respErr := resp.CloseWrite()
-				activity.Record(ctx, "Closed execution input stream for %v: %v, %v", c.Name(), writeErr, respErr)
-			}
+			respErr := resp.CloseWrite()
+			activity.Record(ctx, "Closed execution input stream for %v: %v, %v", c.Name(), writeErr, respErr)
 		}()
 	}
 
diff --git a/plugin/docker/volume.go b/plugin/docker/volume.go
index 887a6a7..d8319e6 100644
--- a/plugin/docker/volume.go
+++ b/plugin/docker/volume.go
@@ -5,7 +5,6 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"io"
 	"io/ioutil"
 	"time"
 
@@ -174,57 +173,3 @@ func (v *volume) VolumeOpen(ctx context.Context, path string) (plugin.SizedReade
 	}
 	return bytes.NewReader(bits), nil
 }
-
-func (v *volume) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
-	// Create a container that mounts a volume and tails a file. Run it and capture the output.
-	cid, err := v.createContainer(ctx, []string{"tail", "-f", mountpoint + path})
-	if err != nil {
-		return nil, err
-	}
-
-	// Manually use this in case of errors. On success, the returned Closer will need to call instead.
-	delete := func() {
-		err := v.client.ContainerRemove(ctx, cid, types.ContainerRemoveOptions{})
-		activity.Record(ctx, "Deleted container %v: %v", cid, err)
-	}
-
-	activity.Record(ctx, "Starting container %v", cid)
-	if err := v.client.ContainerStart(ctx, cid, types.ContainerStartOptions{}); err != nil {
-		delete()
-		return nil, err
-	}
-
-	// Manually use this in case of errors. On success, the returned Closer will need to call instead.
-	killAndDelete := func() {
-		err := v.client.ContainerKill(ctx, cid, "SIGKILL")
-		activity.Record(ctx, "Stopped temporary container %v: %v", cid, err)
-		delete()
-	}
-
-	activity.Record(ctx, "Waiting for container %v", cid)
-	waitC, errC := v.client.ContainerWait(ctx, cid, docontainer.WaitConditionNotRunning)
-	var statusCode int64
-	select {
-	case err := <-errC:
-		killAndDelete()
-		return nil, err
-	case result := <-waitC:
-		statusCode = result.StatusCode
-		activity.Record(ctx, "Container %v finished[%v]: %v", cid, result.StatusCode, result.Error)
-	}
-
-	opts := types.ContainerLogsOptions{ShowStdout: true}
-	if statusCode != 0 {
-		opts.ShowStderr = true
-	}
-
-	activity.Record(ctx, "Gathering log for %v", cid)
-	output, err := v.client.ContainerLogs(ctx, cid, opts)
-	if err != nil {
-		killAndDelete()
-		return nil, err
-	}
-
-	// Wrap the log output in a ReadCloser that stops and kills the container on Close.
-	return plugin.CleanupReader{ReadCloser: output, Cleanup: killAndDelete}, nil
-}
diff --git a/plugin/externalPluginEntry.go b/plugin/externalPluginEntry.go
index 897d104..54fa6c3 100644
--- a/plugin/externalPluginEntry.go
+++ b/plugin/externalPluginEntry.go
@@ -179,7 +179,7 @@ func (s *stdoutStreamer) Close() error {
 }
 
 // Stream streams the entry's content
-func (e *externalPluginEntry) Stream(ctx context.Context) (io.ReadCloser, error) {
+func (e *externalPluginEntry) Stream(ctx context.Context) (io.Reader, error) {
 	cmd, stdoutR, stderrR, err := CreateCommand(
 		e.script.Path(),
 		"stream",
diff --git a/plugin/kubernetes/pod.go b/plugin/kubernetes/pod.go
index 9cbb54f..2d28c0f 100644
--- a/plugin/kubernetes/pod.go
+++ b/plugin/kubernetes/pod.go
@@ -113,7 +113,7 @@ func (p *pod) Open(ctx context.Context) (plugin.SizedReader, error) {
 	return bytes.NewReader(pdInfo.logContent), nil
 }
 
-func (p *pod) Stream(ctx context.Context) (io.ReadCloser, error) {
+func (p *pod) Stream(ctx context.Context) (io.Reader, error) {
 	var tailLines int64 = 10
 	req := p.client.CoreV1().Pods(p.ns).GetLogs(p.Name(), &corev1.PodLogOptions{Follow: true, TailLines: &tailLines})
 	return req.Stream()
@@ -147,33 +147,7 @@ func (p *pod) Exec(ctx context.Context, cmd string, args []string, opts plugin.E
 	outputCh, stdout, stderr := plugin.CreateExecOutputStreams(ctx)
 	exitcode := 0
 	go func() {
-		stdin := opts.Stdin
-		if opts.Tty {
-			// If using a Tty, create an input stream that allows us to send Ctrl-C to end execution;
-			// when a Tty is allocated commands expect user input and will respond to control signals.
-			// Otherwise close input now to ensure commands that depend on EOF execute correctly.
-			r, w := io.Pipe()
-			if stdin != nil {
-				stdin = io.MultiReader(stdin, r)
-			} else {
-				stdin = r
-			}
-
-			go func() {
-				// Close the response on context cancellation. Copying will block until there's more to
-				// read from the exec output. For an action with no more output it may never return.
-				// Asynchronously watch for context cancellation, and when it happens close the response
-				// so the parent goroutine can complete. We expect all contexts to complete eventually.
-				<-ctx.Done()
-
-				// Append Ctrl-C to input to signal end of execution.
-				_, err := w.Write([]byte{0x03})
-				activity.Record(ctx, "Sent ETX on context termination: %v", err)
-				w.Close()
-			}()
-		}
-
-		streamOpts := remotecommand.StreamOptions{Stdout: stdout, Stderr: stderr, Stdin: stdin, Tty: opts.Tty}
+		streamOpts := remotecommand.StreamOptions{Stdout: stdout, Stderr: stderr, Stdin: opts.Stdin}
 		err = executor.Stream(streamOpts)
 		activity.Record(ctx, "Exec on %v complete: %v", p.Name(), err)
 		if exerr, ok := err.(k8exec.ExitError); ok {
diff --git a/plugin/kubernetes/pvc.go b/plugin/kubernetes/pvc.go
index 147650c..d2b18d8 100644
--- a/plugin/kubernetes/pvc.go
+++ b/plugin/kubernetes/pvc.go
@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"time"
 
@@ -172,7 +171,7 @@ func (v *pvc) VolumeList(ctx context.Context) (volume.DirMap, error) {
 }
 
 func (v *pvc) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, error) {
-	// Create a container that mounts a pvc and output the file.
+	// Create a container that mounts a pvc and waits. Use it to download a file.
 	pid, err := v.createPod([]string{"cat", mountpoint + path})
 	activity.Record(ctx, "Reading from: %v", mountpoint+path)
 	if err != nil {
@@ -209,47 +208,3 @@ func (v *pvc) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader,
 	}
 	return bytes.NewReader(bits), nil
 }
-
-func (v *pvc) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
-	// Create a container that mounts a pvc and tail the file.
-	pid, err := v.createPod([]string{"tail", "-f", mountpoint + path})
-	activity.Record(ctx, "Streaming from: %v", mountpoint+path)
-	if err != nil {
-		return nil, err
-	}
-
-	// Manually use this in case of errors. On success, the returned Closer will need to call instead.
-	delete := func() {
-		activity.Record(ctx, "Deleted temporary pod %v: %v", pid, v.podi.Delete(pid, &metav1.DeleteOptions{}))
-	}
-
-	activity.Record(ctx, "Waiting for pod %v", pid)
-	// Start watching for new events related to the pod we created.
-	if err = v.waitForPod(ctx, pid); err != nil && err != errPodTerminated {
-		delete()
-		return nil, err
-	}
-	podErr := err
-
-	activity.Record(ctx, "Gathering log for %v", pid)
-	output, err := v.podi.GetLogs(pid, &corev1.PodLogOptions{}).Stream()
-	if err != nil {
-		delete()
-		return nil, err
-	}
-
-	if podErr == errPodTerminated {
-		bits, err := ioutil.ReadAll(output)
-		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
-		delete()
-		if err != nil {
-			return nil, err
-		}
-		activity.Record(ctx, "Read: %v", bits)
-
-		return nil, errors.New(string(bits))
-	}
-
-	// Wrap the log output in a ReadCloser that stops and kills the container on Close.
-	return plugin.CleanupReader{ReadCloser: output, Cleanup: delete}, nil
-}
diff --git a/plugin/outputStream.go b/plugin/outputStream.go
index b3eb847..3963c5b 100644
--- a/plugin/outputStream.go
+++ b/plugin/outputStream.go
@@ -6,11 +6,18 @@ import (
 	"time"
 )
 
+const (
+	// StdoutID represents Stdout
+	StdoutID = iota
+	// StderrID represents Stderr
+	StderrID
+)
+
 // OutputStream represents stdout/stderr.
 type OutputStream struct {
 	ctx        context.Context
 	sentCtxErr bool
-	id         ExecPacketType
+	id         int8
 	ch         chan ExecOutputChunk
 	closer     *multiCloser
 }
@@ -84,8 +91,8 @@ func CreateExecOutputStreams(ctx context.Context) (<-chan ExecOutputChunk, *Outp
 	outputCh := make(chan ExecOutputChunk)
 	closer := &multiCloser{ch: outputCh, countdown: 2}
 
-	stdout := &OutputStream{ctx: ctx, id: Stdout, ch: outputCh, closer: closer}
-	stderr := &OutputStream{ctx: ctx, id: Stderr, ch: outputCh, closer: closer}
+	stdout := &OutputStream{ctx: ctx, id: StdoutID, ch: outputCh, closer: closer}
+	stderr := &OutputStream{ctx: ctx, id: StderrID, ch: outputCh, closer: closer}
 
 	return outputCh, stdout, stderr
 }
diff --git a/plugin/types.go b/plugin/types.go
index b3266fa..c487231 100644
--- a/plugin/types.go
+++ b/plugin/types.go
@@ -63,39 +63,15 @@ type Root interface {
 }
 
 // ExecOptions is a struct we can add new features to that must be serializable to JSON.
-// Examples of potential features: user, privileged, map of environment variables, timeout.
+// Examples of potential features: user, privileged, tty, map of environment variables, string of stdin, timeout.
 type ExecOptions struct {
-	// Stdin can be used to pass a stream of input to write to stdin when executing the command.
 	Stdin io.Reader
-
-	// Tty instructs the executor to allocate a TTY (pseudo-terminal), which lets Wash communicate
-	// with the running process via its Stdin. The TTY is used to send a process termination signal
-	// (Ctrl+C) via Stdin when the passed-in Exec context is cancelled.
-	//
-	// NOTE TO PLUGIN AUTHORS: The Tty option is only relevant for executors that do not have an API
-	// endpoint to stop a running command (e.g. Docker, Kubernetes). If your executor does have an
-	// API endpoint to stop a running command, then ignore the Tty option. Note that the reason we
-	// make Tty an option instead of having the relevant executors always attach a TTY is because
-	// attaching a TTY can change the behavior of the command that's being executed.
-	//
-	// NOTE TO CALLERS: The Tty option is useful for executing your own stream-like commands (e.g.
-	// tail -f), because it ensures that there are no orphaned processes after the request is
-	// cancelled/finished.
-	Tty bool
 }
 
-// ExecPacketType identifies the packet type.
-type ExecPacketType = string
-
-// Enumerates packet types.
-const (
-	Stdout ExecPacketType = "stdout"
-	Stderr ExecPacketType = "stderr"
-)
-
 // ExecOutputChunk is a struct containing a chunk of the Exec'ed cmd's output.
+// For StreamID, 0 = stdout and 1 = stderr.
 type ExecOutputChunk struct {
-	StreamID  ExecPacketType
+	StreamID  int8
 	Timestamp time.Time
 	Data      string
 	Err       error
@@ -117,7 +93,7 @@ type Execable interface {
 // Streamable is an entry that returns a stream of updates.
 type Streamable interface {
 	Entry
-	Stream(context.Context) (io.ReadCloser, error)
+	Stream(context.Context) (io.Reader, error)
 }
 
 // SizedReader returns a ReaderAt that can report its Size.
diff --git a/volume/core.go b/volume/core.go
index 6513173..4d0a10f 100644
--- a/volume/core.go
+++ b/volume/core.go
@@ -8,7 +8,6 @@ package volume
 
 import (
 	"context"
-	"io"
 	"sort"
 	"time"
 
@@ -26,8 +25,7 @@ type Interface interface {
 	VolumeList(context.Context) (DirMap, error)
 	// Accepts a path and returns the content associated with that path.
 	VolumeOpen(context.Context, string) (plugin.SizedReader, error)
-	// Accepts a path and streams updates to the content associated with that path.
-	VolumeStream(context.Context, string) (io.ReadCloser, error)
+	// TODO: add VolumeStream
 }
 
 // A Dir is a map of files in a directory to their attributes.
@@ -53,7 +51,7 @@ func List(ctx context.Context, impl Interface, path string) ([]plugin.Entry, err
 		if attr.Mode().IsDir() {
 			entries = append(entries, newDir(name, attr, impl, path+"/"+name))
 		} else {
-			entries = append(entries, newFile(name, attr, impl, path+"/"+name))
+			entries = append(entries, newFile(name, attr, impl.VolumeOpen, path+"/"+name))
 		}
 	}
 	// Sort entries so they have a deterministic order.
diff --git a/volume/file.go b/volume/file.go
index 80a9ad8..b0b7cb4 100644
--- a/volume/file.go
+++ b/volume/file.go
@@ -2,24 +2,26 @@ package volume
 
 import (
 	"context"
-	"io"
 	"time"
 
 	"github.com/puppetlabs/wash/plugin"
 )
 
+// contentCB accepts a path and returns the content associated with that path.
+type contentCB = func(context.Context, string) (plugin.SizedReader, error)
+
 // file represents a file in a volume that has content we can access.
 type file struct {
 	plugin.EntryBase
-	impl Interface
-	path string
+	contentcb contentCB
+	path      string
 }
 
 // newFile creates a VolumeFile.
-func newFile(name string, attr plugin.EntryAttributes, impl Interface, path string) *file {
+func newFile(name string, attr plugin.EntryAttributes, cb contentCB, path string) *file {
 	vf := &file{
 		EntryBase: plugin.NewEntry(name),
-		impl:      impl,
+		contentcb: cb,
 		path:      path,
 	}
 	vf.SetAttributes(attr)
@@ -30,9 +32,5 @@ func newFile(name string, attr plugin.EntryAttributes, impl Interface, path stri
 
 // Open returns the content of the file as a SizedReader.
 func (v *file) Open(ctx context.Context) (plugin.SizedReader, error) {
-	return v.impl.VolumeOpen(ctx, v.path)
-}
-
-func (v *file) Stream(ctx context.Context) (io.ReadCloser, error) {
-	return v.impl.VolumeStream(ctx, v.path)
+	return v.contentcb(ctx, v.path)
 }
diff --git a/volume/fs.go b/volume/fs.go
index 01c733d..3ea5373 100644
--- a/volume/fs.go
+++ b/volume/fs.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 
 	"github.com/puppetlabs/wash/activity"
 	"github.com/puppetlabs/wash/plugin"
@@ -42,11 +41,11 @@ func exec(ctx context.Context, executor plugin.Execable, cmdline []string) (*byt
 	for chunk := range result.OutputCh {
 		if chunk.Err != nil {
 			errs = append(errs, chunk.Err)
+		} else if chunk.StreamID == 0 {
+			activity.Record(ctx, "Stdout: %v", chunk.Data)
+			fmt.Fprint(&buf, chunk.Data)
 		} else {
-			activity.Record(ctx, "%v: %v", chunk.StreamID, chunk.Data)
-			if chunk.StreamID == plugin.Stdout {
-				fmt.Fprint(&buf, chunk.Data)
-			}
+			activity.Record(ctx, "Stderr: %v", chunk.Data)
 		}
 	}
 
@@ -86,42 +85,3 @@ func (d *FS) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, e
 	}
 	return bytes.NewReader(buf.Bytes()), nil
 }
-
-// VolumeStream satisfies the Interface required by List to stream file contents.
-func (d *FS) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
-	activity.Record(ctx, "Streaming %v on %v", path, plugin.ID(d.executor))
-	result, err := d.executor.Exec(ctx, "tail", []string{"-f", path}, plugin.ExecOptions{Tty: true})
-	if err != nil {
-		activity.Record(ctx, "Exec error in VolumeRead: %v", err)
-		return nil, err
-	}
-
-	r, w := io.Pipe()
-	go func() {
-		// Exec uses context; if it's canceled, the OutputCh will close. Close the writer.
-		var errs []error
-		for chunk := range result.OutputCh {
-			if chunk.Err != nil {
-				activity.Record(ctx, "Error on exec: %v", chunk.Err)
-				errs = append(errs, chunk.Err)
-				continue
-			}
-
-			activity.Record(ctx, "%v: %v", chunk.StreamID, chunk.Data)
-			if len(errs) == 0 {
-				if _, err := w.Write([]byte(chunk.Data)); err != nil {
-					activity.Record(ctx, "Error copying exec result: %v", err)
-					errs = append(errs, err)
-				}
-			}
-		}
-
-		if len(errs) > 0 {
-			err = w.CloseWithError(fmt.Errorf("Multiple errors from exec output: %v", errs))
-		} else {
-			err = w.Close()
-		}
-		activity.Record(ctx, "Closing write pipe: %v", err)
-	}()
-	return r, nil
-}


Test patch:
diff --git a/cmd/internal/find/parser/parseExpression_test.go b/cmd/internal/find/parser/parseExpression_test.go
index b4fc951..401fd42 100644
--- a/cmd/internal/find/parser/parseExpression_test.go
+++ b/cmd/internal/find/parser/parseExpression_test.go
@@ -67,6 +67,9 @@ func (suite *ParseExpressionTestSuite) TestParseExpressionNotOpParseErrors() {
 	suite.runTestCases(
 		// Test error when "-not" is supplied without an expression
 		nPETC("-not", "-not: no following expression"),
+		// Test error when "-not" is mixed with parentheses
+		nPETC("-not )", "): no beginning '('"),
+		nPETC("( -not )", "-not: no following expression"),
 		// Test error when "-not" is supplied w/ an atom that errors
 		nPETC("-not -name", "-name: requires additional arguments"),
 		// Test error when "-not" is followed by a binary operator
@@ -156,6 +159,7 @@ func (suite *ParseExpressionTestSuite) TestParseExpressionParensEval() {
 func (suite *ParseExpressionTestSuite) TestParseExpressionComplexErrors() {
 	suite.runTestCases(
 		nPETC("( -true ) -a )", "): no beginning '('"),
+		nPETC("-true -a -foo", "-foo: unknown primary or operator"),
 	)
 }
 
diff --git a/cmd/internal/find/parser/parseOptions_test.go b/cmd/internal/find/parser/parseOptions_test.go
index 7f79c9b..04afdaf 100644
--- a/cmd/internal/find/parser/parseOptions_test.go
+++ b/cmd/internal/find/parser/parseOptions_test.go
@@ -79,6 +79,7 @@ func (suite *ParseOptionsTestSuite) TestParseOptionsNoOptions() {
 		nPOTC("--", o, "--"),
 		nPOTC("-true", o, "-true"),
 		nPOTC("-a", o, "-a"),
+		nPOTC("(", o, "("),
 		nPOTC("foo bar baz", o, "foo bar baz"),
 	)
 }
diff --git a/cmd/internal/find/primary/action_test.go b/cmd/internal/find/primary/action_test.go
index 716edfe..de73e3f 100644
--- a/cmd/internal/find/primary/action_test.go
+++ b/cmd/internal/find/primary/action_test.go
@@ -13,12 +13,12 @@ type ActionPrimaryTestSuite struct {
 }
 
 func (suite *ActionPrimaryTestSuite) TestActionPrimaryInsufficientArgsError() {
-	_, _, err := actionPrimary.Parse([]string{"-action"})
+	_, _, err := actionPrimary.parse([]string{"-action"})
 	suite.Equal("-action: requires additional arguments", err.Error())
 }
 
 func (suite *ActionPrimaryTestSuite) TestActionPrimaryInvalidActionError() {
-	_, _, err := actionPrimary.Parse([]string{"-action", "foo"})
+	_, _, err := actionPrimary.parse([]string{"-action", "foo"})
 	suite.Regexp("foo is an invalid action. Valid actions are.*list", err)
 }
 
@@ -39,7 +39,7 @@ func (suite *ActionPrimaryTestSuite) TestActionPrimaryValidInput() {
 		inputStr := func() string {
 			return fmt.Sprintf("Input was '%v'", testCase.input)
 		}
-		p, tokens, err := actionPrimary.Parse([]string{"-action", testCase.input})
+		p, tokens, err := actionPrimary.parse([]string{"-action", testCase.input})
 		if suite.NoError(err, inputStr()) {
 			suite.Equal([]string{}, tokens)
 			e := types.Entry{}
diff --git a/cmd/internal/find/primary/meta/common_test.go b/cmd/internal/find/primary/meta/common_test.go
index 2abd876..d218ba6 100644
--- a/cmd/internal/find/primary/meta/common_test.go
+++ b/cmd/internal/find/primary/meta/common_test.go
@@ -5,7 +5,8 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/predicate"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -14,7 +15,7 @@ import (
 
 type ParserTestSuite struct {
 	suite.Suite
-	parser func(tokens []string) (predicate, []string, error)
+	parser predicate.GenericParser
 }
 
 type parserTestCase struct {
diff --git a/cmd/internal/find/primary/meta/primary_test.go b/cmd/internal/find/primary/meta/primary_test.go
deleted file mode 100644
index 6153979..0000000
--- a/cmd/internal/find/primary/meta/primary_test.go
+++ /dev/null
@@ -1,39 +0,0 @@
-// Package meta contains all the parsing logic for the `meta` primary
-package meta
-
-import (
-	"time"
-
-	"github.com/puppetlabs/wash/cmd/internal/find/params"
-	"github.com/puppetlabs/wash/cmd/internal/find/types"
-	"github.com/stretchr/testify/suite"
-)
-
-type PrimaryTestSuite struct {
-	suite.Suite
-}
-
-func (suite *PrimaryTestSuite) SetupTest() {
-	params.StartTime = time.Now()
-}
-
-func (suite *PrimaryTestSuite) TeardownTest() {
-	params.StartTime = time.Time{}
-}
-
-func (suite *PrimaryTestSuite) TestErrors() {
-	_, _, err := Primary.Parse(toTks("-m -true"))
-	suite.Regexp("-m: key sequences must begin with a '.'", err)
-
-	_, _, err = Primary.Parse(toTks("-m ."))
-	suite.Regexp("-m: expected a key sequence after '.'", err)
-}
-
-func (suite *PrimaryTestSuite) TestValidInput() {
-	p, tokens, err := Primary.Parse(toTks("-m -empty -size"))
-	if suite.NoError(err) {
-		suite.Equal(toTks("-size"), tokens)
-		e := types.Entry{}
-		suite.True(p(e))
-	}
-}
diff --git a/cmd/internal/find/primary/meta_test.go b/cmd/internal/find/primary/meta_test.go
new file mode 100644
index 0000000..58586e2
--- /dev/null
+++ b/cmd/internal/find/primary/meta_test.go
@@ -0,0 +1,43 @@
+package primary
+
+import (
+	"testing"
+	"time"
+
+	"github.com/puppetlabs/wash/cmd/internal/find/types"
+	"github.com/puppetlabs/wash/cmd/internal/find/params"
+	"github.com/stretchr/testify/suite"
+)
+
+type MetaPrimaryTestSuite struct {
+	suite.Suite
+}
+
+func (suite *MetaPrimaryTestSuite) SetupTest() {
+	params.StartTime = time.Now()
+}
+
+func (suite *MetaPrimaryTestSuite) TeardownTest() {
+	params.StartTime = time.Time{}
+}
+
+func (suite *MetaPrimaryTestSuite) TestMetaPrimaryErrors() {
+	_, _, err := metaPrimary.parse([]string{"-m", "-true"})
+	suite.Regexp("-m: key sequences must begin with a '.'", err)
+
+	_, _, err = metaPrimary.parse([]string{"-m", "."})
+	suite.Regexp("-m: expected a key sequence after '.'", err)
+}
+
+func (suite *MetaPrimaryTestSuite) TestMetaPrimaryValidInput() {
+	p, tokens, err := metaPrimary.parse([]string{"-m", "-empty", "-size"})
+	if suite.NoError(err) {
+		suite.Equal([]string{"-size"}, tokens)
+		e := types.Entry{}
+		suite.True(p(e))
+	}
+}
+
+func TestMetaPrimary(t *testing.T) {
+	suite.Run(t, new(MetaPrimaryTestSuite))
+}
diff --git a/cmd/internal/find/primary/name_test.go b/cmd/internal/find/primary/name_test.go
index 6189b62..614f5b9 100644
--- a/cmd/internal/find/primary/name_test.go
+++ b/cmd/internal/find/primary/name_test.go
@@ -12,10 +12,10 @@ type NamePrimaryTestSuite struct {
 }
 
 func (suite *NamePrimaryTestSuite) TestNamePrimaryErrors() {
-	_, _, err := namePrimary.Parse([]string{"-name"})
+	_, _, err := namePrimary.parse([]string{"-name"})
 	suite.Regexp("-name: requires additional arguments", err)
 
-	_, _, err = namePrimary.Parse([]string{"-name", "[a"})
+	_, _, err = namePrimary.parse([]string{"-name", "[a"})
 	suite.Regexp("-name: invalid glob: unexpected end of input", err)
 }
 
@@ -24,7 +24,7 @@ func (suite *NamePrimaryTestSuite) TestNamePrimaryValidInput() {
 	e1.CName = "a"
 	e2 := types.Entry{}
 	e2.CName = "b"
-	p, tokens, err := namePrimary.Parse([]string{"-name", "a"})
+	p, tokens, err := namePrimary.parse([]string{"-name", "a"})
 	if suite.NoError(err) {
 		suite.Equal([]string{}, tokens)
 		suite.Equal(true, p(e1))
diff --git a/cmd/internal/find/primary/numeric/bracket_test.go b/cmd/internal/find/primary/numeric/bracket_test.go
index 6c89cd0..4dc64f1 100644
--- a/cmd/internal/find/primary/numeric/bracket_test.go
+++ b/cmd/internal/find/primary/numeric/bracket_test.go
@@ -3,7 +3,7 @@ package numeric
 import (
 	"testing"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 	"github.com/stretchr/testify/suite"
 )
 
diff --git a/cmd/internal/find/primary/numeric/predicate_test.go b/cmd/internal/find/primary/numeric/predicate_test.go
index ea75990..633ac34 100644
--- a/cmd/internal/find/primary/numeric/predicate_test.go
+++ b/cmd/internal/find/primary/numeric/predicate_test.go
@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/puppetlabs/wash/cmd/internal/find/primary/errz"
+	"github.com/puppetlabs/wash/cmd/internal/find/parser/errz"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -54,13 +54,6 @@ func (suite *PredicateTestSuite) TestParsePredicateValidInput() {
 	}
 }
 
-func (suite *PredicateTestSuite) TestNot() {
-	g := Not(func(v int64) bool {
-		return true
-	})
-	suite.False(g(2))
-}
-
 func TestPredicate(t *testing.T) {
 	suite.Run(t, new(PredicateTestSuite))
 }
diff --git a/cmd/internal/find/primary/path_test.go b/cmd/internal/find/primary/path_test.go
index b509e88..9b4ba4b 100644
--- a/cmd/internal/find/primary/path_test.go
+++ b/cmd/internal/find/primary/path_test.go
@@ -12,10 +12,10 @@ type PathPrimaryTestSuite struct {
 }
 
 func (suite *PathPrimaryTestSuite) TestPathPrimaryErrors() {
-	_, _, err := pathPrimary.Parse([]string{"-path"})
+	_, _, err := pathPrimary.parse([]string{"-path"})
 	suite.Regexp("-path: requires additional arguments", err)
 
-	_, _, err = pathPrimary.Parse([]string{"-path", "[a"})
+	_, _, err = pathPrimary.parse([]string{"-path", "[a"})
 	suite.Regexp("-path: invalid glob: unexpected end of input", err)
 }
 
@@ -24,7 +24,7 @@ func (suite *PathPrimaryTestSuite) TestPathPrimaryValidInput() {
 	e1.NormalizedPath = "a"
 	e2 := types.Entry{}
 	e2.NormalizedPath = "b"
-	p, tokens, err := pathPrimary.Parse([]string{"-path", "a"})
+	p, tokens, err := pathPrimary.parse([]string{"-path", "a"})
 	if suite.NoError(err) {
 		suite.Equal([]string{}, tokens)
 		suite.Equal(true, p(e1))
diff --git a/cmd/internal/find/primary/size_test.go b/cmd/internal/find/primary/size_test.go
index 0e837a0..30d612e 100644
--- a/cmd/internal/find/primary/size_test.go
+++ b/cmd/internal/find/primary/size_test.go
@@ -14,7 +14,7 @@ type SizePrimaryTestSuite struct {
 }
 
 func (suite *SizePrimaryTestSuite) TestSizePrimaryInsufficientArgsError() {
-	_, _, err := sizePrimary.Parse([]string{"-size"})
+	_, _, err := sizePrimary.parse([]string{"-size"})
 	suite.Equal("-size: requires additional arguments", err.Error())
 }
 
@@ -27,7 +27,7 @@ func (suite *SizePrimaryTestSuite) TestSizePrimaryIllegalTimeValueError() {
 		"+1kb",
 	}
 	for _, v := range illegalValues {
-		_, _, err := sizePrimary.Parse([]string{"-size", v})
+		_, _, err := sizePrimary.parse([]string{"-size", v})
 		msg := fmt.Sprintf("-size: %v: illegal size value", v)
 		suite.Equal(msg, err.Error())
 	}
@@ -56,7 +56,7 @@ func (suite *SizePrimaryTestSuite) TestSizePrimaryValidInput() {
 		inputStr := func() string {
 			return fmt.Sprintf("Input was '%v'", testCase.input)
 		}
-		p, tokens, err := sizePrimary.Parse([]string{"-size", testCase.input})
+		p, tokens, err := sizePrimary.parse([]string{"-size", testCase.input})
 		if suite.NoError(err, inputStr()) {
 			suite.Equal([]string{}, tokens)
 			e := types.Entry{}
diff --git a/cmd/internal/find/primary/timeAttr_test.go b/cmd/internal/find/primary/timeAttr_test.go
index 0283f89..d3cb4cd 100644
--- a/cmd/internal/find/primary/timeAttr_test.go
+++ b/cmd/internal/find/primary/timeAttr_test.go
@@ -60,7 +60,7 @@ func (suite *TimeAttrPrimaryTestSuite) TestGetTimeAttrValue() {
 // These tests use the ctimePrimary as the representative test case
 
 func (suite *TimeAttrPrimaryTestSuite) TestTimeAttrPrimaryInsufficientArgsError() {
-	_, _, err := ctimePrimary.Parse([]string{"-ctime"})
+	_, _, err := ctimePrimary.parse([]string{"-ctime"})
 	suite.Equal("-ctime: requires additional arguments", err.Error())
 }
 
@@ -76,7 +76,7 @@ func (suite *TimeAttrPrimaryTestSuite) TestTimeAttrPrimaryIllegalTimeValueError(
 		"+1h30min",
 	}
 	for _, v := range illegalValues {
-		_, _, err := ctimePrimary.Parse([]string{"-ctime", v})
+		_, _, err := ctimePrimary.parse([]string{"-ctime", v})
 		msg := fmt.Sprintf("-ctime: %v: illegal time value", v)
 		suite.Equal(msg, err.Error())
 	}
@@ -106,7 +106,7 @@ func (suite *TimeAttrPrimaryTestSuite) TestTimeAttrPrimaryValidInput() {
 		inputStr := func() string {
 			return fmt.Sprintf("Input was '%v'", testCase.input)
 		}
-		p, tokens, err := ctimePrimary.Parse([]string{"-ctime", testCase.input})
+		p, tokens, err := ctimePrimary.parse([]string{"-ctime", testCase.input})
 		if suite.NoError(err, inputStr()) {
 			suite.Equal([]string{}, tokens)
 			e := types.Entry{}
diff --git a/plugin/cleanupReader_test.go b/plugin/cleanupReader_test.go
deleted file mode 100644
index 35faf83..0000000
--- a/plugin/cleanupReader_test.go
+++ /dev/null
@@ -1,35 +0,0 @@
-package plugin
-
-import (
-	"io"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestCleanupReader(t *testing.T) {
-	called := false
-	cleanup := func() {
-		called = true
-	}
-
-	r, w := io.Pipe()
-	defer w.Close()
-	rdr := CleanupReader{ReadCloser: r, Cleanup: cleanup}
-	go func() {
-		_, err := w.Write([]byte("hello"))
-		assert.NoError(t, err)
-	}()
-
-	buf := make([]byte, 5)
-	n, err := rdr.Read(buf)
-	assert.NoError(t, err)
-	assert.Equal(t, 5, n)
-	assert.Equal(t, "hello", string(buf))
-	assert.False(t, called)
-
-	rdr.Close()
-	assert.True(t, called)
-	_, err = w.Write([]byte("goodbye"))
-	assert.Error(t, err)
-}
diff --git a/plugin/outputStream_test.go b/plugin/outputStream_test.go
index 877064c..c56f7af 100644
--- a/plugin/outputStream_test.go
+++ b/plugin/outputStream_test.go
@@ -15,10 +15,31 @@ type OutputStreamTestSuite struct {
 }
 
 func (suite *OutputStreamTestSuite) EqualChunk(expected ExecOutputChunk, actual ExecOutputChunk) bool {
-	expectedStreamName := expected.StreamID
+	streamName := func(id int8) (string, error) {
+		if id == StdoutID {
+			return "stdout", nil
+		}
+
+		if id == StderrID {
+			return "stderr", nil
+		}
+
+		return "", fmt.Errorf("unknown ID: %v", id)
+	}
+
+	expectedStreamName, err := streamName(expected.StreamID)
+	if err != nil {
+		suite.Fail("could not get the stream name of the expected chunk: %v", err)
+	}
+
+	actualStreamName, err := streamName(actual.StreamID)
+	if err != nil {
+		suite.Fail("actual chunk has an invalid stream ID: %v", err)
+	}
+
 	eqlStreamName := suite.Equal(
 		expectedStreamName,
-		actual.StreamID,
+		actualStreamName,
 		fmt.Sprintf("The sent ExecOutputChunk should have come from the %v stream", expectedStreamName),
 	)
 
@@ -62,7 +83,7 @@ func (suite *OutputStreamTestSuite) TestWrite() {
 	ch := make(chan ExecOutputChunk, 1)
 	defer close(ch)
 
-	stream := OutputStream{ctx: ctx, id: Stdout, ch: ch}
+	stream := OutputStream{ctx: ctx, id: StdoutID, ch: ch}
 
 	// Test a successful write
 	data := []byte("data")
@@ -73,7 +94,7 @@ func (suite *OutputStreamTestSuite) TestWrite() {
 	select {
 	case chunk := <-stream.ch:
 		suite.EqualChunk(
-			ExecOutputChunk{StreamID: Stdout, Data: string(data)},
+			ExecOutputChunk{StreamID: StdoutID, Data: string(data)},
 			chunk,
 		)
 	default:
@@ -87,7 +108,7 @@ func (suite *OutputStreamTestSuite) TestWrite() {
 	select {
 	case chunk := <-stream.ch:
 		suite.EqualChunk(
-			ExecOutputChunk{StreamID: Stdout, Err: ctx.Err()},
+			ExecOutputChunk{StreamID: StdoutID, Err: ctx.Err()},
 			chunk,
 		)
 		suite.Equal(stream.sentCtxErr, true, "The stream should mark that the context's error was sent")
@@ -121,7 +142,7 @@ func (suite *OutputStreamTestSuite) TestClose() {
 func (suite *OutputStreamTestSuite) TestCloseWithError() {
 	newOutputStream := func(ctx context.Context) OutputStream {
 		ch := make(chan ExecOutputChunk, 1)
-		return OutputStream{ctx: ctx, id: Stdout, ch: ch, closer: &multiCloser{ch: ch, countdown: 1}}
+		return OutputStream{ctx: ctx, id: StdoutID, ch: ch, closer: &multiCloser{ch: ch, countdown: 1}}
 	}
 
 	// Test that if err == nil, then nothing was sent to the channel
@@ -176,6 +197,15 @@ func (suite *OutputStreamTestSuite) TestCloseWithError() {
 func (suite *OutputStreamTestSuite) TestCreateExecOutputStreams() {
 	outputCh, stdout, stderr := CreateExecOutputStreams(context.Background())
 
+	assertStreamID := func(streamName string, expectedID int8, stream *OutputStream) {
+		if !suite.Equal(expectedID, stream.id) {
+			msg := fmt.Sprintf("expected %v stream to have stream ID %v, got %v instead", streamName, expectedID, stream.id)
+			suite.FailNow(msg)
+		}
+	}
+	assertStreamID("stdout", StdoutID, stdout)
+	assertStreamID("stderr", StderrID, stderr)
+
 	// Our simulated command alternates writing to stdout + stderr
 	expectedChunksCh := make(chan ExecOutputChunk, 1)
 	go func() {
diff --git a/volume/dir_test.go b/volume/dir_test.go
index 371dd49..f9a7adc 100644
--- a/volume/dir_test.go
+++ b/volume/dir_test.go
@@ -2,7 +2,6 @@ package volume
 
 import (
 	"context"
-	"io"
 	"os"
 	"strings"
 	"testing"
@@ -12,20 +11,16 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-type mockDirEntry struct {
+type mockEntry struct {
 	plugin.EntryBase
 	dmap DirMap
 }
 
-func (m *mockDirEntry) VolumeList(context.Context) (DirMap, error) {
+func (m *mockEntry) VolumeList(context.Context) (DirMap, error) {
 	return m.dmap, nil
 }
 
-func (m *mockDirEntry) VolumeOpen(context.Context, string) (plugin.SizedReader, error) {
-	return nil, nil
-}
-
-func (m *mockDirEntry) VolumeStream(context.Context, string) (io.ReadCloser, error) {
+func (m *mockEntry) VolumeOpen(context.Context, string) (plugin.SizedReader, error) {
 	return nil, nil
 }
 
@@ -34,7 +29,7 @@ func TestVolumeDir(t *testing.T) {
 	assert.Nil(t, err)
 
 	plugin.SetTestCache(datastore.NewMemCache())
-	entry := mockDirEntry{EntryBase: plugin.NewEntry("mine"), dmap: dmap}
+	entry := mockEntry{EntryBase: plugin.NewEntry("mine"), dmap: dmap}
 	entry.SetTestID("/mine")
 
 	assert.NotNil(t, dmap[""]["path"])
diff --git a/volume/file_test.go b/volume/file_test.go
index c32a2ab..0ce1177 100644
--- a/volume/file_test.go
+++ b/volume/file_test.go
@@ -3,8 +3,6 @@ package volume
 import (
 	"context"
 	"errors"
-	"io"
-	"io/ioutil"
 	"strings"
 	"testing"
 	"time"
@@ -13,43 +11,19 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-type mockFileEntry struct {
-	plugin.EntryBase
-	content string
-	err     error
-}
-
-func (m *mockFileEntry) VolumeList(context.Context) (DirMap, error) {
-	return nil, nil
-}
-
-func (m *mockFileEntry) VolumeOpen(context.Context, string) (plugin.SizedReader, error) {
-	if m.err != nil {
-		return nil, m.err
-	}
-	return strings.NewReader(m.content), nil
-}
-
-func (m *mockFileEntry) VolumeStream(context.Context, string) (io.ReadCloser, error) {
-	if m.err != nil {
-		return nil, m.err
-	}
-	return ioutil.NopCloser(strings.NewReader(m.content)), nil
-}
-
 func TestVolumeFile(t *testing.T) {
 	now := time.Now()
 	initialAttr := plugin.EntryAttributes{}
 	initialAttr.SetCtime(now)
-
-	impl := &mockFileEntry{EntryBase: plugin.NewEntry("parent"), content: "hello"}
-	vf := newFile("mine", initialAttr, impl, "my path")
+	vf := newFile("mine", initialAttr, func(ctx context.Context, path string) (plugin.SizedReader, error) {
+		assert.Equal(t, "my path", path)
+		return strings.NewReader("hello"), nil
+	}, "my path")
 
 	attr := plugin.Attributes(vf)
 	expectedAttr := plugin.EntryAttributes{}
 	expectedAttr.SetCtime(now)
 	assert.Equal(t, expectedAttr, attr)
-
 	rdr, err := vf.Open(context.Background())
 	assert.Nil(t, err)
 	if assert.NotNil(t, rdr) {
@@ -59,26 +33,14 @@ func TestVolumeFile(t *testing.T) {
 		assert.Equal(t, int64(n), rdr.Size())
 		assert.Equal(t, "hello", string(buf))
 	}
-
-	rdr2, err := vf.Stream(context.Background())
-	assert.Nil(t, err)
-	if assert.NotNil(t, rdr2) {
-		buf, err := ioutil.ReadAll(rdr2)
-		if assert.NoError(t, err) {
-			assert.Equal(t, "hello", string(buf))
-		}
-	}
 }
 
 func TestVolumeFileErr(t *testing.T) {
-	impl := &mockFileEntry{EntryBase: plugin.NewEntry("parent"), err: errors.New("fail")}
-	vf := newFile("mine", plugin.EntryAttributes{}, impl, "my path")
+	vf := newFile("mine", plugin.EntryAttributes{}, func(ctx context.Context, path string) (plugin.SizedReader, error) {
+		return nil, errors.New("fail")
+	}, "my path")
 
 	rdr, err := vf.Open(context.Background())
 	assert.Nil(t, rdr)
 	assert.Equal(t, errors.New("fail"), err)
-
-	rdr2, err := vf.Stream(context.Background())
-	assert.Nil(t, rdr2)
-	assert.Equal(t, errors.New("fail"), err)
 }
diff --git a/volume/fs_test.go b/volume/fs_test.go
index 257ae5f..3bad252 100644
--- a/volume/fs_test.go
+++ b/volume/fs_test.go
@@ -42,7 +42,7 @@ func createResult(data string) plugin.ExecResult {
 		OutputCh:   outputch,
 		ExitCodeCB: func() (int, error) { return 0, nil },
 	}
-	outputch <- plugin.ExecOutputChunk{StreamID: plugin.Stdout, Data: data}
+	outputch <- plugin.ExecOutputChunk{Data: data}
 	close(outputch)
 	return execResult
 }