5 files changed, 31 insertions, 7 deletions
diff --git a/internal/io/fs/directprocessor.go b/internal/io/fs/directprocessor.go
index 9c564e7..84b78bb 100644
--- a/internal/io/fs/directprocessor.go
+++ b/internal/io/fs/directprocessor.go
@@ -59,8 +59,17 @@ func (dp *DirectProcessor) ProcessFile(ctx context.Context, filePath string) err
 
 // ProcessReader processes an io.Reader directly without channels
 func (dp *DirectProcessor) ProcessReader(ctx context.Context, reader io.Reader, filePath string) error {
-	// Check if we need to preserve line endings (for cat in plain mode)
+	// Check if we need to preserve line endings (for any processor in plain mode)
+	needsLineEndingPreservation := false
+	
 	if catProcessor, ok := dp.processor.(*CatProcessor); ok && catProcessor.plain {
+		needsLineEndingPreservation = true
+	} else if grepProcessor, ok := dp.processor.(*GrepProcessor); ok && grepProcessor.plain {
+		needsLineEndingPreservation = true
+	}
+	// Note: MapProcessor doesn't have a plain mode that requires line ending preservation
+	
+	if needsLineEndingPreservation {
 		return dp.processReaderPreservingLineEndings(ctx, reader, filePath)
 	}
 
diff --git a/internal/io/fs/grepprocessor.go b/internal/io/fs/grepprocessor.go
index ed1c271..c0db7b6 100644
--- a/internal/io/fs/grepprocessor.go
+++ b/internal/io/fs/grepprocessor.go
@@ -140,10 +140,20 @@ func (gp *GrepProcessor) Flush() []byte {
 func (gp *GrepProcessor) formatLine(line []byte, lineNum int, filePath string, stats *stats, sourceID string) []byte {
 	// Format output to match existing behavior
 	if gp.plain {
-		result := make([]byte, len(line)+1)
-		copy(result, line)
-		result[len(line)] = '\n'
-		return result
+		// If line already ends with a line ending, preserve it as-is
+		// Otherwise, add LF for consistency with bufio.Scanner behavior
+		if len(line) > 0 && (line[len(line)-1] == '\n' || (len(line) > 1 && line[len(line)-2] == '\r' && line[len(line)-1] == '\n')) {
+			// Line already has line ending, preserve it exactly
+			result := make([]byte, len(line))
+			copy(result, line)
+			return result
+		} else {
+			// Line doesn't have line ending, add LF
+			result := make([]byte, len(line)+1)
+			copy(result, line)
+			result[len(line)] = '\n'
+			return result
+		}
 	}
 
 	// Format exactly like original basehandler.go for non-plain mode
diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go
index f23c9e5..e8ce19a 100644
--- a/internal/server/handlers/basehandler.go
+++ b/internal/server/handlers/basehandler.go
@@ -262,6 +262,8 @@ func (h *baseHandler) handleOptions(options map[string]string) {
 		if plain, _ := options["plain"]; plain == "true" {
 			dlog.Server.Debug(h.user, "Enabling plain mode")
 			h.plain = true
+		} else {
+			dlog.Server.Debug(h.user, "Plain mode not enabled", "plain option:", plain)
 		}
 		if serverless, _ := options["serverless"]; serverless == "true" {
 			dlog.Server.Debug(h.user, "Enabling serverless mode")
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 14441b8..c75b9fc 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -254,12 +254,14 @@ func (r *readCommand) createProcessor(re regex.Regex, ltx lcontext.LContext, out
 	plain := r.server.plain       // Use actual plain mode from server
 	noColor := false              // Enable colors by default
 	
+	dlog.Server.Debug(r.server.user, "createProcessor: plain mode is", plain)
+	
 	// If there's an existing aggregate (from a 'map' command), we need to feed data to it
 	// Create a lines channel and connect it to the aggregate
 	if r.server.aggregate != nil {
 		dlog.Server.Debug("Using existing aggregate, creating bridge processor")
 		// Create a lines channel for the aggregate with larger buffer
-		linesCh := make(chan *line.Line, 1000)
+		linesCh := make(chan *line.Line, 10000)
 		// Connect the lines channel to the aggregate
 		go func() {
 			r.server.aggregate.NextLinesCh <- linesCh
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 5ef8a1d..e83c4cf 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -13,6 +13,7 @@ import (
 	user "github.com/mimecast/dtail/internal/user/server"
 )
 
+
 // ServerHandler implements the Reader and Writer interfaces to handle
 // the Bi-directional communication between SSH client and server.
 // This handler implements the handler of the SSH server.
@@ -31,7 +32,7 @@ func NewServerHandler(user *user.User, catLimiter,
 	h := ServerHandler{
 		baseHandler: baseHandler{
 			done:             internal.NewDone(),
-			lines:            make(chan *line.Line, 1000),
+			lines:            make(chan *line.Line, 10000),
 			serverMessages:   make(chan string, 10),
 			maprMessages:     make(chan string, 10),
 			ackCloseReceived: make(chan struct{}),