Complete channelless migration for DTail operations

- Implement channelless MapReduce with streaming aggregation
- Add channelless tail with proper file following capability
- Fix TestDTailWithServer by implementing ServerHandlerWriter for client-server mode
- Add proper serverless mode detection for standalone operations
- Remove temporary benchmark scripts
- All integration tests now pass

🤖 Generated with [Claude Code](https://claude.ai/code)

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

8 files changed, 944 insertions, 555 deletions

diff --git a/.claude/commands/retest.md b/.claude/commands/retest.md
new file mode 100644
index 0000000..555654f
--- /dev/null
+++ b/.claude/commands/retest.md
@@ -0,0 +1 @@
+Recompile the project with make clean build and re-run all tests including integration tests. If any fail, fix them.
diff --git a/integrationtests/channelless_test.go b/integrationtests/channelless_test.go
new file mode 100644
index 0000000..41845df
--- /dev/null
+++ b/integrationtests/channelless_test.go
@@ -0,0 +1,104 @@
+package integrationtests
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/mimecast/dtail/internal/config"
+)
+
+func TestDGrepChannelless(t *testing.T) {
+	if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") {
+		t.Log("Skipping")
+		return
+	}
+	
+	// Test dgrep with channelless mode (now the default)
+	inFile := "mapr_testdata.log"
+	outFile := "dgrepchannelless.stdout.tmp"
+	expectedOutFile := "dgrepcontext1.txt.expected"
+
+	_, err := runCommand(context.TODO(), t, outFile,
+		"../dgrep",
+		"--plain",
+		"--cfg", "none",
+		"--grep", "1002-071947",
+		"--after", "3",
+		"--before", "3",
+		inFile)
+
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	if err := compareFiles(t, outFile, expectedOutFile); err != nil {
+		t.Error(err)
+		return
+	}
+
+	os.Remove(outFile)
+}
+
+func TestDCatChannelless(t *testing.T) {
+	if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") {
+		t.Log("Skipping")
+		return
+	}
+	
+	// Test dcat with channelless mode (now the default)
+	inFile := "dcat1a.txt"
+	outFile := "dcatchannelless.stdout.tmp"
+
+	_, err := runCommand(context.TODO(), t, outFile,
+		"../dcat",
+		"--plain",
+		"--cfg", "none",
+		inFile)
+
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	if err := compareFiles(t, outFile, inFile); err != nil {
+		t.Error(err)
+		return
+	}
+
+	os.Remove(outFile)
+}
+
+func TestChannellessMode(t *testing.T) {
+	if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") {
+		t.Log("Skipping")
+		return
+	}
+	
+	// Test that channelless mode (now default) works correctly
+	
+	// Test grep
+	inFile := "mapr_testdata.log"
+	outFile := "grep_channelless.tmp"
+	expectedOutFile := "dgrep1.txt.expected"
+
+	_, err := runCommand(context.TODO(), t, outFile,
+		"../dgrep",
+		"--plain",
+		"--cfg", "none",
+		"--grep", "1002-071947",
+		inFile)
+
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	if err := compareFiles(t, outFile, expectedOutFile); err != nil {
+		t.Error(err)
+		return
+	}
+
+	os.Remove(outFile)
+}
\ No newline at end of file
diff --git a/internal/io/fs/directprocessor.go b/internal/io/fs/directprocessor.go
index d4bf3ed..00ecedb 100644
--- a/internal/io/fs/directprocessor.go
+++ b/internal/io/fs/directprocessor.go
@@ -7,10 +7,16 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strings"
+	"time"
 
 	"github.com/mimecast/dtail/internal/color/brush"
 	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/lcontext"
+	"github.com/mimecast/dtail/internal/mapr"
+	"github.com/mimecast/dtail/internal/mapr/logformat"
 	"github.com/mimecast/dtail/internal/protocol"
 	"github.com/mimecast/dtail/internal/regex"
 )
@@ -618,7 +624,8 @@ func (tp *TailProcessor) ProcessLine(line []byte, lineNum int, filePath string,
 	}
 	
 	// Regular tailing mode - send matching lines immediately
