diff --git a/cmds/dutctl/rawconsole_test.go b/cmds/dutctl/rawconsole_test.go
new file mode 100644
index 0000000..ba7843f
--- /dev/null
+++ b/cmds/dutctl/rawconsole_test.go
@@ -0,0 +1,60 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package main
+
+import (
+	"bytes"
+	"testing"
+)
+
+// TestRawConsoleNeverArmsForNonFileStdin verifies that a non-*os.File stdin
+// (e.g. a piped/scripted run) never switches to raw mode, regardless of the
+// interactive hint, and that arm/disarm are safe no-ops in that case.
+func TestRawConsoleNeverArmsForNonFileStdin(t *testing.T) {
+	console := newRawConsole(&bytes.Buffer{}, true)
+
+	console.arm()
+
+	if console.isActive() {
+		t.Error("isActive() = true for non-file stdin, want false")
+	}
+
+	// disarm must not panic even though arm never engaged.
+	console.disarm()
+}
+
+// TestRawConsoleNeverArmsForScriptedInvocation verifies that when the command
+// was invoked with arguments (interactive=false), raw mode is never armed even
+// though the agent streams console output (modelled here by calling arm).
+func TestRawConsoleNeverArmsForScriptedInvocation(t *testing.T) {
+	// interactive=false models a serial expect/send sequence run.
+	console := newRawConsole(&bytes.Buffer{}, false)
+
+	console.arm()
+
+	if console.isActive() {
+		t.Error("isActive() = true for a scripted (argument-bearing) invocation, want false")
+	}
+
+	console.disarm()
+}
+
+// TestRawConsoleArmIsIdempotent verifies that repeated arm calls (one per
+// console message) do not panic and leave a consistent state. With a non-file
+// stdin it stays inactive; the point is that calling arm many times is safe.
+func TestRawConsoleArmIsIdempotent(t *testing.T) {
+	console := newRawConsole(&bytes.Buffer{}, true)
+
+	for range 5 {
+		console.arm()
+	}
+
+	if console.isActive() {
+		t.Error("isActive() = true for non-file stdin, want false")
+	}
+
+	console.disarm()
+	console.disarm() // double disarm must be safe
+}
diff --git a/cmds/dutctl/rawmode_darwin.go b/cmds/dutctl/rawmode_darwin.go
new file mode 100644
index 0000000..a954fbb
--- /dev/null
+++ b/cmds/dutctl/rawmode_darwin.go
@@ -0,0 +1,15 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build darwin
+
+package main
+
+import "golang.org/x/sys/unix"
+
+// Terminal get/set ioctl request numbers for macOS (BSD-derived).
+const (
+	tcGetReq = unix.TIOCGETA
+	tcSetReq = unix.TIOCSETA
+)
diff --git a/cmds/dutctl/rawmode_linux.go b/cmds/dutctl/rawmode_linux.go
new file mode 100644
index 0000000..954bc29
--- /dev/null
+++ b/cmds/dutctl/rawmode_linux.go
@@ -0,0 +1,15 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build linux
+
+package main
+
+import "golang.org/x/sys/unix"
+
+// Terminal get/set ioctl request numbers for Linux.
+const (
+	tcGetReq = unix.TCGETS
+	tcSetReq = unix.TCSETS
+)
diff --git a/cmds/dutctl/rawmode_other.go b/cmds/dutctl/rawmode_other.go
new file mode 100644
index 0000000..9c14a2e
--- /dev/null
+++ b/cmds/dutctl/rawmode_other.go
@@ -0,0 +1,14 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build !linux && !darwin
+
+package main
+
+// setRawInput is a no-op on platforms without termios support (e.g. Windows).
+// Input stays line-buffered; the interactive serial experience is degraded but
+// the client still builds and runs.
+func setRawInput(_ int) func() {
+	return nil
+}
diff --git a/cmds/dutctl/rawmode_unix.go b/cmds/dutctl/rawmode_unix.go
new file mode 100644
index 0000000..5ca6c8a
--- /dev/null
+++ b/cmds/dutctl/rawmode_unix.go
@@ -0,0 +1,53 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build linux || darwin
+
+package main
+
+import "golang.org/x/sys/unix"
+
+// setRawInput puts the terminal into raw input mode so the interactive serial
+// session behaves like a direct console:
+//
+//   - ECHO/ICANON off: keystrokes are delivered immediately, character at a
+//     time, and not echoed locally (the DUT echoes them back).
+//   - ISIG off: control characters such as Ctrl-C, Ctrl-Z and Ctrl-\ are NOT
+//     turned into local signals; they are forwarded to the DUT as raw bytes.
+//   - IXON off: Ctrl-S/Ctrl-Q flow control is forwarded to the DUT instead of
+//     being swallowed by the local terminal.
+//   - IEXTEN off: Ctrl-V and friends are forwarded literally.
+//   - ICRNL off: a typed CR is sent as CR, not translated to NL.
+//
+// dutctl is exited with the client-side escape sequence (Ctrl-A x), not with a
+// terminal signal — see filterEscape in rpc.go.
+//
+// It returns a restore function, or nil if the fd is not a terminal (in which
+// case input stays line-buffered, which is the correct fallback for pipes).
+//
+// The ioctl request numbers differ per OS (tcGetReq/tcSetReq are defined in the
+// platform-specific files); the termios flags themselves are shared across
+// unix platforms.
+func setRawInput(fileDescriptor int) func() {
+	termios, err := unix.IoctlGetTermios(fileDescriptor, tcGetReq)
+	if err != nil {
+		return nil
+	}
+
+	old := *termios
+
+	termios.Iflag &^= unix.ICRNL | unix.IXON
+	termios.Lflag &^= unix.ECHO | unix.ICANON | unix.ISIG | unix.IEXTEN
+	termios.Cc[unix.VMIN] = 1
+	termios.Cc[unix.VTIME] = 0
+
+	err = unix.IoctlSetTermios(fileDescriptor, tcSetReq, termios)
+	if err != nil {
+		return nil
+	}
+
+	return func() {
+		_ = unix.IoctlSetTermios(fileDescriptor, tcSetReq, &old)
+	}
+}
diff --git a/cmds/dutctl/rpc.go b/cmds/dutctl/rpc.go
index 161997f..1d68e7b 100644
--- a/cmds/dutctl/rpc.go
+++ b/cmds/dutctl/rpc.go
@@ -5,7 +5,6 @@
 package main
 
 import (
-	"bufio"
 	"context"
 	"errors"
 	"fmt"
@@ -14,6 +13,8 @@ import (
 	"log/slog"
 	"os"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"connectrpc.com/connect"
@@ -35,6 +36,84 @@ var errInterrupted = errors.New("interrupted")
 // streaming Run deliberately has no overall deadline (see runRPC).
 const unaryTimeout = 30 * time.Second
 
+// rawConsole lazily switches the terminal to raw input mode, and prints the
+// interactive session banner, the first time the agent actually streams console
+// output. One-shot commands (power, flash) only ever send Print messages, so
+// they never arm it: the terminal is left untouched and no misleading banner is
+// shown. A serial session sends its "Connected" banner as console output within
+// milliseconds, which arms it just in time for keystroke forwarding.
+//
+// The zero value is not usable; build one with newRawConsole.
+type rawConsole struct {
+	fd      int         // stdin file descriptor (valid only when canRaw is true)
+	canRaw  bool        // stdin is an *os.File, so raw mode may be attempted
+	once    sync.Once   // guards the one-time arm
+	active  atomic.Bool // true once raw mode is on; read by the send goroutine
+	mu      sync.Mutex  // guards restore
+	restore func()      // set by arm, called by disarm; nil until/unless armed
+}
+
+// newRawConsole returns a console that may switch the terminal to raw mode only
+// when both: interactive is true (the command was invoked without arguments, so
+// it is a hand-driven session rather than a scripted one) and stdin is a real
+// *os.File. A pipe, /dev/null, or any argument-bearing (scripted) invocation
+// yields canRaw=false, so it never changes the terminal nor prints the banner.
+func newRawConsole(stdin io.Reader, interactive bool) *rawConsole {
+	console := &rawConsole{}
+
+	if f, ok := stdin.(*os.File); ok && interactive {
+		console.fd = int(f.Fd())
+		console.canRaw = true
+	}
+
+	return console
+}
+
+// arm switches the terminal to raw mode and prints the session banner on its
+// first call; later calls are no-ops. setRawInput returns nil for a non-terminal
+// fd, so arming is silently skipped when stdin is not a TTY. Safe to call from
+// any goroutine.
+func (rc *rawConsole) arm() {
+	rc.once.Do(func() {
+		if !rc.canRaw {
+			return
+		}
+
+		restore := setRawInput(rc.fd)
+		if restore == nil {
+			return // not a terminal — keep input line-buffered
+		}
+
+		rc.mu.Lock()
+		rc.restore = restore
+		rc.mu.Unlock()
+
+		rc.active.Store(true)
+
+		// The escape sequence is the only way to quit while raw mode is on.
+		fmt.Fprint(os.Stderr, "\r\n[dutctl] interactive session — press Ctrl-A then x to quit\r\n")
+	})
+}
+
+// isActive reports whether raw mode is currently engaged. The send goroutine
+// uses it to decide whether to apply the Ctrl-A escape filter to stdin.
+func (rc *rawConsole) isActive() bool {
+	return rc.active.Load()
+}
+
+// disarm restores the terminal if arm switched it to raw mode. It is safe to
+// defer unconditionally and safe to call when arm never fired.
+func (rc *rawConsole) disarm() {
+	rc.mu.Lock()
+	restore := rc.restore
+	rc.restore = nil
+	rc.mu.Unlock()
+
+	if restore != nil {
+		restore()
+	}
+}
+
 func (app *application) listRPC(ctx context.Context) error {
 	ctx, cancel := context.WithTimeout(ctx, unaryTimeout)
 	defer cancel()
@@ -218,20 +297,83 @@ func (app *application) detailsRPC(ctx context.Context, device, command, keyword
 	return nil
 }
 
+// Interactive escape sequence. Ctrl-A is the prefix; the following key decides:
+//   - 'x'/'X'/Ctrl-X: quit dutctl.
+//   - Ctrl-A: send a single literal Ctrl-A to the DUT.
+//   - anything else: forward both the prefix and the key unchanged.
+//
+// Every other byte (Ctrl-C, Ctrl-D, Ctrl-Z, ...) is forwarded to the DUT.
+const (
+	escapePrefix   = 0x01 // Ctrl-A
+	escapeQuitCtrl = 0x18 // Ctrl-X
+)
+
+// filterEscape applies the interactive escape state machine to in. It returns
+// the bytes to forward to the DUT and whether the quit sequence was seen.
+// escapePending carries the "prefix seen" state across reads, so the prefix and
+// its following key may arrive in separate stdin reads.
+func filterEscape(in []byte, escapePending *bool) ([]byte, bool) {
+	out := make([]byte, 0, len(in))
+
+	for _, char := range in {
+		if *escapePending {
+			*escapePending = false
+
+			switch char {
+			case 'x', 'X', escapeQuitCtrl:
+				return out, true
+			case escapePrefix: // prefix pressed twice -> send one literal Ctrl-A
+				out = append(out, escapePrefix)
+			default: // not an escape; forward the prefix and this byte
+				out = append(out, escapePrefix, char)
+			}
+
+			continue
+		}
+
+		if char == escapePrefix {
+			*escapePending = true
+
+			continue
+		}
+
+		out = append(out, char)
+	}
+
+	return out, false
+}
+
 // runRPC executes command on device, streaming module output and forwarding
 // stdin and file transfers until the run ends. It returns nil on normal
 // completion, errInterrupted when a signal (Ctrl-C) ended the run, or a wrapped
 // error from a worker goroutine (stream send/receive or file I/O). A connect
 // status from the agent surfaces through the returned error; exit() renders it.
 //
-//nolint:funlen,cyclop,gocognit,maintidx // coordinates two streaming worker goroutines; inherently branchy
+//nolint:funlen,cyclop,gocognit,gocyclo,maintidx // two streaming workers plus raw-console handling; inherently branchy
 func (app *application) runRPC(ctx context.Context, device, command string, cmdArgs []string) error {
 	const numWorkers = 2 // The send and receive worker goroutines
 
+	// Raw input mode (no echo, no canonical line buffering, no local signal
+	// generation) is needed only for a live, hand-driven console session, so that
+	// each keystroke — including Ctrl-C — is forwarded to the DUT immediately and
+	// echoed by the remote side. Two conditions must hold, so it is armed lazily:
+	//
+	//   - The invocation is interactive, i.e. the command was given no arguments.
+	//     A command with arguments is parameterised/scripted (e.g. a serial
+	//     expect/send sequence the agent drives on its own); raw mode there would
+	//     only mislead and would steal Ctrl-C from aborting the client.
+	//   - The agent actually streams console output, which one-shot commands
+	//     (power, flash) never do — so they leave the terminal untouched.
+	//
+	// Piped/scripted stdin is never switched to raw mode and is forwarded verbatim.
+	console := newRawConsole(app.stdin, len(cmdArgs) == 0)
+	defer console.disarm()
+
 	// ctx is the shared signal context from dispatch, cancelled on SIGINT/SIGTERM
 	// so Ctrl-C terminates gracefully (running the normal teardown and flushing the
 	// warning summary) instead of killing the process. A stream has no overall
-	// deadline. runCtx is the child the workers cancel on completion.
+	// deadline. runCtx is the child the workers cancel on completion, including on
+	// the interactive quit sequence (Ctrl-A x).
 	runCtx, cancelRunCtx := context.WithCancel(ctx)
 	defer cancelRunCtx()
 
@@ -302,6 +444,11 @@ func (app *application) runRPC(ctx context.Context, device, command string, cmdA
 					Metadata: metadata,
 				})
 			case *pb.RunResponse_Console:
+				// First console output means this is a live console session:
+				// switch the terminal to raw mode now (no-op for non-TTY stdin)
+				// so keystrokes forward correctly and the banner is truthful.
+				console.arm()
+
 				switch consoleData := msg.Console.Data.(type) {
 				case *pb.Console_Stdout:
 					app.formatter.WriteContent(output.Content{
@@ -378,18 +525,28 @@ func (app *application) runRPC(ctx context.Context, device, command string, cmdA
 		}
 	}()
 
-	// Send routine — reads lines from stdin and forwards them to the server.
+	// Send routine — reads raw bytes from stdin and forwards them to the server.
 	//
 	// Unlike the receive routine this goroutine intentionally does NOT defer
-	// cancel(). When stdin reaches EOF (e.g. /dev/null in non-interactive
-	// runs) this goroutine returns immediately. If it cancelled the context
-	// on exit, the receive routine would be torn down before it could read
-	// and print the server's response.
+	// cancel() for the EOF case. When stdin reaches EOF (e.g. /dev/null in
+	// non-interactive runs) this goroutine returns immediately; if it cancelled
+	// the context on exit, the receive routine would be torn down before it
+	// could read and print the server's response.
 	//
-	// Only the receive routine drives context cancellation so that all
-	// server output is processed before the RPC terminates.
+	// It DOES cancel when the user types the interactive escape sequence
+	// (Ctrl-A x): that is an explicit request to end the session.
+	//
+	// app.stdin.Read is not interruptible by runCtx, so on a normal one-shot
+	// completion this goroutine stays blocked in Read after runRPC returns and is
+	// only reclaimed when the process exits. That is acceptable here because dutctl
+	// runs one command per invocation and exits immediately; a longer-lived caller
+	// of runRPC would need a closeable stdin to reclaim it.
 	go func() {
-		reader := bufio.NewReader(app.stdin)
+		const stdinBufSize = 256
+
+		buf := make([]byte, stdinBufSize)
+
+		var escapePending bool
 
 		for {
 			select {
@@ -400,7 +557,7 @@ func (app *application) runRPC(ctx context.Context, device, command string, cmdA
 			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)
@@ -409,17 +566,35 @@ func (app *application) runRPC(ctx context.Context, device, command string, cmdA
 				return
 			}
 
-			err = stream.Send(&pb.RunRequest{
-				Msg: &pb.RunRequest_Console{
-					Console: &pb.Console{
-						Data: &pb.Console_Stdin{
-							Stdin: []byte(text),
+			payload := buf[:nRead]
+
+			quit := false
+			if console.isActive() {
+				// Intercept the escape sequence; forward everything else
+				// (including Ctrl-C) untouched.
+				payload, quit = filterEscape(payload, &escapePending)
+			}
+
+			if len(payload) > 0 {
+				sendErr := stream.Send(&pb.RunRequest{
+					Msg: &pb.RunRequest_Console{
+						Console: &pb.Console{
+							Data: &pb.Console_Stdin{
+								Stdin: payload,
+							},
 						},
 					},
-				},
-			})
-			if err != nil {
-				errChan <- fmt.Errorf("sending RPC message: %w", err)
+				})
+				if sendErr != nil {
+					errChan <- fmt.Errorf("sending RPC message: %w", sendErr)
+
+					return
+				}
+			}
+
+			if quit {
+				slog.Debug("send routine terminating", "reason", "escape sequence")
+				cancelRunCtx()
 
 				return
 			}
diff --git a/cmds/dutctl/rpc_escape_test.go b/cmds/dutctl/rpc_escape_test.go
new file mode 100644
index 0000000..9477c4f
--- /dev/null
+++ b/cmds/dutctl/rpc_escape_test.go
@@ -0,0 +1,145 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package main
+
+import "testing"
+
+// TestFilterEscapeSingleCall covers escape handling when the prefix and the
+// following key arrive within one stdin read.
+func TestFilterEscapeSingleCall(t *testing.T) {
+	tests := []struct {
+		name     string
+		in       []byte
+		wantOut  []byte
+		wantQuit bool
+	}{
+		{
+			name:    "plain bytes pass through",
+			in:      []byte("hello"),
+			wantOut: []byte("hello"),
+		},
+		{
+			name:    "ctrl-c is forwarded to the DUT",
+			in:      []byte{0x03},
+			wantOut: []byte{0x03},
+		},
+		{
+			name:    "ctrl-d is forwarded to the DUT",
+			in:      []byte{0x04},
+			wantOut: []byte{0x04},
+		},
+		{
+			name:    "ctrl-z is forwarded to the DUT",
+			in:      []byte{0x1a},
+			wantOut: []byte{0x1a},
+		},
+		{
+			name:     "ctrl-a then x quits",
+			in:       []byte{escapePrefix, 'x'},
+			wantOut:  []byte{},
+			wantQuit: true,
+		},
+		{
+			name:     "ctrl-a then X quits",
+			in:       []byte{escapePrefix, 'X'},
+			wantOut:  []byte{},
+			wantQuit: true,
+		},
+		{
+			name:     "ctrl-a then ctrl-x quits",
+			in:       []byte{escapePrefix, escapeQuitCtrl},
+			wantOut:  []byte{},
+			wantQuit: true,
+		},
+		{
+			name:    "ctrl-a ctrl-a sends one literal ctrl-a",
+			in:      []byte{escapePrefix, escapePrefix},
+			wantOut: []byte{escapePrefix},
+		},
+		{
+			name:    "ctrl-a then other key forwards both",
+			in:      []byte{escapePrefix, 'z'},
+			wantOut: []byte{escapePrefix, 'z'},
+		},
+		{
+			name:     "text before quit sequence is forwarded",
+			in:       append([]byte("ab"), escapePrefix, 'x'),
+			wantOut:  []byte("ab"),
+			wantQuit: true,
+		},
+		{
+			name:    "text around literal ctrl-a",
+			in:      append(append([]byte("a"), escapePrefix, escapePrefix), 'b'),
+			wantOut: append([]byte{'a', escapePrefix}, 'b'),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pending := false
+
+			out, quit := filterEscape(tt.in, &pending)
+
+			if string(out) != string(tt.wantOut) {
+				t.Errorf("out = %q, want %q", out, tt.wantOut)
+			}
+
+			if quit != tt.wantQuit {
+				t.Errorf("quit = %v, want %v", quit, tt.wantQuit)
+			}
+
+			if pending {
+				t.Errorf("escapePending left true, want false")
+			}
+		})
+	}
+}
+
+// TestFilterEscapeAcrossReads verifies the escape prefix and its following key
+// are handled correctly when they arrive in separate stdin reads.
+func TestFilterEscapeAcrossReads(t *testing.T) {
+	pending := false
+
+	// Read 1: lone prefix — nothing forwarded yet, state armed.
+	out, quit := filterEscape([]byte{escapePrefix}, &pending)
+	if len(out) != 0 || quit {
+		t.Fatalf("read 1: out=%q quit=%v, want empty/false", out, quit)
+	}
+
+	if !pending {
+		t.Fatal("read 1: escapePending = false, want true")
+	}
+
+	// Read 2: the quit key arrives separately.
+	out, quit = filterEscape([]byte{'x'}, &pending)
+	if !quit {
+		t.Errorf("read 2: quit = false, want true")
+	}
+
+	if len(out) != 0 {
+		t.Errorf("read 2: out = %q, want empty", out)
+	}
+
+	if pending {
+		t.Errorf("read 2: escapePending = true, want false")
+	}
+}
+
+// TestFilterEscapePrefixThenForward verifies an armed prefix followed by a
+// normal byte in the next read forwards both bytes and disarms.
+func TestFilterEscapePrefixThenForward(t *testing.T) {
+	pending := false
+
+	filterEscape([]byte{escapePrefix}, &pending) // arm
+
+	out, quit := filterEscape([]byte{'k'}, &pending)
+	if quit {
+		t.Errorf("quit = true, want false")
+	}
+
+	if string(out) != string([]byte{escapePrefix, 'k'}) {
+		t.Errorf("out = %q, want %q", out, []byte{escapePrefix, 'k'})
+	}
+}
diff --git a/go.mod b/go.mod
index d978879..906c2ec 100644
--- a/go.mod
+++ b/go.mod
@@ -11,6 +11,7 @@ require (
 	go.bug.st/serial v1.8.0
 	golang.org/x/crypto v0.54.0
 	golang.org/x/mod v0.38.0
+	golang.org/x/sys v0.47.0
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -33,7 +34,6 @@ require (
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/rogpeppe/go-internal v1.6.1 // indirect
 	golang.org/x/net v0.56.0 // indirect
-	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/term v0.45.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
 )
diff --git a/internal/dutagent/session/broker_test.go b/internal/dutagent/session/broker_test.go
index d462acb..eab38da 100644
--- a/internal/dutagent/session/broker_test.go
+++ b/internal/dutagent/session/broker_test.go
@@ -35,7 +35,9 @@ func (s *testStream) Send(_ *pb.RunResponse) error {
 
 func (s *testStream) Receive() (*pb.RunRequest, error) {
 	if s.recvBlock {
-		<-s.unblockCh // blocks until the test closes it; simulates a long receive
+		// unblockCh must be created by the test before Start so this goroutine
+		// only reads it, never writes it (a write here would race the test).
+		<-s.unblockCh // will block until closed; simulates a long receive
 	}
 
 	idx := s.recvCalls
@@ -133,7 +135,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)
 
@@ -145,7 +148,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):
-		// Timed out waiting for the forwarded stdin payload.
+		t.Fatal("timeout: stdin payload was not forwarded to stdinCh")
 	}
 
 	cancel() // simulate module completion
diff --git a/internal/dutagent/session/worker.go b/internal/dutagent/session/worker.go
index 2228622..6dc4e18 100644
--- a/internal/dutagent/session/worker.go
+++ b/internal/dutagent/session/worker.go
@@ -119,6 +119,10 @@ func toClientWorker(ctx context.Context, stream Stream, s *backend) error {
 func fromClientWorker(ctx context.Context, stream Stream, s *backend) error {
 	l := log.FromContext(ctx)
 
+	// 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
@@ -202,6 +206,10 @@ func fromClientWorker(ctx context.Context, stream Stream, s *backend) error {
 
 					l.Debug("received stdin from client", "bytes", len(stdin))
 
+					// This is the only writer to stdinCh, which is why fromClientWorker
+					// may close it on return (see the deferred close above) without
+					// risking a send on a closed channel. Keep it that way: a second
+					// writer would turn that close into a panic.
 					select {
 					case <-ctx.Done():
 						return nil
diff --git a/pkg/module/serial/engine.go b/pkg/module/serial/engine.go
index b81bae7..7559916 100644
--- a/pkg/module/serial/engine.go
+++ b/pkg/module/serial/engine.go
@@ -5,9 +5,11 @@
 package serial
 
 import (
+	"bytes"
 	"context"
 	"io"
 	"regexp"
+	"sync"
 	"time"
 )
 
@@ -31,6 +33,18 @@ const matchWindow = 64 * 1024
 // readChunk is the size of a single read from the serial port.
 const readChunk = 4096
 
+// reconnectInterval is the delay between attempts to reopen the serial device
+// after it has disappeared (e.g. an FTDI chip that powers down with the DUT).
+const reconnectInterval = 500 * time.Millisecond
+
+// deviceLossGraceDefault is how long a quiet port is tolerated before the module
+// checks whether the device node still exists. A read timeout and a removed
+// device look identical (both yield a data-less read), so the module waits this
+// long, then confirms via the node check before treating quiet as loss. Kept
+// short so a real disconnect is noticed quickly, but long enough not to stat the
+// node on every idle read.
+const deviceLossGraceDefault = 2 * time.Second
+
 // engine drives a serial port for the scripted send/expect mode. It owns a
 // rolling match buffer (never reset on newline) and tees every byte read from
 // the port to an output sink (the client console in scripted mode).
@@ -46,19 +60,181 @@ type engine struct {
 	// by -keep-escapes); remainder carries a partial sequence split across reads.
 	filter    bool
 	remainder []byte
+
+	// mu guards p, which is swapped when the device disappears and is reopened
+	// (see reconnect). The read loop copies p under mu, then reads without the
+	// lock; the interactive stdin pump writes through portWrite under mu. The
+	// lock is never held across a blocking read, so the two cannot deadlock.
+	mu sync.Mutex
+
+	// reopen reopens the device after loss; a nil reopen disables auto-reconnect
+	// (used by the direct engine unit tests and the post-send drain). present
+	// reports whether the device node still exists; grace is the quiet period
+	// tolerated before the node is checked. All three are set together via
+	// withReconnect.
+	reopen  func() (port, error)
+	present func() bool
+	grace   time.Duration
 }
 
 func newEngine(p port, sink io.Writer, filter bool) *engine {
 	return &engine{p: p, sink: sink, buf: make([]byte, 0, readChunk), filter: filter}
 }
 
+// withReconnect enables auto-reconnect on device loss for this engine. reopen
+// re-opens the device, present reports whether its node still exists, and grace
+// is how long a quiet port is tolerated before the node is checked. Run wires
+// this up for the streaming and expect modes; it is left unset for the direct
+// engine unit tests and for the post-send drain, which stay non-reconnecting.
+func (e *engine) withReconnect(reopen func() (port, error), present func() bool, grace time.Duration) {
+	e.reopen = reopen
+	e.present = present
+	e.grace = grace
+}
+
+// currentPort returns the live port handle under the lock, so a caller can read
+// or write it without racing a reconnect that swaps it.
+func (e *engine) currentPort() port {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	return e.p
+}
+
+// setPort installs a freshly opened port after a reconnect.
+func (e *engine) setPort(p port) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	e.p = p
+}
+
+// closeCurrent closes the live port (best-effort) and clears the handle, so a
+// dropped session is fully discarded and the stdin pump stops writing to it.
+func (e *engine) closeCurrent() {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	if e.p != nil {
+		_ = e.p.Close()
+		e.p = nil
+	}
+}
+
+// portWrite writes data to the live port under the lock. It drops the data when
+// the device is gone (handle nil during a reconnect) rather than blocking.
+func (e *engine) portWrite(data []byte) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	if e.p == nil {
+		return nil // device gone; drop input until it reconnects
+	}
+
+	for len(data) > 0 {
+		n, err := e.p.Write(data)
+		if err != nil {
+			return err
+		}
+
+		data = data[n:]
+	}
+
+	return nil
+}
+
+// marker writes a status line straight to the output sink (used for the
+// disconnect/reconnect banners, which must appear even mid-stream).
+func (e *engine) marker(msg string) {
+	_, _ = e.sink.Write([]byte(msg))
+}
+
+// read reads one chunk from the live port. The handle is copied under the lock
+// so a concurrent reconnect is safe, while the blocking read runs without the
+// lock so it never stalls an interactive stdin write.
+func (e *engine) read(buf []byte) (int, error) {
+	p := e.currentPort()
+	if p == nil {
+		return 0, nil
+	}
+
+	return p.Read(buf)
+}
+
+// handleReadLoss decides whether a data-less read reflects a vanished device
+// and, if reconnect is enabled, waits for the device to return. It reports
+// whether a reconnect happened (the caller should retry the read) and returns
+// ctx.Err() if the wait was cancelled. With reconnect disabled it is a no-op
+// returning (false, nil), so non-reconnecting callers keep their old behaviour.
+func (e *engine) handleReadLoss(ctx context.Context, readErr error, lastData *time.Time) (bool, error) {
+	if e.reopen == nil {
+		return false, nil
+	}
+
+	// A hard read error is an outright device error → reconnect immediately. A
+	// quiet read (0, nil) only counts as loss once the device node has been gone
+	// for the grace period: an idle read and a removed device look identical, so
+	// the node check disambiguates a merely quiet console from a vanished one.
+	if readErr == nil {
+		if e.present == nil || e.present() || time.Since(*lastData) <= e.grace {
+			return false, nil
+		}
+	}
+
+	err := e.reconnect(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	*lastData = time.Now()
+
+	return true, nil
+}
+
+// reconnect closes the vanished port and retries opening it until the device
+// reappears, the deadline fires, or the session is cancelled. It mirrors tio's
+// auto-reconnect so a DUT power-cycle that drops the FTDI device does not end
+// the session. On return either a fresh port is installed (nil error) or ctx is
+// done (its error).
+func (e *engine) reconnect(ctx context.Context) error {
+	e.closeCurrent()
+
+	e.marker("\n--- Serial device disconnected, waiting to reconnect ---\n")
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		p, err := e.reopen()
+		if err == nil {
+			_ = p.ResetInputBuffer()
+			e.setPort(p)
+			e.marker("\n--- Serial device reconnected ---\n")
+
+			return nil
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(reconnectInterval):
+		}
+	}
+}
+
 // readUntil reads from the port, forwarding every byte to the sink, until
 // pattern matches the accumulated output or ctx is done (global timeout /
 // cancel).
 // On a match, the matched span and everything before it is consumed from the
 // buffer, so a later expect only sees subsequent output.
+//
+//nolint:cyclop // match-check, resilient read, and device-loss handling share one loop; splitting hurts clarity
 func (e *engine) readUntil(ctx context.Context, pattern *regexp.Regexp) error {
 	readBuf := make([]byte, readChunk)
+	lastData := time.Now()
 
 	for {
 		// Check the whole buffer for a match FIRST, so output a prior step left
@@ -80,8 +256,10 @@ func (e *engine) readUntil(ctx context.Context, pattern *regexp.Regexp) error {
 		default:
 		}
 
-		n, err := e.p.Read(readBuf)
+		n, err := e.read(readBuf)
 		if n > 0 {
+			lastData = time.Now()
+
 			out := e.clean(readBuf[:n])
 			if len(out) > 0 {
 				// Tee: forward to the client, then feed the matcher (re-checked
@@ -97,9 +275,20 @@ func (e *engine) readUntil(ctx context.Context, pattern *regexp.Regexp) error {
 			continue
 		}
 
-		// go.bug.st returns (0, nil) on a read timeout — keep waiting and let
-		// the ctx check above enforce the overall deadline. Any real error
-		// (port closed / device gone) ends the step.
+		// No data. go.bug.st returns (0, nil) on a read timeout; a real error
+		// means the device is gone. With reconnect enabled, wait for the device
+		// to return and keep matching against the preserved buffer. Otherwise a
+		// hard error ends the step, while a timeout just keeps waiting for the
+		// deadline (checked at the top of the loop).
+		lost, rerr := e.handleReadLoss(ctx, err, &lastData)
+		if rerr != nil {
+			return rerr
+		}
+
+		if lost {
+			continue
+		}
+
 		if err != nil {
 			return err
 		}
@@ -109,13 +298,9 @@ func (e *engine) readUntil(ctx context.Context, pattern *regexp.Regexp) error {
 // write sends payload to the port, looping over short writes, then resets the
 // match buffer so the next expect starts from output produced after the send.
 func (e *engine) write(payload []byte) error {
-	for len(payload) > 0 {
-		n, err := e.p.Write(payload)
-		if err != nil {
-			return err
-		}
-
-		payload = payload[n:]
+	err := e.portWrite(payload)
+	if err != nil {
+		return err
 	}
 
 	e.buf = e.buf[:0]
@@ -135,12 +320,18 @@ func (e *engine) capBuffer() {
 }
 
 // pump reads from the port and forwards every byte to the sink until ctx is
-// done (a normal end for a watch/drain, so it returns nil). On a real read
-// error it returns the error, unless swallowReadErr is set — used by the
-// post-send drain, where the device may have rebooted from the final send and
-// the sequence has already completed successfully.
-func (e *engine) pump(ctx context.Context, swallowReadErr bool) error {
+// done (a normal end for a watch/drain, so it returns nil).
+//
+// With reconnectOnLoss set, a vanished device is handled transparently: the pump
+// waits for it to reappear and keeps streaming (used by monitor and interactive
+// mode). With it clear, a read error ends the pump — returning the error, or nil
+// when swallowReadErr is set, as the post-send drain does since the device may
+// have rebooted from that final send.
+//
+//nolint:cyclop // read/emit/reconnect/idle dispatch in one streaming loop; the branch count is inherent
+func (e *engine) pump(ctx context.Context, reconnectOnLoss, swallowReadErr bool) error {
 	readBuf := make([]byte, readChunk)
+	lastData := time.Now()
 
 	for {
 		select {
@@ -149,8 +340,10 @@ func (e *engine) pump(ctx context.Context, swallowReadErr bool) error {
 		default:
 		}
 
-		n, err := e.p.Read(readBuf)
+		n, err := e.read(readBuf)
 		if n > 0 {
+			lastData = time.Now()
+
 			out := e.clean(readBuf[:n])
 			if len(out) > 0 {
 				_, werr := e.sink.Write(out)
@@ -160,6 +353,18 @@ func (e *engine) pump(ctx context.Context, swallowReadErr bool) error {
 			}
 		}
 
+		if reconnectOnLoss && e.reopen != nil {
+			// handleReadLoss reconnects on a hard error or a vanished node; on
+			// ctx cancellation it returns an error, which for a pump is a normal
+			// end. A benign idle read falls through and the loop continues.
+			_, rerr := e.handleReadLoss(ctx, err, &lastData)
+			if rerr != nil {
+				return nil //nolint:nilerr // ctx cancelled mid-reconnect is a normal end for a pump
+			}
+
+			continue
+		}
+
 		if err != nil {
 			if swallowReadErr {
 				return nil
@@ -172,19 +377,64 @@ func (e *engine) pump(ctx context.Context, swallowReadErr bool) error {
 
 // monitor streams serial output to the sink until ctx is done. It does no
 // matching; the deadline or a client cancel is a normal end, so it returns nil
-// in those cases and only fails on a real read error.
+// in those cases. With reconnect wired up (see Run) it survives device loss.
 func (e *engine) monitor(ctx context.Context) error {
-	return e.pump(ctx, false)
+	return e.pump(ctx, true, false)
 }
 
 // drain forwards serial output for up to d, so the DUT's reply to a final send
 // is visible before the run closes. It is best-effort: a read error (e.g. the
-// device rebooted from that send) ends it without failing the run.
+// device rebooted from that send) ends it without failing the run, and it does
+// not reconnect — the sequence has already completed.
 func (e *engine) drain(ctx context.Context, d time.Duration) error {
 	drainCtx, cancel := context.WithTimeout(ctx, d)
 	defer cancel()
 
-	return e.pump(drainCtx, true)
+	return e.pump(drainCtx, false, true)
+}
+
+// stdinChunk is the read size for the interactive stdin pump.
+const stdinChunk = 256
+
+// interactive bridges the client console and the serial port: it forwards
+// client keystrokes (stdin) to the port and streams port output to the sink,
+// until ctx is done. Device loss is handled by the pump's reconnect path (wired
+// via withReconnect). It returns nil on ctx cancellation, a normal end that the
+// caller maps to the right result.
+func (e *engine) interactive(ctx context.Context, stdin io.Reader) error {
+	// done stops the stdin pump from writing to the port once interactive
+	// returns and the port is about to be closed.
+	done := make(chan struct{})
+	defer close(done)
+
+	go e.stdinPump(stdin, done)
+
+	return e.pump(ctx, true, false)
+}
+
+// stdinPump forwards client keystrokes to the serial port until stdin reaches
+// EOF (the agent closes stdin on session teardown). It is the only goroutine
+// that may briefly outlive interactive(); it touches no match state and drops
+// writes once done is closed, so it can neither race the read loop nor write to
+// a closed port.
+func (e *engine) stdinPump(stdin io.Reader, done <-chan struct{}) {
+	buf := make([]byte, stdinChunk)
+
+	for {
+		n, err := stdin.Read(buf)
+		if n > 0 {
+			select {
+			case <-done:
+				return // interactive exited — do not write to the closing port.
+			default:
+				_ = e.portWrite(buf[:n])
+			}
+		}
+
+		if err != nil {
+			return // stdin EOF on teardown, or a read error.
+		}
+	}
 }
 
 // escByte is the ASCII escape that begins ANSI/VT escape sequences.
@@ -244,9 +494,18 @@ func (e *engine) clean(chunk []byte) []byte {
 // change still matches. An incomplete sequence at the end of data is stored in
 // remainder to be prepended to the next read.
 func stripEscapes(data []byte, remainder *[]byte) []byte {
-	result := make([]byte, 0, len(data))
 	*remainder = nil
 
+	// Most serial output carries no escape sequence; when there is none, return
+	// the input untouched. The result is consumed synchronously by the caller
+	// (written to the sink or copied into the match buffer) before the next read,
+	// so aliasing data is safe.
+	if bytes.IndexByte(data, escByte) < 0 {
+		return data
+	}
+
+	result := make([]byte, 0, len(data))
+
 	for idx := 0; idx < len(data); idx++ {
 		if data[idx] != escByte {
 			result = append(result, data[idx])
diff --git a/pkg/module/serial/serial.go b/pkg/module/serial/serial.go
index 8d442e8..789d4a2 100644
--- a/pkg/module/serial/serial.go
+++ b/pkg/module/serial/serial.go
@@ -2,14 +2,15 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-// Package serial provides a dutagent module that runs a scripted send/expect
-// sequence against a DUT's serial port.
+// Package serial provides a dutagent module that streams a DUT's serial console,
+// runs a scripted send/expect sequence, or bridges the console interactively.
 package serial
 
 import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"strings"
 	"time"
 
@@ -47,11 +48,8 @@ const sendDrain = time.Second
 // substitute a fake port for the real hardware.
 type portOpener func(name string, baud int) (port, error)
 
-// Serial observes the DUT's serial output and runs a scripted sequence of
-// send/expect steps against the serial port.
-//
-// The module is non-interactive: all input is supplied up front as arguments,
-// which makes it suitable for scripts and automated callers.
+// Serial streams the DUT's serial output, runs a scripted send/expect sequence,
+// or bridges the console interactively, depending on its arguments.
 type Serial struct {
 	Port  string // Port is the path to the serial device on the dutagent.
 	Baud  int    // Baud is the baud rate of the serial device. If unset, DefaultBaudRate is used.
@@ -66,17 +64,28 @@ type Serial struct {
 	// open opens the serial port. It defaults to defaultOpenPort in Run; tests
 	// set it to a fake. Opening happens per Run, never on the struct.
 	open portOpener
+
+	// portPresent reports whether the configured device node still exists; it
+	// defaults to a filesystem stat of Port and is overridden in tests. It lets
+	// the read loop tell a quiet-but-healthy console apart from a vanished
+	// device when deciding whether to reconnect.
+	portPresent func() bool
+
+	// deviceLossGrace overrides deviceLossGraceDefault in tests so the
+	// device-loss path can be exercised quickly. Zero uses the default.
+	deviceLossGrace time.Duration
 }
 
 // Ensure implementing the Module interface.
 var _ module.Module = &Serial{}
 
-const abstract = `Scripted serial connection to the DUT
+const abstract = `Serial connection to the DUT
 `
 
 const usage = `
 ARGUMENTS:
 	[-t <duration>] [-eol cr|lf|crlf|none] [-keep-escapes]                 (monitor: stream output)
+	[-t <duration>] [-keep-escapes] -i                                     (interactive console)
 	[-t <duration>] [-eol cr|lf|crlf|none] [-keep-escapes] [--] <step>...  (run a step sequence)
 
 	step := expect <regex> | send <data> | send-raw <data>
@@ -84,15 +93,24 @@ ARGUMENTS:
 `
 
 const description = `
-The serial module automates interaction with the DUT's serial console. All
-input is provided up front as arguments, so it suits scripts and automated
-callers.
+The serial module interacts with the DUT's serial console. It runs in one of
+three modes, selected by its arguments:
+
+  - MONITOR (no arguments): stream the serial output to the client until the
+    session is cancelled, or until -t elapses (a success).
+  - INTERACTIVE (-i): bridge the client console to the serial port, forwarding
+    keystrokes to the port and streaming output back, until the session is
+    cancelled or -t elapses. Takes no steps.
+  - STEP SEQUENCE (one or more steps): run the scripted steps in order (see
+    below); suits scripts and automated callers.
+
+Serial output is forwarded to the client in monitor and interactive modes, while
+waiting for an expect, and (if the last step is a send) for a moment afterwards
+so its reply is visible.
 
-With no steps it runs in MONITOR mode: it streams the serial output to the
-client until the session is cancelled, or until -t elapses (a success). With
-one or more steps it runs them in order (see below). Serial output is forwarded
-to the client in monitor mode, while waiting for an expect, and (if the last
-step is a send) for a moment afterwards so its reply is visible.
+If the serial device disappears mid-session (e.g. an FTDI chip that powers down
+when the DUT loses power), the module waits for it to reappear and reconnects
+automatically instead of ending the session.
 
 STEPS (executed in order; the run fails on the first expect that times out):
 	expect <regex>   Wait until the serial output matches the RE2 regular
@@ -117,6 +135,8 @@ FLAGS (before the steps):
 	                       which is what serial consoles expect on Enter.
 	-keep-escapes          Keep terminal escape sequences (cursor moves, colour,
 	                       queries) instead of stripping them from the output.
+	-i                     Interactive: bridge the client console to the serial
+	                       port. Cannot be combined with steps.
 
 Terminal escape sequences (cursor moves, colour, queries) are stripped from the
 output before it is shown or matched, unless -keep-escapes is given. Expect
@@ -127,6 +147,7 @@ right after a send can match your own input rather than the device's reply.
 
 EXAMPLES:
 	monitor the console:            (no arguments)
+	interactive console:            -i
 	wait for a boot marker:         -- expect 'Welcome to'
 	login then run a command:       -- expect 'login:' send root expect '# ' send reboot
 	send Ctrl-C then expect shell:  -- send-raw '\x03' expect '$ '
@@ -202,6 +223,37 @@ func defaultOpenPort(name string, baud int) (port, error) {
 	return serialPort, nil
 }
 
+// reopenFunc returns a closure that reopens the configured device; the engine
+// uses it to reconnect after the device disappears.
+func (s *Serial) reopenFunc(opener portOpener) func() (port, error) {
+	return func() (port, error) {
+		return opener(s.Port, s.Baud)
+	}
+}
+
+// presentFunc reports whether the configured device node still exists. It
+// defaults to a filesystem stat of Port and is overridden in tests.
+func (s *Serial) presentFunc() func() bool {
+	if s.portPresent != nil {
+		return s.portPresent
+	}
+
+	return func() bool {
+		_, err := os.Stat(s.Port)
+
+		return err == nil
+	}
+}
+
+// graceDur is the quiet period tolerated before the device-loss node check.
+func (s *Serial) graceDur() time.Duration {
+	if s.deviceLossGrace > 0 {
+		return s.deviceLossGrace
+	}
+
+	return deviceLossGraceDefault
+}
+
 // Run opens the configured serial port and either streams its output or
 // executes a step sequence. With no steps it runs in monitor mode, streaming
 // until the session is cancelled or -t elapses (both a success). With steps it
@@ -230,7 +282,6 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string
 	if err != nil {
 		return err
 	}
-	defer serialPort.Close()
 
 	// Discard any stale bytes left in the kernel/driver RX buffer from a
 	// previous session, otherwise a step could match data from the last boot.
@@ -241,11 +292,8 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string
 
 	l.Info(fmt.Sprintf("connected to %s at %d baud", s.Port, s.Baud))
 
-	clientOut := newClientWriter(session)
-	clientOut.markerf("--- Connected to %s at %d baud ---\n", s.Port, s.Baud)
-
-	// loopCtx carries the per-sequence deadline (-t). The original ctx is kept
-	// for the post-send drain so the drain gets its own full window.
+	// loopCtx carries the per-run deadline (-t). The original ctx is kept for the
+	// post-send drain so the drain gets its own full window.
 	loopCtx := ctx
 
 	if cfg.timeout > 0 {
@@ -257,7 +305,17 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string
 		defer cancel()
 	}
 
+	if cfg.interactive {
+		return s.runInteractive(ctx, loopCtx, session, serialPort, opener, cfg)
+	}
+
+	clientOut := newClientWriter(session)
+	clientOut.markerf("--- Connected to %s at %d baud ---\n", s.Port, s.Baud)
+
 	eng := newEngine(serialPort, clientOut, !cfg.keepEscapes)
+	eng.withReconnect(s.reopenFunc(opener), s.presentFunc(), s.graceDur())
+
+	defer eng.closeCurrent()
 
 	// Monitor mode: no steps — stream the console until cancelled or -t elapses.
 	if len(cfg.steps) == 0 {
@@ -319,6 +377,43 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string
 	return nil
 }
 
+// runInteractive bridges the client console to the serial port for a live
+// session: keystrokes are forwarded to the port and port output is streamed
+// back, until the client disconnects (ctx cancel) or -t elapses. The device is
+// reopened automatically if it disappears mid-session.
+func (s *Serial) runInteractive(
+	baseCtx, loopCtx context.Context,
+	session module.Session,
+	initialPort port,
+	opener portOpener,
+	cfg scriptConfig,
+) error {
+	l := log.FromContext(baseCtx)
+
+	stdin, stdout, _ := session.Console()
+
+	l.Debug("interactive mode; bridging console to serial port")
+	fmt.Fprintf(stdout, "--- Connected to %s at %d baud ---\n", s.Port, s.Baud)
+
+	eng := newEngine(initialPort, stdout, !cfg.keepEscapes)
+	eng.withReconnect(s.reopenFunc(opener), s.presentFunc(), s.graceDur())
+
+	defer eng.closeCurrent()
+
+	// Context cancellation — the client disconnecting (an interactive quit or
+	// agent teardown) or the -t deadline — is the normal end of an open-ended
+	// session, as in monitor mode, so interactive returns nil for it. Only a
+	// genuine engine error (e.g. a broken client stream) fails the run.
+	err := eng.interactive(loopCtx, stdin)
+	if err != nil {
+		return fmt.Errorf("interactive: %w", err)
+	}
+
+	fmt.Fprintf(stdout, "\n--- Connection closed ---\n")
+
+	return nil
+}
+
 // sleepCtx pauses for d, returning ctx.Err() if ctx is done first. A
 // non-positive d returns nil immediately.
 func sleepCtx(ctx context.Context, d time.Duration) error {
@@ -353,10 +448,12 @@ func stepError(idx int, failedStep step, timeout time.Duration, err error) error
 }
 
 // clientWriter forwards serial output to the client (it is the engine's output
-// sink) and tracks whether the stream is at the start of a line. Status lines
-// go through markerf, which inserts a newline first when the preceding output
-// (e.g. a prompt with no trailing newline) did not end one — so every marker
-// lands on its own line.
+// sink in monitor and step modes) and tracks whether the stream is at the start
+// of a line. Status lines go through markerf, which inserts a newline first when
+// the preceding output (e.g. a prompt with no trailing newline) did not end one
+// — so every marker lands on its own line.
+//
+// Interactive mode instead uses the stdout writer from session.Console().
 type clientWriter struct {
 	session     module.Session
 	atLineStart bool
diff --git a/pkg/module/serial/serial_reconnect_test.go b/pkg/module/serial/serial_reconnect_test.go
new file mode 100644
index 0000000..b992565
--- /dev/null
+++ b/pkg/module/serial/serial_reconnect_test.go
@@ -0,0 +1,334 @@
+// Copyright 2025 Blindspot Software
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package serial
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/BlindspotSoftware/dutctl/internal/test/mock"
+)
+
+// errDeviceGone models the hard read error a removed USB serial adapter surfaces.
+var errDeviceGone = errors.New("read /dev/ttyUSB0: input/output error")
+
+// livePort is a goroutine-safe port fake for the interactive tests, where the
+// stdin pump writes to the port concurrently with the read loop and the test
+// polling written bytes. Queued chunks are delivered by Read in order; once
+// drained it optionally reports a one-shot disconnect, then emulates a serial
+// read timeout so the read loop spins at a bounded rate.
+type livePort struct {
+	mu            sync.Mutex
+	out           [][]byte
+	written       []byte
+	closed        bool
+	disconnectErr error
+	disconnected  bool
+}
+
+func (p *livePort) queue(b []byte) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.out = append(p.out, b)
+}
+
+func (p *livePort) Read(b []byte) (int, error) {
+	p.mu.Lock()
+	if len(p.out) > 0 {
+		chunk := p.out[0]
+		n := copy(b, chunk)
+
+		if n < len(chunk) {
+			p.out[0] = chunk[n:]
+		} else {
+			p.out = p.out[1:]
+		}
+		p.mu.Unlock()
+
+		return n, nil
+	}
+
+	if p.disconnectErr != nil && !p.disconnected {
+		p.disconnected = true
+		err := p.disconnectErr
+		p.mu.Unlock()
+
+		return 0, err
+	}
+	p.mu.Unlock()
+
+	// Emulate the real port's read timeout so the read loop is bounded, not a
+	// busy-wait. The lock is released first so a concurrent write never blocks.
+	time.Sleep(2 * time.Millisecond)
+
+	return 0, nil
+}
+
+func (p *livePort) Write(b []byte) (int, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.written = append(p.written, b...)
+
+	return len(b), nil
+}
+
+func (p *livePort) ResetInputBuffer() error { return nil }
+
+func (p *livePort) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.closed = true
+
+	return nil
+}
+
+func (p *livePort) writtenString() string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	return string(p.written)
+}
+
+// syncBuffer is a goroutine-safe io.Writer for capturing interactive stdout,
+// which the read loop writes concurrently with the test reading it.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (s *syncBuffer) Write(b []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.buf.Write(b)
+}
+
+func (s *syncBuffer) String() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.buf.String()
+}
+
+// interactiveSession wires a mock.Session with a pipe-backed stdin (so the stdin
+// pump unblocks on cleanup) and a capturing stdout. It returns the session, the
+// stdout capture, and the stdin write end.
+func interactiveSession(t *testing.T) (*mock.Session, *syncBuffer, *io.PipeWriter) {
+	t.Helper()
+
+	pr, pw := io.Pipe()
+	stdout := &syncBuffer{}
+
+	t.Cleanup(func() { _ = pw.Close() }) // EOF unblocks the stdin pump
+
+	return &mock.Session{Stdin: pr, Stdout: stdout, Stderr: io.Discard}, stdout, pw
+}
+
+// waitFor polls cond until it holds or a short deadline elapses.
+func waitFor(t *testing.T, cond func() bool) {
+	t.Helper()
+
+	deadline := time.After(2 * time.Second)
+
+	for !cond() {
+		select {
+		case <-deadline:
+			t.Fatal("condition not met within deadline")
+		case <-time.After(5 * time.Millisecond):
+		}
+	}
+}
+
+func TestRunInteractiveForwardsStdinToPort(t *testing.T) {
+	p := &livePort{}
+
+	s := &Serial{Port: "/dev/fake", Baud: DefaultBaudRate}
+	s.open = func(_ string, _ int) (port, error) { return p, nil }
+
+	sess, _, stdin := interactiveSession(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	runErr := make(chan error, 1)
+
+	go func() { runErr <- s.Run(ctx, sess, "-i") }()
+
+	if _, err := stdin.Write([]byte("reboot\n")); err != nil {
+		t.Fatalf("write stdin: %v", err)
+	}
+
+	waitFor(t, func() bool { return p.writtenString() == "reboot\n" })
+
+	cancel()
+
+	select {
+	case err := <-runErr:
+		// Cancellation is the normal end of an interactive session (client
+		// disconnect / quit); see runInteractive.
+		if err != nil {
+			t.Errorf("Run error = %v, want nil on clean end", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+}
+
+// TestRunInteractiveReconnectsOnDeviceLoss verifies an interactive session
+// survives the serial device disappearing, reopens it, and keeps streaming.
+func TestRunInteractiveReconnectsOnDeviceLoss(t *testing.T) {
+	port1 := &livePort{disconnectErr: errDeviceGone}
+	port1.queue([]byte("booting...\n"))
+
+	port2 := &livePort{}
+	port2.queue([]byte("shell> "))
+
+	opens := 0
+
+	s := &Serial{Port: "/dev/ttyUSB0", Baud: DefaultBaudRate}
+	s.open = func(_ string, _ int) (port, error) {
+		opens++
+		if opens == 1 {
+			return port1, nil
+		}
+
+		return port2, nil
+	}
+
+	sess, stdout, _ := interactiveSession(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	runErr := make(chan error, 1)
+
+	go func() { runErr <- s.Run(ctx, sess, "-i") }()
+
+	waitFor(t, func() bool {
+		g := stdout.String()
+
+		return strings.Contains(g, "booting") &&
+			strings.Contains(g, "reconnected") &&
+			strings.Contains(g, "shell>")
+	})
+
+	cancel()
+
+	select {
+	case err := <-runErr:
+		// Cancellation is the normal end of an interactive session (client
+		// disconnect / quit); see runInteractive.
+		if err != nil {
+			t.Errorf("Run error = %v, want nil on clean end", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+
+	if opens < 2 {
+		t.Errorf("open called %d times, want >= 2 (reconnect)", opens)
+	}
+}
+
+// TestRunExpectReconnectsOnDeviceLoss verifies the expect path survives the
+// device disappearing mid-wait, reopens it, and matches the prompt that appears
+// only after the reconnect (the firmware-CI power-cycle case).
+func TestRunExpectReconnectsOnDeviceLoss(t *testing.T) {
+	port1 := &fakePort{reads: [][]byte{[]byte("booting...\n")}, readErr: errDeviceGone}
+	port2 := &fakePort{reads: [][]byte{[]byte("dut login: ")}}
+
+	opens := 0
+
+	s := &Serial{Port: "/dev/ttyUSB0", Baud: DefaultBaudRate}
+	s.open = func(_ string, _ int) (port, error) {
+		opens++
+		if opens == 1 {
+			return port1, nil
+		}
+
+		return port2, nil
+	}
+
+	rec := &recordingSession{}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	if err := s.Run(ctx, rec, "expect", "login:"); err != nil {
+		t.Fatalf("Run returned error, want nil: %v", err)
+	}
+
+	if opens < 2 {
+		t.Errorf("open called %d times, want >= 2 (reconnect)", opens)
+	}
+
+	got := rec.out.String()
+	for _, want := range []string{"booting", "disconnected", "reconnected", "matched"} {
+		if !strings.Contains(got, want) {
+			t.Errorf("output missing %q: %q", want, got)
+		}
+	}
+}
+
+// TestRunReconnectAbortsOnTimeout verifies the expect timeout still fires while
+// the module is waiting for a missing device to come back.
+func TestRunReconnectAbortsOnTimeout(t *testing.T) {
+	port1 := &fakePort{reads: [][]byte{[]byte("hello\n")}, readErr: errDeviceGone}
+
+	opens := 0
+
+	s := &Serial{Port: "/dev/ttyUSB0", Baud: DefaultBaudRate}
+	s.open = func(_ string, _ int) (port, error) {
+		opens++
+		if opens == 1 {
+			return port1, nil
+		}
+
+		return nil, errDeviceGone // device never comes back
+	}
+
+	err := s.Run(context.Background(), &recordingSession{}, "-t", "300ms", "expect", "will-never-appear")
+	if err == nil || !strings.Contains(err.Error(), "timeout") {
+		t.Fatalf("Run error = %v, want timeout while reconnecting", err)
+	}
+}
+
+// TestRunReconnectsWhenDeviceNodeVanishes covers the loss path real hardware
+// hits: a removed adapter surfaces to Read as an idle timeout (NOT a distinct
+// error), so the module must notice the vanished device node and start
+// reconnecting instead of spinning forever on what looks like a benign idle.
+func TestRunReconnectsWhenDeviceNodeVanishes(t *testing.T) {
+	port1 := &fakePort{} // no queued data: idle reads return (0, nil), like a quiet port
+
+	opens := 0
+
+	s := &Serial{Port: "/dev/fake", Baud: DefaultBaudRate}
+	s.deviceLossGrace = 20 * time.Millisecond    // suspect loss quickly in the test
+	s.portPresent = func() bool { return false } // the device node has vanished
+	s.open = func(_ string, _ int) (port, error) {
+		opens++
+		if opens == 1 {
+			return port1, nil // initial open succeeds
+		}
+
+		return nil, errDeviceGone // device stays gone while reconnecting
+	}
+
+	rec := &recordingSession{}
+
+	err := s.Run(context.Background(), rec, "-t", "300ms", "expect", "will-never-appear")
+	if err == nil || !strings.Contains(err.Error(), "timeout") {
+		t.Fatalf("Run error = %v, want timeout", err)
+	}
+
+	if got := rec.out.String(); !strings.Contains(got, "disconnected, waiting to reconnect") {
+		t.Errorf("expected reconnect to start after the device node vanished, got %q", got)
+	}
+}
diff --git a/pkg/module/serial/step.go b/pkg/module/serial/step.go
index af309a7..79ac59e 100644
--- a/pkg/module/serial/step.go
+++ b/pkg/module/serial/step.go
@@ -58,6 +58,7 @@ func truncate(text string, limit int) string {
 type scriptConfig struct {
 	timeout     time.Duration
 	keepEscapes bool
+	interactive bool
 	steps       []step
 }
 
@@ -74,11 +75,13 @@ func parseArgs(args []string) (scriptConfig, error) {
 		timeout     time.Duration
 		eolName     string
 		keepEscapes bool
+		interactive bool
 	)
 
 	fs.DurationVar(&timeout, "t", 0, "global timeout for the whole run (e.g. 30s, 3m); 0 = no timeout")
 	fs.StringVar(&eolName, "eol", defaultEOL, "line ending appended by 'send': cr|lf|crlf|none")
 	fs.BoolVar(&keepEscapes, "keep-escapes", false, "keep terminal escape sequences in the output instead of stripping them")
+	fs.BoolVar(&interactive, "i", false, "interactive: bridge the client console to the serial port (no steps)")
 
 	err := fs.Parse(args)
 	if err != nil {
@@ -95,7 +98,14 @@ func parseArgs(args []string) (scriptConfig, error) {
 		return scriptConfig{}, err
 	}
 
-	return scriptConfig{timeout: timeout, keepEscapes: keepEscapes, steps: steps}, nil
+	// Interactive is a whole-session mode: the client drives the port live, so a
+	// scripted step sequence would have nothing to act on. Reject the combination
+	// rather than silently ignoring one.
+	if interactive && len(steps) > 0 {
+		return scriptConfig{}, fmt.Errorf("interactive mode (-i) takes no steps, got %d", len(steps))
+	}
+
+	return scriptConfig{timeout: timeout, keepEscapes: keepEscapes, interactive: interactive, steps: steps}, nil
 }
 
 // resolveEOL maps the -eol flag value to the bytes appended by a 'send' step.