feat: DTail fork — server/client feature development

Squashed development of the snonux/dtail fork's product code (internal/, cmd/)
since diverging from mimecast/dtail. Major areas:

- Read/output path: the former "turbo" channel-less path is now the single,
  default server-side read/output path for cat/grep/tail and MapReduce; the old
  channel-based path and its config/env toggles were removed.
- MapReduce: single aggregate implementation (server + serverless) fed directly
  by a processor pipeline, with input-exhausted finalization via the shutdown
  coordinator; high-concurrency and data-race fixes.
- Journal source reads (journal:unit.service) via journalctl, Linux-gated behind
  a journal-v1 capability.
- Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys,
  registered over an authenticated session (AUTHKEY), checked before
  authorized_keys.
- Interactive query reload (--interactive-query) with SESSION START/UPDATE
  generation boundaries and capability negotiation.
- Client-side deadlines: --timeout / --shutdownAfter as context deadlines;
  follow shutdown handling.
- Client logging: diagnostics-only daily log by default, opt-in payload tee via
  --log-payload.
- Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel
  leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

47 files changed, 4168 insertions, 501 deletions

diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go
index c50c7a1..3281353 100644
--- a/internal/mapr/aggregateset.go
+++ b/internal/mapr/aggregateset.go
@@ -46,6 +46,10 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 		case Sum:
 			fallthrough
 		case Avg:
+			fallthrough
+		case Percentage:
+			fallthrough
+		case Percentile:
 			value := set.FValues[storage]
 			s.addFloat(storage, value)
 		case Min:
@@ -67,21 +71,25 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 	return nil
 }
 
-// Serialize the aggregate set so it can be sent over the wire.
-func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) {
+// Serialize the aggregate set so it can be sent over the wire. Returns true
+// when the serialized message was successfully sent, and false when the
+// context was cancelled before the send completed. Callers that own the
+// source state (e.g. Aggregate) must re-merge unsent sets so data is
+// not silently lost.
+func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) bool {
 	dlog.Common.Trace("Serialising mapr.AggregateSet", s)
 	sb := pool.BuilderBuffer.Get().(*strings.Builder)
 	defer pool.RecycleBuilderBuffer(sb)
 
 	sb.WriteString(groupKey)
 	sb.WriteString(protocol.AggregateDelimiter)
-	sb.WriteString(fmt.Sprintf("%d", s.Samples))
+	sb.WriteString(strconv.Itoa(s.Samples))
 	sb.WriteString(protocol.AggregateDelimiter)
 
 	for k, v := range s.FValues {
 		sb.WriteString(k)
 		sb.WriteString(protocol.AggregateKVDelimiter)
-		sb.WriteString(fmt.Sprintf("%v", v))
+		sb.WriteString(strconv.FormatFloat(v, 'f', -1, 64))
 		sb.WriteString(protocol.AggregateDelimiter)
 	}
 
@@ -94,7 +102,9 @@ func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<-
 
 	select {
 	case ch <- sb.String():
+		return true
 	case <-ctx.Done():
+		return false
 	}
 }
 
