Interactive serial module

cmds/dutctl/rpc.go

@@ -5,7 +5,6 @@
 package main
 
 import (
-	"bufio"
 	"context"
 	"errors"
 	"fmt"
@@ -17,6 +16,7 @@ import (
 
 	"connectrpc.com/connect"
 	"github.com/BlindspotSoftware/dutctl/internal/output"
+	"golang.org/x/sys/unix"
 
 	pb "github.com/BlindspotSoftware/dutctl/protobuf/gen/dutctl/v1"
 )
@@ -92,10 +92,45 @@ func (app *application) detailsRPC(device, command, keyword string) error {
 	return nil
 }
 
+// setRawInput puts the terminal into raw input mode: disables local echo and
+// canonical (line-buffered) mode so each keystroke is available immediately.
+// It returns a restore function, or nil if the fd is not a terminal.
+func setRawInput(fileDescriptor int) func() {
+	termios, err := unix.IoctlGetTermios(fileDescriptor, unix.TCGETS)
+	if err != nil {
+		return nil
+	}
+
+	old := *termios
+
+	termios.Iflag &^= unix.ICRNL
+	termios.Lflag &^= unix.ECHO | unix.ICANON
+	termios.Cc[unix.VMIN] = 1
+	termios.Cc[unix.VTIME] = 0
+
+	err = unix.IoctlSetTermios(fileDescriptor, unix.TCSETS, termios)
+	if err != nil {
+		return nil
+	}
+
+	return func() {
+		_ = unix.IoctlSetTermios(fileDescriptor, unix.TCSETS, &old)
+	}
+}
+
 //nolint:funlen,cyclop,gocognit
 func (app *application) runRPC(device, command string, cmdArgs []string) error {
 	const numWorkers = 2 // The send and receive worker goroutines
 
+	// Set raw input mode: disable local echo and canonical mode so each
+	// keystroke is sent immediately. Interactive modules like serial rely
+	// on the remote side to echo input.
+	if f, ok := app.stdin.(*os.File); ok {
+		if restore := setRawInput(int(f.Fd())); restore != nil {
+			defer restore()
+		}
+	}
+
 	runCtx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -230,7 +265,9 @@ func (app *application) runRPC(device, command string, cmdArgs []string) error {
 	go func() {
 		defer cancel()
 
-		reader := bufio.NewReader(app.stdin)
+		const stdinBufSize = 256
+
+		buf := make([]byte, stdinBufSize)
 
 		for {
 			select {
@@ -241,7 +278,7 @@ func (app *application) runRPC(device, command string, cmdArgs []string) error {
 			default:
 			}
 
-			text, err := reader.ReadString('\n')
+			nRead, err := app.stdin.Read(buf)
 			if err != nil {
 				if !errors.Is(err, io.EOF) {
 					errChan <- fmt.Errorf("reading stdin: %w", err)
@@ -254,7 +291,7 @@ func (app *application) runRPC(device, command string, cmdArgs []string) error {
 				Msg: &pb.RunRequest_Console{
 					Console: &pb.Console{
 						Data: &pb.Console_Stdin{
-							Stdin: []byte(text),
+							Stdin: buf[:nRead],
 						},
 					},
 				},

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07
 	golang.org/x/crypto v0.48.0
 	golang.org/x/net v0.50.0
+	golang.org/x/sys v0.41.0
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -32,6 +33,5 @@ require (
 	github.com/olekukonko/tablewriter v1.0.9 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/rogpeppe/go-internal v1.6.1 // indirect
-	golang.org/x/sys v0.41.0 // indirect
 	golang.org/x/text v0.34.0 // indirect
 )

internal/dutagent/broker.go

@@ -45,8 +45,8 @@
 // for module execution and a channel signaling worker termination or errors.
 // Multiple calls are idempotent; subsequent calls return the already initialized session and channel.
 //
-//nolint:ireturn // returning interface module.Session is intentional for abstraction boundary
+
 func (b *Broker) Start(ctx context.Context, s Stream) (module.Session, <-chan error) {
 	b.once.Do(func() {
 		b.init()
 		b.stream = s

internal/dutagent/broker_test.go

@@ -132,7 +132,8 @@ func TestBroker_StdinForwarding(t *testing.T) {
 	b := &Broker{}
 	stdinPayload := []byte("user input")
 	req := &pb.RunRequest{Msg: &pb.RunRequest_Console{Console: &pb.Console{Data: &pb.Console_Stdin{Stdin: stdinPayload}}}}
-	stream := &testStream{recvReqs: []*pb.RunRequest{req}, recvErrs: []error{nil}} // after first req, EOF
+	// No recvErrs: testStream returns reqs in order then EOF by default.
+	stream := &testStream{recvReqs: []*pb.RunRequest{req}}
 	ctx, cancel := context.WithCancel(context.Background())
 	sess, errCh := b.Start(ctx, stream)
 
@@ -144,8 +145,7 @@ func TestBroker_StdinForwarding(t *testing.T) {
 			t.Fatalf("stdin mismatch: got %q want %q", string(data), string(stdinPayload))
 		}
 	case <-time.After(200 * time.Millisecond):
-		// Expected to fail until refactor ensures proper sequencing / closure.
-		// (Current code may work but channel closure semantics will still fail later.)
+		t.Fatal("timeout: stdin payload was not forwarded to stdinCh")
 	}
 
 	cancel() // simulate module completion

internal/dutagent/worker.go

@@ -100,6 +100,10 @@ func toClientWorker(ctx context.Context, stream Stream, s *session) error {
 //
 //nolint:cyclop,funlen,gocognit
 func fromClientWorker(ctx context.Context, stream Stream, s *session) error {
+	// Close stdinCh on exit so ChanReader.Read returns io.EOF and any module
+	// goroutine blocked on stdin (e.g. the serial inner goroutine) unblocks cleanly.
+	defer close(s.stdinCh)
+
 	type recvResult struct {
 		req *pb.RunRequest
 		err error