diff --git a/backend/pkg/plugin/exec/controller.go b/backend/pkg/plugin/exec/controller.go index 6628111f..673322fc 100644 --- a/backend/pkg/plugin/exec/controller.go +++ b/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 diff --git a/backend/pkg/terminal/manager.go b/backend/pkg/terminal/manager.go index f26f3a77..31432832 100644 --- a/backend/pkg/terminal/manager.go +++ b/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) diff --git a/backend/pkg/terminal/manager_windows.go b/backend/pkg/terminal/manager_windows.go index c50681b5..2dcb56de 100644 --- a/backend/pkg/terminal/manager_windows.go +++ b/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) diff --git a/ui/providers/BottomDrawer/containers/Terminal.tsx b/ui/providers/BottomDrawer/containers/Terminal.tsx index b7d3f750..4a077cdb 100644 --- a/ui/providers/BottomDrawer/containers/Terminal.tsx +++ b/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 }); });