@@ -177,6 +187,10 @@ func (s *AggregateSet) Aggregate(key string, agg AggregateOperation, value strin
 	case Sum:
 		fallthrough
 	case Avg:
+		fallthrough
+	case Percentage:
+		fallthrough
+	case Percentile:
 		s.addFloat(key, f)
 	case Min:
 		s.addFloatMin(key, f)
diff --git a/internal/mapr/client/aggregate.go b/internal/mapr/client/aggregate.go
index 2e9b61a..9989e8f 100644
--- a/internal/mapr/client/aggregate.go
+++ b/internal/mapr/client/aggregate.go
@@ -12,30 +12,46 @@ import (
 
 // Aggregate mapreduce data on the DTail client side.
 type Aggregate struct {
-	// This is the mapr query specified on the command line.
-	query *mapr.Query
 	// This represents aggregated data of a single remote server.
 	group *mapr.GroupSet
-	// This represents the merged aggregated data of all servers.
-	globalGroup *mapr.GlobalGroupSet
+	// Shared per-client session state.
+	session *SessionState
+	// The currently tracked shared generation.
+	generation uint64
 	// The server we aggregate the data for (logging and debugging purposes only)
 	server string
 }
 
 // NewAggregate create new client aggregator.
-func NewAggregate(server string, query *mapr.Query,
-	globalGroup *mapr.GlobalGroupSet) *Aggregate {
+func NewAggregate(server string, session *SessionState) *Aggregate {
+	generation := uint64(0)
+	if session != nil {
+		generation = session.Snapshot().Generation
+	}
 
 	return &Aggregate{
-		query:       query,
-		group:       mapr.NewGroupSet(),
-		globalGroup: globalGroup,
-		server:      server,
+		group:      mapr.NewGroupSet(),
+		session:    session,
+		generation: generation,
+		server:     server,
 	}
 }
 
 // Aggregate data from mapr log line into local (and global) group sets.
 func (a *Aggregate) Aggregate(message string) error {
+	if a.session == nil {
+		return fmt.Errorf("missing client mapreduce session state")
+	}
+
+	snapshot := a.session.Snapshot()
+	if snapshot.Query == nil || snapshot.GlobalGroup == nil {
+		return fmt.Errorf("missing client mapreduce query state")
+	}
+	if snapshot.Generation != a.generation {
+		a.group.InitSet()
+		a.generation = snapshot.Generation
+	}
+
 	parts := strings.Split(message, protocol.AggregateDelimiter)
 	if len(parts) < 4 {
 		return fmt.Errorf("aggregate message without any real data")
@@ -51,7 +67,7 @@ func (a *Aggregate) Aggregate(message string) error {
 	set := a.group.GetSet(groupKey)
 	var addedSamples bool
 
-	for _, sc := range a.query.Select {
+	for _, sc := range snapshot.Query.Select {
 		if val, ok := fields[sc.FieldStorage]; ok {
 			if err := set.Aggregate(sc.FieldStorage, sc.Operation, val, true); err != nil {
 				dlog.Client.Error(err)
@@ -65,9 +81,9 @@ func (a *Aggregate) Aggregate(message string) error {
 	}
 
 	// Merge data from group into global group.
-	isMerged, err := a.globalGroup.MergeNoblock(a.query, a.group)
+	isMerged, err := snapshot.GlobalGroup.MergeNoblock(snapshot.Query, a.group)
 	if err != nil {
-		panic(err)
+		return fmt.Errorf("unable to merge aggregate data for server %s: %w", a.server, err)
 	}
 	if isMerged {
 		// Re-init local group (make it empty again).
@@ -76,15 +92,40 @@ func (a *Aggregate) Aggregate(message string) error {
 	return nil
 }
 
+// Flush merges any pending per-server aggregate state into the shared global group.
+// The normal hot path uses MergeNoblock to avoid stalling on the global merge lock.
+// During shutdown we need a blocking flush so the last local batch is not lost.
+func (a *Aggregate) Flush() error {
+	if a.session == nil {
+		return fmt.Errorf("missing client mapreduce session state")
+	}
+
+	snapshot := a.session.Snapshot()
+	if snapshot.Query == nil || snapshot.GlobalGroup == nil {
+		return nil
+	}
+	if snapshot.Generation != a.generation {
+		a.group.InitSet()
+		a.generation = snapshot.Generation
+		return nil
+	}
+
+	if err := snapshot.GlobalGroup.Merge(snapshot.Query, a.group); err != nil {
+		return fmt.Errorf("unable to flush aggregate data for server %s: %w", a.server, err)
+	}
+	a.group.InitSet()
+	return nil
+}
+
 // Create a map of key-value pairs from a part list such as ["foo=bar",  "bar=baz"].
 func (a *Aggregate) makeFields(parts []string) map[string]string {
 	fields := make(map[string]string, len(parts))
 	for _, part := range parts {
-		kv := strings.SplitN(part, protocol.AggregateKVDelimiter, 2)
-		if len(kv) != 2 {
+		key, value, ok := strings.Cut(part, protocol.AggregateKVDelimiter)
+		if !ok {
 			continue
 		}
-		fields[kv[0]] = kv[1]
+		fields[key] = value
 	}
 	return fields
 }
diff --git a/internal/mapr/client/aggregate_test.go b/internal/mapr/client/aggregate_test.go
new file mode 100644
index 0000000..3387a63
--- /dev/null
+++ b/internal/mapr/client/aggregate_test.go
@@ -0,0 +1,96 @@
+package client
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/mimecast/dtail/internal/mapr"
+	"github.com/mimecast/dtail/internal/protocol"
+)
+
+func TestAggregateResetsPendingLocalStateOnGenerationChange(t *testing.T) {
+	query := mustSessionStateQuery(t, "select status,count(status) from stats group by status")
+	state := NewSessionState(query)
+	aggregate := NewAggregate("srv1", state)
+	countStorage := aggregateCountStorage(t, query)
+
+	oldSet := aggregate.group.GetSet("ERROR")
+	oldSet.Samples = 1
+	oldSet.FValues[countStorage] = 1
+
+	rawQuery := "select status,count(status) from warnings group by status"
+	if _, err := state.CommitQuery(rawQuery, 2); err != nil {
+		t.Fatalf("CommitQuery() error = %v", err)
+	}
+
+	snapshot := state.Snapshot()
+	message := strings.Join([]string{
+		"WARN",
+		"1",
+		aggregateCountStorage(t, snapshot.Query) + protocol.AggregateKVDelimiter + "1",
+		"",
+	}, protocol.AggregateDelimiter)
+
+	if err := aggregate.Aggregate(message); err != nil {
+		t.Fatalf("Aggregate() error = %v", err)
+	}
+
+	result, numRows, err := snapshot.GlobalGroup.Result(snapshot.Query, 10, nil)
+	if err != nil {
+		t.Fatalf("Result() error = %v", err)
+	}
+	if numRows != 1 {
+		t.Fatalf("numRows = %d, want 1", numRows)
+	}
+	if !strings.Contains(result, "1") {
+		t.Fatalf("expected one new-generation aggregate row, got %q", result)
+	}
+}
+
+func TestAggregateRejectsMalformedMessage(t *testing.T) {
+	query := mustSessionStateQuery(t, "select count(status) from stats group by status")
+	state := NewSessionState(query)
+	aggregate := NewAggregate("srv1", state)
+
+	if err := aggregate.Aggregate("broken"); err == nil {
+		t.Fatalf("expected Aggregate() to reject malformed messages")
+	}
+}
+
+func TestAggregateFlushMergesPendingLocalState(t *testing.T) {
+	query := mustSessionStateQuery(t, "select status,count(status) from stats group by status")
+	state := NewSessionState(query)
+	aggregate := NewAggregate("srv1", state)
+	countStorage := aggregateCountStorage(t, query)
+
+	set := aggregate.group.GetSet("ERROR")
+	set.Samples = 3
+	set.FValues[countStorage] = 3
+
+	if err := aggregate.Flush(); err != nil {
+		t.Fatalf("Flush() error = %v", err)
+	}
+
+	result, numRows, err := state.Snapshot().GlobalGroup.Result(query, 10, nil)
+	if err != nil {
+		t.Fatalf("Result() error = %v", err)
+	}
+	if numRows != 1 {
+		t.Fatalf("numRows = %d, want 1", numRows)
+	}
+	if !strings.Contains(result, "3") {
+		t.Fatalf("expected flushed aggregate row, got %q", result)
+	}
+}
+
+func aggregateCountStorage(t *testing.T, query *mapr.Query) string {
+	t.Helper()
+
+	for _, selectCondition := range query.Select {
+		if selectCondition.Operation == mapr.Count {
+			return selectCondition.FieldStorage
+		}
+	}
+	t.Fatalf("query %q does not contain count() storage", query.RawQuery)
+	return ""
+}
diff --git a/internal/mapr/client/session_state.go b/internal/mapr/client/session_state.go
new file mode 100644
index 0000000..1983644
--- /dev/null
+++ b/internal/mapr/client/session_state.go
@@ -0,0 +1,95 @@
+package client
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/mimecast/dtail/internal/mapr"
+)
+
+// SessionSnapshot captures the current client-side mapreduce session state.
+type SessionSnapshot struct {
+	Generation  uint64
+	Query       *mapr.Query
+	GlobalGroup *mapr.GlobalGroupSet
+	LastResult  string
+}
+
+// SessionState keeps the mutable mapreduce query state shared by the client
+// reporter and per-server handlers.
+type SessionState struct {
+	mu         sync.RWMutex
+	generation uint64
+	query      *mapr.Query
+	global     *mapr.GlobalGroupSet
+	lastResult string
+	changedCh  chan struct{}
+}
+
+// NewSessionState returns a new shared mapreduce session state.
+func NewSessionState(query *mapr.Query) *SessionState {
+	return &SessionState{
+		query:     query,
+		global:    mapr.NewGlobalGroupSet(),
+		changedCh: make(chan struct{}, 1),
+	}
+}
+
+// Snapshot returns a point-in-time copy of the shared mapreduce state.
+func (s *SessionState) Snapshot() SessionSnapshot {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	return SessionSnapshot{
+		Generation:  s.generation,
+		Query:       s.query,
+		GlobalGroup: s.global,
+		LastResult:  s.lastResult,
+	}
+}
+
+// Changes returns a channel that is signaled whenever a new generation is committed.
+func (s *SessionState) Changes() <-chan struct{} {
+	return s.changedCh
+}
+
+// CommitQuery resets the shared aggregation state for a newly accepted query generation.
+func (s *SessionState) CommitQuery(rawQuery string, generation uint64) (*mapr.Query, error) {
+	query, err := mapr.NewQuery(rawQuery)
+	if err != nil {
+		return nil, fmt.Errorf("parse session query: %w", err)
+	}
+
+	s.mu.Lock()
+	s.generation = generation
+	s.query = query
+	s.global = mapr.NewGlobalGroupSet()
+	s.lastResult = ""
+	s.mu.Unlock()
+
+	s.notifyChange()
+	return query, nil
+}
+
+// CommitRenderedResult stores the last rendered result for the active generation.
+func (s *SessionState) CommitRenderedResult(generation uint64, result string) (changed bool, ok bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.generation != generation {
+		return false, false
+	}
+	if s.lastResult == result {
+		return false, true
+	}
+
+	s.lastResult = result
+	return true, true
+}
+
+func (s *SessionState) notifyChange() {
+	select {
+	case s.changedCh <- struct{}{}:
+	default:
+	}
+}
diff --git a/internal/mapr/client/session_state_test.go b/internal/mapr/client/session_state_test.go
new file mode 100644
index 0000000..f43ca70
--- /dev/null
+++ b/internal/mapr/client/session_state_test.go
@@ -0,0 +1,81 @@
+package client
+
+import (
+	"testing"
+
+	"github.com/mimecast/dtail/internal/mapr"
+)
+
+func TestSessionStateCommitQueryResetsGenerationAndResults(t *testing.T) {
+	query := mustSessionStateQuery(t, "select count(status) from stats group by status")
+	state := NewSessionState(query)
+
+	initial := state.Snapshot()
+	group := mapr.NewGroupSet()
+	set := group.GetSet("ERROR")
+	set.Samples = 1
+	set.FValues[query.Select[0].FieldStorage] = 1
+	if err := initial.GlobalGroup.Merge(query, group); err != nil {
+		t.Fatalf("Merge() error = %v", err)
+	}
+	if changed, ok := state.CommitRenderedResult(initial.Generation, "old-result"); !ok || !changed {
+		t.Fatalf("CommitRenderedResult() = changed:%v ok:%v, want changed and ok", changed, ok)
+	}
+
+	rawQuery := "select count(status) from warnings group by status"
+	updatedQuery, err := state.CommitQuery(rawQuery, 3)
+	if err != nil {
+		t.Fatalf("CommitQuery() error = %v", err)
+	}
+	if updatedQuery == nil || updatedQuery.RawQuery != rawQuery {
+		t.Fatalf("unexpected updated query: %#v", updatedQuery)
+	}
+
+	select {
+	case <-state.Changes():
+	default:
+		t.Fatalf("expected change notification after CommitQuery")
+	}
+
+	updated := state.Snapshot()
+	if updated.Generation != 3 {
+		t.Fatalf("generation = %d, want 3", updated.Generation)
+	}
+	if updated.Query == nil || updated.Query.RawQuery != rawQuery {
+		t.Fatalf("unexpected query after commit: %#v", updated.Query)
+	}
+	if !updated.GlobalGroup.IsEmpty() {
+		t.Fatalf("expected committed global group to be reset")
+	}
+	if updated.LastResult != "" {
+		t.Fatalf("last result = %q, want empty", updated.LastResult)
+	}
+}
+
+func TestSessionStateCommitQueryRejectsInvalidQuery(t *testing.T) {
+	query := mustSessionStateQuery(t, "select count(status) from stats group by status")
+	state := NewSessionState(query)
+	before := state.Snapshot()
+
+	if _, err := state.CommitQuery("select from", 5); err == nil {
+		t.Fatalf("expected CommitQuery() to reject invalid query")
+	}
+
+	after := state.Snapshot()
+	if after.Generation != before.Generation {
+		t.Fatalf("generation changed on invalid query: got %d want %d", after.Generation, before.Generation)
+	}
+	if after.Query == nil || after.Query.RawQuery != before.Query.RawQuery {
+		t.Fatalf("query changed on invalid query: before=%#v after=%#v", before.Query, after.Query)
+	}
+}
+
+func mustSessionStateQuery(t *testing.T, queryStr string) *mapr.Query {
+	t.Helper()
+
+	query, err := mapr.NewQuery(queryStr)
+	if err != nil {
+		t.Fatalf("NewQuery(%q) error = %v", queryStr, err)
+	}
+	return query
+}
diff --git a/internal/mapr/funcs/function.go b/internal/mapr/funcs/function.go
index 418d86f..2f21d5a 100644
--- a/internal/mapr/funcs/function.go
+++ b/internal/mapr/funcs/function.go
@@ -20,20 +20,10 @@ type Function struct {
 type FunctionStack []Function
 
 // NewFunctionStack parses the input string, e.g. foo(bar("arg")) and returns
-// a corresponding function stack.
+// a corresponding function stack. It returns an error for malformed inputs
+// such as unbalanced parentheses (e.g. "foo(", "foo(bar)baz").
 func NewFunctionStack(in string) (FunctionStack, string, error) {
 	var fs FunctionStack
-	getCallback := func(name string) (CallbackFunc, error) {
-		var cb CallbackFunc
-		switch name {
-		case "md5sum":
-			return Md5Sum, nil
-		case "maskdigits":
-			return MaskDigits, nil
-		default:
-			return cb, fmt.Errorf("unknown function '%s'", name)
-		}
-	}
 
 	aux := in
 	for strings.HasSuffix(aux, ")") {
@@ -43,16 +33,64 @@ func NewFunctionStack(in string) (FunctionStack, string, error) {
 		}
 		name := aux[0:index]
 
-		call, err := getCallback(name)
+		call, err := lookupCallback(name)
 		if err != nil {
 			return fs, "", err
 		}
 		fs = append(fs, Function{name, call})
+		// Strip the outer function name and its enclosing parens, leaving
+		// only the argument expression for the next iteration.
 		aux = aux[index+1 : len(aux)-1]
 	}
+
+	// Validate that no unbalanced parentheses remain in the argument string.
+	// Inputs like "foo(bar)baz" leave "bar)baz" after stripping, and inputs
+	// ending with "(" (no closing ")") are accepted as plain field literals
+	// without this check — both produce silently wrong behavior.
+	if err := validateParenBalance(aux, in); err != nil {
+		return fs, "", err
+	}
+
 	return fs, aux, nil
 }
 
+// lookupCallback maps a function name to its CallbackFunc implementation.
+// It returns an error for unrecognised names so callers get a clear message.
+func lookupCallback(name string) (CallbackFunc, error) {
+	switch name {
+	case "md5sum":
+		return Md5Sum, nil
+	case "maskdigits":
+		return MaskDigits, nil
+	default:
+		var zero CallbackFunc
+		return zero, fmt.Errorf("unknown function '%s'", name)
+	}
+}
+
+// validateParenBalance checks that the remaining argument string contains no
+// unbalanced parentheses. A negative depth means a stray ')' was found; a
+// non-zero depth after the loop means an unclosed '(' was found. The original
+// full expression is included in the error message for context.
+func validateParenBalance(aux, original string) error {
+	depth := 0
+	for _, r := range aux {
+		switch r {
+		case '(':
+			depth++
+		case ')':
+			depth--
+		}
+		if depth < 0 {
+			return fmt.Errorf("malformed function expression %q: unexpected ')' in argument", original)
+		}
+	}
+	if depth != 0 {
+		return fmt.Errorf("malformed function expression %q: unclosed '(' in argument", original)
+	}
+	return nil
+}
+
 // Call the function stack.
 func (fs FunctionStack) Call(str string) string {
 	for i := len(fs) - 1; i >= 0; i-- {
diff --git a/internal/mapr/funcs/function_test.go b/internal/mapr/funcs/function_test.go
index 8b5d8b7..8227817 100644
--- a/internal/mapr/funcs/function_test.go
+++ b/internal/mapr/funcs/function_test.go
@@ -2,51 +2,91 @@ package funcs
 
 import "testing"
 
-func TestFunction(t *testing.T) {
-	input := "md5sum($line)"
-	fs, arg, err := NewFunctionStack(input)
-	if err != nil {
-		t.Errorf("error parsing function input '%s': %s (%v)\n",
-			input, err.Error(), fs)
-	}
-	if arg != "$line" {
-		t.Errorf("error parsing function input '%s': expected argument '$line' but "+
-			"got '%s' (%v)\n", input, arg, fs)
-	}
-	t.Log(input, fs, arg)
+func TestFunctionStackValid(t *testing.T) {
+	t.Parallel()
 
-	result := fs.Call(input)
-	if result != "b38699013d79e50d9d122433753959c1" {
-		t.Errorf("error executing function stack '%s': expected result "+
-			"'b38699013d79e50d9d122433753959c1' but got '%s' (%v)\n", input, result, fs)
+	type want struct {
+		arg string
+		// result of calling the returned function stack on the original input
+		callResult string
 	}
 
-	input = "maskdigits(md5sum(maskdigits($line)))"
-	fs, arg, err = NewFunctionStack(input)
-	if err != nil {
-		t.Errorf("error parsing function input '%s': %s (%v)\n", input, err.Error(), fs)
-	}
-	if arg != "$line" {
-		t.Errorf("error parsing function input '%s': expected argument '$line' but "+
-			"got '%s' (%v)\n", input, arg, fs)
+	cases := []struct {
+		input string
+		want  want
+	}{
+		{
+			input: "md5sum($line)",
+			want:  want{arg: "$line", callResult: "b38699013d79e50d9d122433753959c1"},
+		},
+		{
+			input: "maskdigits(md5sum(maskdigits($line)))",
+			want:  want{arg: "$line", callResult: ".fac.bbe..bb.........d...a.c..b."},
+		},
+		{
+			// An argument containing nested parens that are balanced is valid.
+			input: "md5sum($foo)",
+			want:  want{arg: "$foo"},
+		},
+		{
+			// Plain field with no function wrapper is a degenerate stack (empty).
+			input: "$line",
+			want:  want{arg: "$line"},
+		},
 	}
-	t.Log(input, fs, arg)
 
-	result = fs.Call(input)
-	if result != ".fac.bbe..bb.........d...a.c..b." {
-		t.Errorf("error executing function stack '%s': expected result "+
-			"'.fac.bbe..bb.........d...a.c..b.' but got '%s' (%v)\n", input, result, fs)
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.input, func(t *testing.T) {
+			t.Parallel()
+			fs, arg, err := NewFunctionStack(tc.input)
+			if err != nil {
+				t.Fatalf("unexpected error for input %q: %v (stack %v)", tc.input, err, fs)
+			}
+			if arg != tc.want.arg {
+				t.Errorf("arg: got %q, want %q", arg, tc.want.arg)
+			}
+			if tc.want.callResult != "" {
+				got := fs.Call(tc.input)
+				if got != tc.want.callResult {
+					t.Errorf("Call(%q) = %q, want %q", tc.input, got, tc.want.callResult)
+				}
+			}
+		})
 	}
+}
+
+// TestFunctionStackMalformed verifies that NewFunctionStack rejects expressions
+// that are structurally invalid. Before the fix, several of these were silently
+// accepted and produced wrong results.
+func TestFunctionStackMalformed(t *testing.T) {
+	t.Parallel()
 
-	input = "md5sum$line)"
-	if fs, _, err := NewFunctionStack(input); err == nil {
-		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n",
-			input, fs)
+	cases := []string{
+		// Missing opening paren — no function call syntax at all.
+		"md5sum$line)",
+		// Known outer function but inner call is missing its closing paren.
+		"md5sum(makedigits$line))",
+		// Stray ')' inside the argument after stripping: "bar)baz" remains.
+		// Before the fix this was silently accepted and produced wrong output.
+		"md5sum(bar)baz)",
+		// Input ends with '(' — no closing ')' so the loop never strips,
+		// but the argument string itself contains an unclosed '('.
+		// Before the fix this was accepted as a plain field literal.
+		"foo(",
+		// Empty outer call — the name portion is empty (index == 0) which
+		// is caught by the existing index <= 0 guard.
+		"()",
 	}
 
-	input = "md5sum(makedigits$line))"
-	if fs, _, err := NewFunctionStack(input); err == nil {
-		t.Errorf("Expected error parsing function input '%s' (%v) but got no error\n",
-			input, fs)
+	for _, input := range cases {
+		input := input