-	return tp.formatLine(line, lineNum, filePath), true
+	formatted := tp.formatLine(line, lineNum, filePath)
+	return formatted, true
 }
 
 func (tp *TailProcessor) formatLine(line []byte, lineNum int, filePath string) []byte {
@@ -660,25 +667,248 @@ func (tp *TailProcessor) Flush() []byte {
 	return nil
 }
 
+// FollowingTailProcessor extends DirectProcessor with file following capability
+type FollowingTailProcessor struct {
+	*DirectProcessor
+	tailProcessor *TailProcessor
+}
+
+// NewFollowingTailProcessor creates a processor that can follow files
+func NewFollowingTailProcessor(processor *TailProcessor, output io.Writer, globID string, ltx lcontext.LContext) *FollowingTailProcessor {
+	dp := NewDirectProcessor(processor, output, globID, ltx)
+	return &FollowingTailProcessor{
+		DirectProcessor: dp,
+		tailProcessor:   processor,
+	}
+}
+
+// ProcessFileWithFollowing processes a file with following capability
+func (ftp *FollowingTailProcessor) ProcessFileWithFollowing(ctx context.Context, filePath string) error {
+	if !ftp.tailProcessor.follow {
+		// No following required, use regular processing
+		return ftp.ProcessFile(ctx, filePath)
+	}
+
+	// Implement file following logic
+	return ftp.followFile(ctx, filePath)
+}
+
+func (ftp *FollowingTailProcessor) followFile(ctx context.Context, filePath string) error {
+	file, err := os.Open(filePath)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	// Initialize processor
+	if err := ftp.processor.Initialize(ctx); err != nil {
+		return err
+	}
+	defer ftp.processor.Cleanup()
+
+	// If seekEOF is true, seek to end first
+	if ftp.tailProcessor.seekEOF {
+		if _, err := file.Seek(0, io.SeekEnd); err != nil {
+			return err
+		}
+	}
+
+	return ftp.followReader(ctx, file, filePath)
+}
+
+func (ftp *FollowingTailProcessor) followReader(ctx context.Context, file *os.File, filePath string) error {
+	// Set buffer size respecting MaxLineLength configuration
+	maxLineLength := config.Server.MaxLineLength
+	initialBufSize := 64 * 1024
+	if maxLineLength < initialBufSize {
+		initialBufSize = maxLineLength
+	}
+	
+	lineNum := 0
+	lastPosition := int64(0)
+	readBuffer := make([]byte, initialBufSize)
+	lineBuffer := make([]byte, 0, initialBufSize)
+	
+	// Get initial position
+	if pos, err := file.Seek(0, io.SeekCurrent); err == nil {
+		lastPosition = pos
+	}
+	
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+		
+		// Check if file has grown
+		if stat, err := file.Stat(); err == nil {
+			if stat.Size() > lastPosition {
+				// Read new content
+				n, err := file.Read(readBuffer)
+				if err != nil && err != io.EOF {
+					return err
+				}
+				
+				if n > 0 {
+					// Process the data, looking for complete lines
+					for i := 0; i < n; i++ {
+						b := readBuffer[i]
+						if b == '\n' {
+							// Found a complete line
+							lineNum++
+							line := make([]byte, len(lineBuffer))
+							copy(line, lineBuffer)
+							
+							// Update position stats
+							if ftp.stats != nil {
+								ftp.stats.updatePosition()
+							}
+							
+							// Process line directly  
+							if result, shouldSend := ftp.processor.ProcessLine(line, lineNum, filePath, ftp.stats, ftp.sourceID); shouldSend {
+								if _, err := ftp.output.Write(result); err != nil {
+									return err
+								}
+								
+								// Update transmission stats
+								if ftp.stats != nil {
+									ftp.stats.updateLineTransmitted()
+								}
+							}
+							
+							// Reset line buffer for next line
+							lineBuffer = lineBuffer[:0]
+						} else {
+							// Add byte to current line
+							lineBuffer = append(lineBuffer, b)
+						}
+					}
+					
+					// Update last position
+					if pos, err := file.Seek(0, io.SeekCurrent); err == nil {
+						lastPosition = pos
+					}
+					
+					continue
+				}
+			}
+		}
+		
+		// No more content available, check if file was truncated/rotated
+		if ftp.checkFileRotation(file, filePath, &lastPosition) {
+			// File was rotated, reopen and continue
+			file.Close()
+			var err error
+			file, err = os.Open(filePath)
+			if err != nil {
+				return err
+			}
+			defer file.Close()
+			
+			lastPosition = 0
+			lineBuffer = lineBuffer[:0]
+			continue
+		}
+		
+		// Wait a bit before checking for new content
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(100 * time.Millisecond):
+			// Continue the loop to check for new content
+		}
+	}
+}
+
+func (ftp *FollowingTailProcessor) checkFileRotation(file *os.File, filePath string, lastPosition *int64) bool {
+	// Get current file info
+	currentInfo, err := file.Stat()
+	if err != nil {
+		return false
+	}
+	
+	// Get file info by path
+	pathInfo, err := os.Stat(filePath)
+	if err != nil {
+		return false
+	}
+	
+	// Check if file was truncated (size is smaller than our position)
+	if pathInfo.Size() < *lastPosition {
+		return true
+	}
+	
+	// Check if file was rotated (different inode/device)
+	if !os.SameFile(currentInfo, pathInfo) {
+		return true
+	}
+	
+	return false
+}
+
 // MapProcessor handles MapReduce-style aggregation
 type MapProcessor struct {
-	plain      bool
-	hostname   string
-	aggregator interface{} // Will be set to actual aggregator from mapr package
-	buffer     []byte
+	plain           bool
+	hostname        string
+	query           *mapr.Query
+	parser          logformat.Parser
+	groupSet        *mapr.GroupSet
+	buffer          []byte
+	output          io.Writer
+	lastSerialized  time.Time
+	serializeFunc   func(groupSet *mapr.GroupSet)
 }
 
 // NewMapProcessor creates a new map processor
