fix: terminal input and cleanup

backend/pkg/plugin/exec/controller.go

@@ -95,8 +95,14 @@ type controller struct {
 func (c *controller) ServiceStartup(ctx context.Context, options application.ServiceOptions) error {
 	c.app = application.Get()
 	c.ctx = ctx
-	go c.runMux()      // plugin mux
-	go c.runLocalMux() // local terminal should be muxed separately to avoid latency
+
+	// Initialize the terminal manager synchronously so c.terminalManager is
+	// safe to read before any goroutine starts.
+	manager, inMux, outMux, resizeMux := terminal.NewManager(ctx, c.logger)
+	c.terminalManager = manager
+
+	go c.runMux()                                // plugin mux
+	go c.runLocalMux(inMux, outMux, resizeMux)   // local terminal should be muxed separately to avoid latency
 	return nil
 }
 
@@ -117,30 +123,17 @@ func (c *controller) safeSend(ch chan exec.StreamInput, input exec.StreamInput)
 	return nil
 }
 
-func listenOnOut(
-	cancel chan struct{},
-	source chan exec.StreamOutput,
-	target chan exec.StreamOutput,
+func (c *controller) runLocalMux(
+	inMux chan exec.StreamInput,
+	outMux chan exec.StreamOutput,
+	resizeMux chan exec.StreamResize,
 ) {
-	for {
-		select {
-		case <-cancel:
-			return
-		case output := <-source:
-			target <- output
-		}
-	}
-}
-
-func (c *controller) runLocalMux() {
-	manager, inMux, outMux, resizeMux := terminal.NewManager(c.logger)
-	c.terminalManager = manager
 	for {
 		select {
 		case <-c.ctx.Done():
 			return
 		case input := <-inMux:
-			if err := manager.WriteSession(input.SessionID, input.Data); err != nil {
+			if err := c.terminalManager.WriteSession(input.SessionID, input.Data); err != nil {
 				c.logger.Errorw(context.Background(), "error writing to session", "error", err)
 			}
 		case output := <-outMux:
@@ -181,7 +174,7 @@ func (c *controller) runLocalMux() {
 
 			c.app.Event.Emit(eventkey, output.Data)
 		case resize := <-resizeMux:
-			if err := manager.ResizeSession(resize.SessionID, resize.Rows, resize.Cols); err != nil {
+			if err := c.terminalManager.ResizeSession(resize.SessionID, resize.Rows, resize.Cols); err != nil {
 				c.logger.Errorw(context.Background(), "error resizing session", "error", err)
 			}
 		}
@@ -194,7 +187,6 @@ func (c *controller) runMux() {
 		case <-c.ctx.Done():
 			return
 		case input := <-c.inputMux:
-			c.logger.Debugw(context.Background(), "got input", "sessionID", input.SessionID, "payloadSize", len(input.Data))
 			// Capture the channel under lock, then send outside it to avoid
 			// holding RLock during a potentially blocking send.
 			var ch chan exec.StreamInput

backend/pkg/terminal/manager.go

@@ -9,7 +9,6 @@ import (
 	"os"
 	"os/exec"
 	"os/signal"
-	"regexp"
 	"sync"
 	"syscall"
 	"time"
@@ -33,9 +32,11 @@ const (
 
 // Manager manages terminal sessions, allowing creation, attachment, and more.
 type Manager struct {
+	ctx      context.Context
 	log      logging.Logger
 	sessions map[string]*sdkexec.Session
 	ptys     map[string]*os.File
+	cmds     map[string]*exec.Cmd
 	cancels  map[string]context.CancelFunc
 	buffers  map[string]*sdkexec.OutputBuffer
 
@@ -49,22 +50,29 @@ type Manager struct {
 // latency sensitive with the local manager, we're going to directly return
 // the channels for in and out that the exec controller will use.
 func NewManager(
+	ctx context.Context,
 	log logging.Logger,
 ) (*Manager, chan sdkexec.StreamInput, chan sdkexec.StreamOutput, chan sdkexec.StreamResize) {
 	inMux := make(chan sdkexec.StreamInput)
 	outMux := make(chan sdkexec.StreamOutput)
 	resizeMux := make(chan sdkexec.StreamResize)
 
-	return &Manager{
+	mgr := &Manager{
+		ctx:       ctx,
 		log:       log.Named("TerminalManager"),
 		sessions:  make(map[string]*sdkexec.Session),
 		ptys:      make(map[string]*os.File),
+		cmds:      make(map[string]*exec.Cmd),
 		cancels:   make(map[string]context.CancelFunc),
 		buffers:   make(map[string]*sdkexec.OutputBuffer),
 		inMux:     inMux,
 		outMux:    outMux,
 		resizeMux: resizeMux,
-	}, inMux, outMux, resizeMux
+	}
+
+	go mgr.forwardSignals()
+
+	return mgr, inMux, outMux, resizeMux
 }
 
 // GetSession returns a session by its ID.
@@ -80,15 +88,15 @@ func (m *Manager) GetSession(sessionID string) (*sdkexec.Session, error) {
 
 // ListSessions returns a list of details for all active sessions.
 func (m *Manager) ListSessions(_ *types.PluginContext) []*sdkexec.Session {
-	m.mux.Lock()
-	defer m.mux.Unlock()
+	m.mux.RLock()
+	defer m.mux.RUnlock()
 	sessions := make([]*sdkexec.Session, 0, len(m.sessions))
 
 	for _, session := range m.sessions {
 		sessions = append(sessions, session)
 	}
 
-	m.log.Debugw(context.Background(), "listed sessions", "sessions", sessions)
+	m.log.Debugw(context.Background(), "listed sessions", "count", len(sessions))
 	return sessions
 }
 
@@ -98,10 +106,10 @@ func (m *Manager) StartSession(
 	opts sdkexec.SessionOptions,
 ) (*sdkexec.Session, error) {
 	logger := m.log.With(logging.Any("action", "StartSession"))
-	logger.Debugw(context.Background(), "starting session", "options", opts, "context", pCtx)
+	logger.Debugw(context.Background(), "starting session", "command", opts.Command, "tty", opts.TTY)
 
-	// Set up the command to run in a new pseudo-terminal.
-	ctx, cancel := context.WithCancel(context.Background())
+	// Derive from manager context so shutdown cascades to all sessions.
+	ctx, cancel := context.WithCancel(m.ctx)
 
 	// determine the default shell from the commands passed in, since we may want to add flags
 	shell := DefaultLocalShell
@@ -174,9 +182,9 @@ func (m *Manager) StartSession(
 	}
 
 	m.mux.Lock()
-	logger.Debugw(ctx, "past lock")
 	m.sessions[opts.ID] = session
 	m.ptys[opts.ID] = ptyFile
+	m.cmds[opts.ID] = cmd
 	m.cancels[opts.ID] = cancel
 	m.buffers[opts.ID] = sdkexec.NewDefaultOutputBuffer()
 	m.mux.Unlock()
@@ -188,7 +196,7 @@ func (m *Manager) StartSession(
 
 	// Start handling terminal output in a separate goroutine.
 	go m.handleOutStream(ctx, opts.ID, ptyFile)
-	go m.handleSignals(ctx, opts.ID, cmd)
+	go m.handleSessionClose(ctx, opts.ID)
 	go m.handleWaitForCompletion(ctx, opts.ID, cmd)
 
 	return session, nil
@@ -217,47 +225,44 @@ func (m *Manager) handleWaitForCompletion(_ context.Context, sessionID string, c
 	m.terminateSession(sessionID)
 }
 
-func (m *Manager) handleSignals(ctx context.Context, sessionID string, cmd *exec.Cmd) {
+// forwardSignals listens for host-process signals once and forwards them to
+// all active terminal child processes. Runs for the lifetime of the Manager.
+func (m *Manager) forwardSignals() {
 	ch := make(chan os.Signal, 1)
-	signal.Notify(ch, syscall.SIGTERM)
-	signal.Notify(ch, syscall.SIGINT)
-	signal.Notify(ch, syscall.SIGQUIT)
-
-	defer func() { signal.Stop(ch); close(ch) }()
+	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)
+	defer signal.Stop(ch)
 
 	for {
 		select {
 		case sig := <-ch:
-			switch sig {
-			case syscall.SIGTERM:
-				m.log.Debugw(ctx, "SIGTERM received")
-				cmd.Process.Signal(syscall.SIGTERM)
-			case syscall.SIGINT:
-				m.log.Debugw(ctx, "SIGINT received")
-				cmd.Process.Signal(syscall.SIGINT)
-			case syscall.SIGQUIT:
-				m.log.Debugw(ctx, "SIGQUIT received")
-				cmd.Process.Signal(syscall.SIGQUIT)
-			}
-		case <-ctx.Done():
-			m.log.Debugw(ctx,
-				"context cancelled, stopping signal handling",
-				"session", sessionID,
-			)
-
-			// signal to ide we're done
-			m.outMux <- sdkexec.StreamOutput{
-				SessionID: sessionID,
-				Target:    sdkexec.StreamTargetStdOut,
-				Data:      []byte("Session terminated"),
-				Signal:    sdkexec.StreamSignalClose,
+			m.mux.RLock()
+			for sid, cmd := range m.cmds {
+				if cmd.Process != nil {
+					if err := cmd.Process.Signal(sig); err != nil {
+						m.log.Debugw(context.Background(), "failed to forward signal",
+							"signal", sig, "session", sid, "error", err)
+					}
+				}
 			}
-
+			m.mux.RUnlock()
+		case <-m.ctx.Done():
 			return
 		}
 	}
 }
 
+// handleSessionClose waits for a session's context to be cancelled and emits
+// a CLOSE signal to the frontend.
+func (m *Manager) handleSessionClose(ctx context.Context, sessionID string) {
+	<-ctx.Done()
+	m.outMux <- sdkexec.StreamOutput{
+		SessionID: sessionID,
+		Target:    sdkexec.StreamTargetStdOut,
+		Data:      []byte("Session terminated"),
+		Signal:    sdkexec.StreamSignalClose,
+	}
+}
+
 func (m *Manager) handleOutStream(
 	_ context.Context,
 	sessionID string,
@@ -326,14 +331,6 @@ func (m *Manager) writeToSession(sessionID string, bytes []byte) error {
 	return nil
 }
 
-// cleanPTYOutput removes the `%` symbol and its associated escape sequences.
-func cleanPTYOutput(output string) string {
-	// Define a regex pattern to match the escape sequence for `%`
-	pattern := `\x1b\[1m\x1b\[7m%\x1b\[27m\x1b\[1m\x1b\[0m`
-	re := regexp.MustCompile(pattern)
-	return re.ReplaceAllString(output, "")
-}
-
 // WriteSession writes data to the session's input.
 func (m *Manager) WriteSession(sessionID string, input []byte) error {
 	return m.writeToSession(sessionID, input)
@@ -356,7 +353,7 @@ func (m *Manager) AttachSession(sessionID string) (*sdkexec.Session, []byte, err
 	if buffer != nil {
 		data = buffer.GetAll()
 	}
-	m.log.Debugw(context.Background(), fmt.Sprintf("session buffer data: %q", data))
+	m.log.Debugw(context.Background(), "session buffer loaded", "session", sessionID, "bufferSize", len(data))
 
 	// pointer, no need to reassign
 	session.Attached = true
@@ -422,6 +419,7 @@ func (m *Manager) terminateSessionLocked(sessionID string) {
 	}
 	delete(m.sessions, sessionID)
 	delete(m.ptys, sessionID)
+	delete(m.cmds, sessionID)
 	delete(m.cancels, sessionID)
 	delete(m.buffers, sessionID)
 	m.log.Debugw(context.Background(), "session terminated", "session", sessionID)

backend/pkg/terminal/manager_windows.go

@@ -3,6 +3,7 @@
 package terminal
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"sync"
@@ -28,6 +29,7 @@ type Manager struct {
 }
 
 func NewManager(
+	_ context.Context,
 	log logging.Logger,
 ) (*Manager, chan sdkexec.StreamInput, chan sdkexec.StreamOutput, chan sdkexec.StreamResize) {
 	inMux := make(chan sdkexec.StreamInput)

ui/providers/BottomDrawer/containers/Terminal.tsx

@@ -319,7 +319,7 @@ export default function TerminalContainer({ sessionId, tab }: Props) {
       attachedRef.current = true;
       fitAddon.fit()
       terminal.onData(data => {
-        ExecClient.WriteSession(sessionId, data)
+        ExecClient.WriteSession(sessionId, Base64.encode(data))
           .catch((err: unknown) => {
             log.error(new Error(parseAppError(err).detail), { event: 'write_session', sessionId });
           });