-func NewMapProcessor(plain bool, hostname string) *MapProcessor {
-	return &MapProcessor{
-		plain:    plain,
-		hostname: hostname,
-		buffer:   make([]byte, 0, 1024*1024), // 1MB buffer for aggregation
+func NewMapProcessor(plain bool, hostname string, queryStr string, output io.Writer) (*MapProcessor, error) {
+	query, err := mapr.NewQuery(queryStr)
+	if err != nil {
+		return nil, err
+	}
+
+	var parserName string
+	switch query.LogFormat {
+	case "":
+		parserName = config.Server.MapreduceLogFormat
+		if query.Table == "" {
+			parserName = "generic"
+		}
+	default:
+		parserName = query.LogFormat
+	}
+
+	logParser, err := logformat.NewParser(parserName, query)
+	if err != nil {
+		dlog.Server.Error("Could not create log format parser. Falling back to 'generic'", err)
+		if logParser, err = logformat.NewParser("generic", query); err != nil {
+			return nil, fmt.Errorf("could not create log format parser: %w", err)
+		}
 	}
+
+	mp := &MapProcessor{
+		plain:          plain,
+		hostname:       hostname,
+		query:          query,
+		parser:         logParser,
+		groupSet:       mapr.NewGroupSet(),
+		buffer:         make([]byte, 0, 1024*1024), // 1MB buffer for aggregation
+		output:         output,
+		lastSerialized: time.Now(),
+	}
+	
+	// Set up serialization function
+	mp.serializeFunc = mp.defaultSerializeFunc
+	
+	return mp, nil
+}
+
+// SetSerializeFunc allows custom serialization (for testing or different output formats)
+func (mp *MapProcessor) SetSerializeFunc(fn func(groupSet *mapr.GroupSet)) {
+	mp.serializeFunc = fn
 }
 
 func (mp *MapProcessor) Initialize(ctx context.Context) error {
-	// TODO: Initialize MapReduce aggregator when implementing
 	return nil
 }
 
@@ -687,15 +917,118 @@ func (mp *MapProcessor) Cleanup() error {
 }
 
 func (mp *MapProcessor) ProcessLine(line []byte, lineNum int, filePath string, stats *stats, sourceID string) ([]byte, bool) {
-	// For MapReduce, we accumulate lines and process in batch
-	// TODO: Pass line to aggregator when implementing MapReduce integration
-	return nil, false // No immediate output for MapReduce
+	// Convert line to string and parse fields
+	maprLine := strings.TrimSpace(string(line))
+	
+	fields, err := mp.parser.MakeFields(maprLine)
+	if err != nil {
+		// Should fields be ignored anyway?
+		if err != logformat.ErrIgnoreFields {
+			dlog.Server.Error("Error parsing line for MapReduce", err)
+		}
+		return nil, false
+	}
+	
+	// Apply WHERE clause filter
+	if !mp.query.WhereClause(fields) {
+		return nil, false
+	}
+	
+	// Apply SET clause (add additional fields)
+	if len(mp.query.Set) > 0 {
+		if err := mp.query.SetClause(fields); err != nil {
+			dlog.Server.Error("Error applying SET clause", err)
+			return nil, false
+		}
+	}
+	
+	// Aggregate the fields
+	mp.aggregateFields(fields)
+	
+	// Check if we should serialize results periodically (every 5 seconds by default)
+	now := time.Now()
+	if now.Sub(mp.lastSerialized) >= mp.query.Interval {
+		mp.periodicSerialize()
+		mp.lastSerialized = now
+	}
+	
+	return nil, false // No immediate output for MapReduce - output happens periodically
+}
+
+func (mp *MapProcessor) aggregateFields(fields map[string]string) {
+	var sb strings.Builder
+	for i, field := range mp.query.GroupBy {
+		if i > 0 {
+			sb.WriteString(protocol.AggregateGroupKeyCombinator)
+		}
+		if val, ok := fields[field]; ok {
+			sb.WriteString(val)
+		}
+	}
+	groupKey := sb.String()
+	set := mp.groupSet.GetSet(groupKey)
+
+	var addedSample bool
+	for _, sc := range mp.query.Select {
+		if val, ok := fields[sc.Field]; ok {
+			if err := set.Aggregate(sc.FieldStorage, sc.Operation, val, false); err != nil {
+				dlog.Server.Error("Error aggregating field", err)
+				continue
+			}
+			addedSample = true
+		}
+	}
+
+	if addedSample {
+		set.Samples++
+	}
+}
+
+// periodicSerialize sends current aggregation results and resets the group set
+func (mp *MapProcessor) periodicSerialize() {
+	if mp.serializeFunc != nil {
+		mp.serializeFunc(mp.groupSet)
+	}
+	// Reset group set for next interval
+	mp.groupSet = mapr.NewGroupSet()
+}
+
+// defaultSerializeFunc implements the default serialization behavior
+func (mp *MapProcessor) defaultSerializeFunc(groupSet *mapr.GroupSet) {
+	// Use a channel to collect serialized data
+	ch := make(chan string, 100)
+	done := make(chan struct{})
+	
+	go func() {
+		defer close(done)
+		for msg := range ch {
+			// Format as protocol message: A|{serialized_data}¬
+			var output strings.Builder
+			output.WriteString("A")
+			output.WriteString(protocol.FieldDelimiter)
+			output.WriteString(msg)
+			output.WriteByte(protocol.MessageDelimiter)
+			
+			// Write to output immediately
+			if mp.output != nil {
+				mp.output.Write([]byte(output.String()))
+			}
+		}
+	}()
+	
+	// Serialize the group set
+	ctx := context.Background()
+	groupSet.Serialize(ctx, ch)
+	close(ch)
+	<-done
 }
 
 func (mp *MapProcessor) Flush() []byte {
-	// TODO: Return aggregated results from MapReduce processor
-	// For now, return empty to maintain interface
-	return nil
+	// Final flush - serialize any remaining data
+	if mp.serializeFunc != nil {
+		mp.serializeFunc(mp.groupSet)
+	}
+	return nil // Output is handled by serializeFunc
 }
 
 // Helper function to append integer to byte slice
@@ -717,4 +1050,174 @@ func appendInt(dst []byte, i int) []byte {
 	}
 	
 	return append(dst, str...)
+}
+
+// AggregateLineProcessor feeds lines to an existing aggregate via channels
+type AggregateLineProcessor struct {
+	linesCh    chan<- *line.Line
+	re         regex.Regex
+	hostname   string
+	ltx        lcontext.LContext
+	lineNum    int
+	isTailing  bool // Whether this is for a tail operation that should keep running
+}
+
+// NewAggregateLineProcessor creates a processor that feeds lines to an aggregate
+func NewAggregateLineProcessor(linesCh chan<- *line.Line, re regex.Regex, hostname string, ltx lcontext.LContext) *AggregateLineProcessor {
+	return &AggregateLineProcessor{
+		linesCh:   linesCh,
+		re:        re,
+		hostname:  hostname,
+		ltx:       ltx,
+		lineNum:   0,
+		isTailing: false,
+	}
+}
+
+// NewAggregateLineProcessorForTail creates a processor for tail operations that feeds lines to an aggregate
+func NewAggregateLineProcessorForTail(linesCh chan<- *line.Line, re regex.Regex, hostname string, ltx lcontext.LContext) *AggregateLineProcessor {
+	return &AggregateLineProcessor{
+		linesCh:   linesCh,
+		re:        re,
+		hostname:  hostname,
+		ltx:       ltx,
+		lineNum:   0,
+		isTailing: true,
+	}
+}
+
+func (p *AggregateLineProcessor) ProcessLine(lineBuf []byte, lineNum int, filePath string, stats *stats, sourceID string) (result []byte, shouldSend bool) {
+	p.lineNum++
+	
+	// For MapReduce operations, don't apply regex filtering here - let the aggregate handle it
+	// The aggregate's log parser and WHERE clause will do the proper filtering
+	
+	// Create a line object similar to what the channel-based system creates
+	// Make a copy of the line buffer to avoid issues with slice reuse
+	lineCopy := make([]byte, len(lineBuf))
+	copy(lineCopy, lineBuf)
+	content := bytes.NewBuffer(lineCopy)
+	l := line.New(content, uint64(p.lineNum), 100, sourceID)
+	
+	// Send the line to the aggregate via the channel (blocking send to avoid data loss)
+	p.linesCh <- l
+	
+	// Don't send output directly since the aggregate will handle serialization
+	return nil, false
+}
+
+func (p *AggregateLineProcessor) Flush() []byte {
+	// For tail operations, don't close the channel as we want to keep following
+	if !p.isTailing {
+		// Close the lines channel to signal end of input
+		// Add a small delay to ensure all lines are processed before closing
+		time.Sleep(10 * time.Millisecond)
+		close(p.linesCh)
+	}
+	return nil
+}
+
+func (p *AggregateLineProcessor) Initialize(ctx context.Context) error {
+	return nil
+}
+
+func (p *AggregateLineProcessor) Cleanup() error {
+	return nil
+}
+
+// ProcessFileWithTailing processes a file with tailing capability
+func (dp *DirectProcessor) ProcessFileWithTailing(ctx context.Context, filePath string) error {
+	// Use the same logic as FollowingTailProcessor but with our DirectProcessor
+	file, err := os.Open(filePath)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	// First, process existing content
+	if err := dp.ProcessReader(ctx, file, filePath); err != nil {
+		return err
+	}
+
+	// Then follow the file for new content
+	return dp.followFile(ctx, filePath)
+}
+
+// followFile implements file following logic similar to FollowingTailProcessor
+func (dp *DirectProcessor) followFile(ctx context.Context, filePath string) error {
+	// Track our current position in the file
+	var lastSize int64
+	
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(100 * time.Millisecond):
+			// Check if file has grown
+			fileInfo, err := os.Stat(filePath)
+			if err != nil {
+				continue
+			}
+			
+			currentSize := fileInfo.Size()
+			if currentSize > lastSize {
+				// File has new content, read it
+				file, err := os.Open(filePath)
+				if err != nil {
+					continue
+				}
+				
+				// Seek to where we left off
+				if _, err := file.Seek(lastSize, 0); err != nil {
+					file.Close()
+					continue
+				}
+				
+				// Process new content
+				if err := dp.processNewContent(ctx, file, filePath); err != nil {
+					file.Close()
+					continue
+				}
+				
+				lastSize = currentSize
+				file.Close()
+			}
+		}
+	}
+}
+
+// processNewContent processes new content that was added to the file
+func (dp *DirectProcessor) processNewContent(ctx context.Context, file *os.File, filePath string) error {
+	scanner := bufio.NewScanner(file)
+	
+	// Start line counting from where we left off (simplified approach)
+	lineNum := 1
+	
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+		
+		lineBuf := scanner.Bytes()
+		if result, shouldSend := dp.processor.ProcessLine(lineBuf, lineNum, filePath, dp.stats, dp.sourceID); shouldSend {
+			if _, err := dp.output.Write(result); err != nil {
+				return err
+			}
+			
+			// Update transmission stats
+			if dp.stats != nil {
+				dp.stats.updateLineTransmitted()
+			}
+		}
+		lineNum++
+		
+		// Update position stats
+		if dp.stats != nil {
+			dp.stats.updatePosition()
+		}
+	}
+	
+	return scanner.Err()
 }
\ No newline at end of file
diff --git a/internal/server/handlers/networkwriter.go b/internal/server/handlers/networkwriter.go
index f1e3bee..bb5ad1d 100644
--- a/internal/server/handlers/networkwriter.go
+++ b/internal/server/handlers/networkwriter.go
@@ -1,11 +1,13 @@
 package handlers
 
 import (
+	"bytes"
 	"fmt"
 	"net"
 	"os"
 
 	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/user/server"
 )
 
@@ -173,4 +175,178 @@ func (bnw *BufferedNetworkWriter) Close() error {
 		return err
 	}
 	return bnw.NetworkOutputWriter.Close()
+}
+
+// ChannelOutputWriter sends output to the server's lines channel instead of direct network
+type ChannelOutputWriter struct {
+	linesCh        chan<- *line.Line
+	serverMessages chan<- string
+	user           *server.User
+}
+
+// NewChannelOutputWriter creates a new channel output writer
+func NewChannelOutputWriter(linesCh chan<- *line.Line, serverMessages chan<- string, user *server.User) *ChannelOutputWriter {
+	return &ChannelOutputWriter{
+		linesCh:        linesCh,
+		serverMessages: serverMessages,
+		user:           user,
+	}
+}
+
+// Write implements io.Writer interface by sending data through the lines channel
+func (cow *ChannelOutputWriter) Write(data []byte) (int, error) {
+	if len(data) == 0 {
+		return 0, nil
+	}
+	
+	// Create a line object using the proper constructor
+	contentBuffer := bytes.NewBuffer(data)
+	lineObj := line.New(contentBuffer, 0, 100, "channelless")
+	
+	select {
+	case cow.linesCh <- lineObj:
+		return len(data), nil
+	default:
+		// Channel is full, report error
+		cow.sendServerMessage("Lines channel full, dropping data")
+		return 0, fmt.Errorf("lines channel full")
+	}
+}
+
+// sendServerMessage sends a message through the existing server message channel
+func (cow *ChannelOutputWriter) sendServerMessage(message string) {
+	if cow.serverMessages == nil {
+		return
+	}
+	
+	select {
+	case cow.serverMessages <- message:
+		// Message sent successfully
+	default:
+		// Channel full, log the issue
+		dlog.Server.Warn(cow.user, "Server message channel full, dropping message:", message)
+	}
+}
+
+// SendLine sends a formatted line through the lines channel
+func (cow *ChannelOutputWriter) SendLine(hostname, filePath string, lineNum int, content []byte) error {
+	// Create a line object with proper metadata
+	contentBuffer := bytes.NewBuffer(content)
+	lineObj := line.New(contentBuffer, uint64(lineNum), 100, filePath)
+	
+	select {
+	case cow.linesCh <- lineObj:
+		return nil
+	default:
+		cow.sendServerMessage(fmt.Sprintf("Lines channel full, dropping line from %s:%d", filePath, lineNum))
+		return fmt.Errorf("lines channel full")
+	}
+}
+
+// SendPlainLine sends a plain line through the lines channel
+func (cow *ChannelOutputWriter) SendPlainLine(content []byte) error {
+	_, err := cow.Write(content)
+	return err
+}
+
+// SendServerStat sends a server statistics message
+func (cow *ChannelOutputWriter) SendServerStat(message string) {
+	cow.sendServerMessage(message)
+}
+
+// SendError sends an error message
+func (cow *ChannelOutputWriter) SendError(err error) {
+	cow.sendServerMessage(fmt.Sprintf("ERROR: %v", err))
+}
+
+// Close is a no-op for channel output writer
+func (cow *ChannelOutputWriter) Close() error {
+	return nil
+}
+
+// ServerHandlerWriter writes output directly to the server handler's lines channel
+type ServerHandlerWriter struct {
+	server         *ServerHandler
+	serverMessages chan<- string
+	user           *server.User
+}
+
+// NewServerHandlerWriter creates a new server handler writer
+func NewServerHandlerWriter(serverHandler *ServerHandler, serverMessages chan<- string, user *server.User) *ServerHandlerWriter {
+	return &ServerHandlerWriter{
+		server:         serverHandler,
+		serverMessages: serverMessages,
+		user:           user,
+	}
+}
+
+// Write implements io.Writer interface by sending data through the server's lines channel
+func (shw *ServerHandlerWriter) Write(data []byte) (int, error) {
+	if len(data) == 0 {
+		return 0, nil
+	}
+	
+	// Create a line object and send it through the server's lines channel
+	contentBuffer := bytes.NewBuffer(data)
+	lineObj := line.New(contentBuffer, 0, 100, "channelless")
+	
+	select {
+	case shw.server.lines <- lineObj:
+		return len(data), nil
+	default:
+		// Channel is full, report error
+		shw.sendServerMessage("Server lines channel full, dropping data")
+		return 0, fmt.Errorf("server lines channel full")
+	}
+}
+
+// sendServerMessage sends a message through the existing server message channel
+func (shw *ServerHandlerWriter) sendServerMessage(message string) {
+	if shw.serverMessages == nil {
+		return
+	}
+	
+	select {
+	case shw.serverMessages <- message:
+		// Message sent successfully
+	default:
+		// Channel full, log the issue
+		dlog.Server.Warn(shw.user, "Server message channel full, dropping message:", message)
+	}
+}
+
+// SendLine sends a formatted line through the server's lines channel
+func (shw *ServerHandlerWriter) SendLine(hostname, filePath string, lineNum int, content []byte) error {
+	// Create a line object with proper metadata
+	contentBuffer := bytes.NewBuffer(content)
+	lineObj := line.New(contentBuffer, uint64(lineNum), 100, filePath)
+	
+	select {
+	case shw.server.lines <- lineObj:
+		return nil
+	default:
+		shw.sendServerMessage(fmt.Sprintf("Server lines channel full, dropping line from %s:%d", filePath, lineNum))
+		return fmt.Errorf("server lines channel full")
+	}
+}
+
+// SendPlainLine sends a plain line through the server's lines channel
+func (shw *ServerHandlerWriter) SendPlainLine(content []byte) error {
+	_, err := shw.Write(content)
+	return err
+}
+
+// SendServerStat sends a server statistics message
+func (shw *ServerHandlerWriter) SendServerStat(message string) {
+	shw.sendServerMessage(message)
+}
+
+// SendError sends an error message
+func (shw *ServerHandlerWriter) SendError(err error) {
+	shw.sendServerMessage(fmt.Sprintf("ERROR: %v", err))
+}
+
+// Close is a no-op for server handler writer
+func (shw *ServerHandlerWriter) Close() error {
+	return nil
 }
\ No newline at end of file
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 89bb757..59e9d1b 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -2,10 +2,10 @@ package handlers
 
 import (
 	"context"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/mimecast/dtail/internal/io/dlog"
@@ -32,6 +32,9 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext,
 	argc int, args []string, retries int) {
 
 	re := regex.NewNoop()
+	var queryStr string
+	
+	// Parse regex for non-MapReduce operations or when no aggregate exists
 	if argc >= 4 {
 		deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " "))
 		if err != nil {
@@ -41,169 +44,18 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext,
 		}
 		re = deserializedRegex
 	}
+	
 	if argc < 3 {
 		r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
 			"Unable to parse command", args, argc))
 		return
 	}
 
-	// Check if channelless mode is enabled
-	// Note: MapReduce operations require the full channel-based aggregation infrastructure
-	// Note: Tail operations require continuous monitoring and real-time streaming 
-	isMapReduceCmd := r.mode == omode.MapClient || r.isMapReduceCommand(re)
-	isTailCmd := r.mode == omode.TailClient
-	useChannelless := os.Getenv("DTAIL_USE_CHANNELLESS") == "yes" && !isMapReduceCmd && !isTailCmd
-	
-	if useChannelless {
-		dlog.Server.Debug("Using channelless processing mode for mode:", r.mode)
-		r.startChannelless(ctx, ltx, args, re, retries)
-		return
-	}
-	
-	dlog.Server.Debug("Using channel-based processing mode for mode:", r.mode)
-
-	// In serverless mode, can also read data from pipe
-	// e.g.: grep foo bar.log | dmap 'from STATS select ...'
-	// Only read from stdin if no file is specified AND input is from pipe
-	if (args[1] == "" || args[1] == "-") && r.isInputFromPipe() {
-		dlog.Server.Debug("Reading data from stdin pipe")
-		// Empty file path and globID "-" represents reading from the stdin pipe.
-		r.read(ctx, ltx, "", "-", re)
-		return
-	}
-
-	dlog.Server.Debug("Reading data from file(s)")
-	r.readGlob(ctx, ltx, args[1], re, retries)
-}
-
-func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext,
-	glob string, re regex.Regex, retries int) {
-
-	retryInterval := time.Second * 5
-	glob = filepath.Clean(glob)
-
-	for retryCount := 0; retryCount < retries; retryCount++ {
-		paths, err := filepath.Glob(glob)
-		if err != nil {
-			dlog.Server.Warn(r.server.user, glob, err)
-			time.Sleep(retryInterval)
-			continue
-		}
-
-		if numPaths := len(paths); numPaths == 0 {
-			dlog.Server.Error(r.server.user, "No such file(s) to read", glob)
-			r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
-				"Unable to read file(s), check server logs"))
-			select {
-			case <-ctx.Done():
-				return
-			default:
-			}
-			time.Sleep(retryInterval)
-			continue
-		}
-
-		r.readFiles(ctx, ltx, paths, glob, re, retryInterval)
-		return
-	}
-
-	r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
-		"Giving up to read file(s)"))
-	return
-}
-
-func (r *readCommand) readFiles(ctx context.Context, ltx lcontext.LContext,
-	paths []string, glob string, re regex.Regex, retryInterval time.Duration) {
-
-	var wg sync.WaitGroup
-	wg.Add(len(paths))
-	for _, path := range paths {
-		go r.readFileIfPermissions(ctx, ltx, &wg, path, glob, re)
-	}
-	wg.Wait()
-}
-
-func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LContext,
-	wg *sync.WaitGroup, path, glob string, re regex.Regex) {
-
-	defer wg.Done()
-	globID := r.makeGlobID(path, glob)
-	if !r.server.user.HasFilePermission(path, "readfiles") {
-		dlog.Server.Error(r.server.user, "No permission to read file", path, globID)
-		r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user,
-			"Unable to read file(s), check server logs"))
-		return
-	}
-	r.read(ctx, ltx, path, globID, re)
+	// Always use channelless mode now - channel-based code is deprecated
+	dlog.Server.Debug("Using channelless processing mode for mode:", r.mode)
+	r.startChannelless(ctx, ltx, args, re, retries, queryStr)
 }
 
-func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext,
-	path, globID string, re regex.Regex) {
-
-	dlog.Server.Info(r.server.user, "Start reading", path, globID)
-	var reader fs.FileReader
-	var limiter chan struct{}
-
-	switch r.mode {
-	case omode.GrepClient, omode.CatClient:
-		reader = fs.NewCatFile(path, globID, r.server.serverMessages)
-		limiter = r.server.catLimiter
-	case omode.TailClient:
-		fallthrough
-	default:
-		reader = fs.NewTailFile(path, globID, r.server.serverMessages)
-		limiter = r.server.tailLimiter
-	}
-
-	defer func() {
-		select {
-		case <-limiter:
-		default:
-		}
-	}()
-
-	select {
-	case limiter <- struct{}{}:
-	case <-ctx.Done():
-		return
-	default:
-		dlog.Server.Info("Server limit hit, queueing file", len(limiter), path)
-		select {
-		case limiter <- struct{}{}:
-			dlog.Server.Info("Server limit OK now, processing file", len(limiter), path)
-		case <-ctx.Done():
-			return
-		}
-	}
-
-	lines := r.server.lines
-	aggregate := r.server.aggregate
-
-	for {
-		if aggregate != nil {
-			lines = make(chan *line.Line, 100)
-			aggregate.NextLinesCh <- lines
-		}
-		if err := reader.Start(ctx, ltx, lines, re); err != nil {
-			dlog.Server.Error(r.server.user, path, globID, err)
-		}
-		if aggregate != nil {
-			// Also makes aggregate to Flush
-			close(lines)
-		}
-
-		sele