From 5ed618494d5a3dfceb4aca4a75ca98c97af919d1 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:05:47 +0800 Subject: [PATCH] feat(shell): run bash on a real PTY via creack/pty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The shell manager used os.Pipe() for stdin/stdout/stderr, so bash ran without a tty. Programs that check isatty() (python, htop, less, vim-style REPLs) would drop echo, ignore line-editing, and leak literal 0x1b bytes into the program — Python's REPL reported 'invalid non-printable character U+001B' on input. Switch to github.com/creack/pty:StartWithSize with an 80x24 default and TERM=xterm-256color in cmd.Env. The PTY file replaces both the stdin and stdout pipes; sess.Stdin is the pty *os.File and the output capture goroutine reads from it directly. waitExit no longer needs to close sess.Stdin — the capture goroutine's defer handles it when the process dies and the pty returns EOF. - go.mod / go.sum: add creack/pty v1.1.24. - internal/shell/manager.go: rewrite Start to use pty.StartWithSize; captureOutput reads from the pty; waitExit simplified; race-clean exited map to avoid the cmd.ProcessState data race. - internal/shell/manager_test.go (new): TestShellPTYIsRealTerminal (echo with echo + output interleaved proves PTY echo is on) and TestShellPTYTermEnv (TERM=xterm-256color propagates). E2E against the real bash: 'python3 -q' returns the >>> prompt, print(2+2) returns 4, exit() returns to bash, 0 ESC bytes leak. Conversation: 019f35fa-601b-73c1-bd9e-221971c31a56 --- go.mod | 1 + go.sum | 2 + internal/shell/manager.go | 116 ++++++++++++++++++--------------- internal/shell/manager_test.go | 86 ++++++++++++++++++++++++ 4 files changed, 152 insertions(+), 53 deletions(-) create mode 100644 internal/shell/manager_test.go diff --git a/go.mod b/go.mod index c821c5f..8f86493 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module codespace go 1.25.0 require ( + github.com/creack/pty v1.1.24 github.com/gin-gonic/gin v1.12.0 github.com/gorilla/websocket v1.5.3 gopkg.in/yaml.v3 v3.0.1 diff --git a/go.sum b/go.sum index f3880d6..4c21633 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,8 @@ github.com/bytedance/sonic/loader v0.5.1 h1:Ygpfa9zwRCCKSlrp5bBP/b/Xzc3VxsAW+5NI github.com/bytedance/sonic/loader v0.5.1/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= github.com/cloudwego/base64x v0.1.7 h1:NppS+Fgzg5ovhn4NkUXaDT3x9jldgH5ToMCqzBSi2zI= github.com/cloudwego/base64x v0.1.7/go.mod h1:Cu1PV9zfrSf7ET2tIbWbbEy7jO7HHJ13q4X2SQ8aWYg= +github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= +github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/shell/manager.go b/internal/shell/manager.go index 2cf6da2..6b48b80 100644 --- a/internal/shell/manager.go +++ b/internal/shell/manager.go @@ -9,6 +9,8 @@ import ( "syscall" "codespace/internal/util" + + "github.com/creack/pty" ) const ( @@ -35,6 +37,8 @@ type LocalManager struct { args []string mu sync.Mutex sessions map[string]*Session + exitedMu sync.Mutex + exited map[string]struct{} } // NewManager creates a LocalManager with the given command and arguments. @@ -44,6 +48,7 @@ func NewManager(command string, args []string) *LocalManager { command: normalizeCommand(command), args: args, sessions: make(map[string]*Session), + exited: make(map[string]struct{}), } } @@ -60,69 +65,57 @@ func (m *LocalManager) Start(workspaceID string, workspaceRoot string) error { m.mu.Lock() defer m.mu.Unlock() - if sess, ok := m.sessions[workspaceID]; ok && sess.Cmd.Process != nil { - if sess.Cmd.ProcessState == nil { + if _, ok := m.sessions[workspaceID]; ok { + m.exitedMu.Lock() + _, exited := m.exited[workspaceID] + m.exitedMu.Unlock() + if !exited { return util.New(util.CodeConflict, "shell already running") } } cmd := exec.Command(m.command, m.args...) cmd.Dir = workspaceRoot - cmd.Env = append(os.Environ(), fmt.Sprintf("HOME=%s", workspaceRoot)) + cmd.Env = append(os.Environ(), + "TERM=xterm-256color", + fmt.Sprintf("HOME=%s", workspaceRoot), + ) - stdinR, stdinW, err := os.Pipe() + ptyF, err := pty.StartWithSize(cmd, &pty.Winsize{Rows: 24, Cols: 80, X: 0, Y: 0}) if err != nil { - return util.Wrap(util.CodeInternal, "failed to create stdin pipe", err) + return util.Wrap(util.CodeInternal, "failed to start shell on pty", err) } - stdoutR, stdoutW, err := os.Pipe() - if err != nil { - stdinR.Close() - stdinW.Close() - return util.Wrap(util.CodeInternal, "failed to create stdout pipe", err) - } - - cmd.Stdin = stdinR - cmd.Stdout = stdoutW - cmd.Stderr = stdoutW - - if err := cmd.Start(); err != nil { - stdinR.Close() - stdinW.Close() - stdoutR.Close() - stdoutW.Close() - return util.Wrap(util.CodeInternal, "failed to start shell", err) - } - - stdinR.Close() - stdoutW.Close() - sess := &Session{ WorkspaceID: workspaceID, Root: workspaceRoot, Cmd: cmd, - Stdin: stdinW, + Stdin: ptyF, Subscribers: make(map[Subscription]struct{}), } m.sessions[workspaceID] = sess + m.exitedMu.Lock() + delete(m.exited, workspaceID) + m.exitedMu.Unlock() + outputDone := make(chan struct{}) - go m.captureOutput(sess, stdoutR, outputDone) + go m.captureOutput(sess, ptyF, outputDone) go m.waitExit(sess, outputDone) return nil } -// captureOutput reads from the merged stdout/stderr pipe and fans out each -// chunk to all subscribers using non-blocking sends. -func (m *LocalManager) captureOutput(sess *Session, stdoutR *os.File, done chan<- struct{}) { +// captureOutput reads from the PTY master and fans out each chunk to all +// subscribers using non-blocking sends. +func (m *LocalManager) captureOutput(sess *Session, ptyF *os.File, done chan<- struct{}) { defer close(done) - defer stdoutR.Close() + defer ptyF.Close() buf := make([]byte, outputBufferSize) for { - n, err := stdoutR.Read(buf) + n, err := ptyF.Read(buf) if n > 0 { chunk := make([]byte, n) copy(chunk, buf[:n]) @@ -135,19 +128,24 @@ func (m *LocalManager) captureOutput(sess *Session, stdoutR *os.File, done chan< } if err != nil { if err != io.EOF { - // Ignore read errors; the pipe is closing. + // Ignore read errors; the PTY is closing. } break } } } -// waitExit waits for the shell to exit, then records the exit status, closes -// the stdin writer, and closes every subscriber channel exactly once. +// waitExit waits for the shell to exit, then records the exit status and +// closes every subscriber channel exactly once. The PTY master is closed by +// captureOutput once reads finish. func (m *LocalManager) waitExit(sess *Session, outputDone <-chan struct{}) { _ = sess.Cmd.Wait() <-outputDone + m.exitedMu.Lock() + m.exited[sess.WorkspaceID] = struct{}{} + m.exitedMu.Unlock() + sess.mu.Lock() defer sess.mu.Unlock() @@ -158,10 +156,6 @@ func (m *LocalManager) waitExit(sess *Session, outputDone <-chan struct{}) { } } - if sess.Stdin != nil { - _ = sess.Stdin.Close() - } - for sub := range sess.Subscribers { sub.(*subscription).closeChan() } @@ -174,26 +168,35 @@ func (m *LocalManager) Stop(workspaceID string) error { defer m.mu.Unlock() sess, ok := m.sessions[workspaceID] - if !ok || sess.Cmd.Process == nil { + if !ok { return util.New(util.CodeNotFound, "no running shell for workspace") } - if err := sess.Cmd.Process.Kill(); err != nil { - return util.Wrap(util.CodeInternal, "failed to stop shell", err) + if sess.Cmd.Process != nil { + _ = sess.Cmd.Process.Kill() } delete(m.sessions, workspaceID) + m.exitedMu.Lock() + delete(m.exited, workspaceID) + m.exitedMu.Unlock() return nil } // Restart stops (if running) then starts the shell. func (m *LocalManager) Restart(workspaceID string, workspaceRoot string) error { m.mu.Lock() - if sess, ok := m.sessions[workspaceID]; ok && sess.Cmd.Process != nil { - if sess.Cmd.ProcessState == nil { + if sess, ok := m.sessions[workspaceID]; ok { + m.exitedMu.Lock() + _, exited := m.exited[workspaceID] + m.exitedMu.Unlock() + if !exited && sess.Cmd.Process != nil { _ = sess.Cmd.Process.Kill() - delete(m.sessions, workspaceID) } + delete(m.sessions, workspaceID) + m.exitedMu.Lock() + delete(m.exited, workspaceID) + m.exitedMu.Unlock() } m.mu.Unlock() @@ -203,14 +206,17 @@ func (m *LocalManager) Restart(workspaceID string, workspaceRoot string) error { // Status returns the current shell status for the workspace. func (m *LocalManager) Status(workspaceID string) Status { m.mu.Lock() - defer m.mu.Unlock() - sess, ok := m.sessions[workspaceID] - if !ok || sess.Cmd.Process == nil { + m.mu.Unlock() + + if !ok { return Status{WorkspaceID: workspaceID, Running: false} } - if sess.Cmd.ProcessState != nil { + m.exitedMu.Lock() + _, exited := m.exited[workspaceID] + m.exitedMu.Unlock() + if exited { return Status{WorkspaceID: workspaceID, Running: false} } @@ -235,10 +241,14 @@ func (m *LocalManager) Subscribe(workspaceID string) (Subscription, error) { sess.mu.Lock() sub := newSubscription(sess) sess.Subscribers[sub] = struct{}{} - if sess.Cmd.ProcessState != nil { + sess.mu.Unlock() + + m.exitedMu.Lock() + _, exited := m.exited[workspaceID] + m.exitedMu.Unlock() + if exited { sub.closeChan() } - sess.mu.Unlock() return sub, nil } diff --git a/internal/shell/manager_test.go b/internal/shell/manager_test.go new file mode 100644 index 0000000..642f30b --- /dev/null +++ b/internal/shell/manager_test.go @@ -0,0 +1,86 @@ +package shell + +import ( + "bytes" + "context" + "io" + "strings" + "testing" + "time" +) + +func startTestShell(t *testing.T) (*LocalManager, io.WriteCloser, Subscription) { + t.Helper() + mgr := NewManager("bash", []string{"-i"}) + root := t.TempDir() + if err := mgr.Start("test-ws", root); err != nil { + t.Fatalf("Start failed: %v", err) + } + + stdin, err := mgr.Stdin("test-ws") + if err != nil { + t.Fatalf("Stdin failed: %v", err) + } + + sub, err := mgr.Subscribe("test-ws") + if err != nil { + t.Fatalf("Subscribe failed: %v", err) + } + + t.Cleanup(func() { + sub.Close() + _ = mgr.Stop("test-ws") + }) + + return mgr, stdin, sub +} + +func waitForMarker(t *testing.T, sub Subscription, stdin io.Writer, input, marker string) string { + t.Helper() + + if _, err := stdin.Write([]byte(input)); err != nil { + t.Fatalf("failed to write to stdin: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var out bytes.Buffer + for { + select { + case chunk, ok := <-sub.Output(): + if !ok { + t.Fatalf("subscription closed before seeing %q; output so far:\n%s", marker, out.String()) + } + out.Write(chunk) + if strings.Contains(out.String(), marker) { + return out.String() + } + case <-ctx.Done(): + t.Fatalf("timed out waiting for %q; output so far:\n%s", marker, out.String()) + } + } +} + +func TestShellPTYIsRealTerminal(t *testing.T) { + mgr, stdin, sub := startTestShell(t) + _ = mgr + + out := waitForMarker(t, sub, stdin, "echo HELLO_PTY_TEST\n", "HELLO_PTY_TEST") + + if !strings.Contains(out, "echo HELLO_PTY_TEST") { + t.Errorf("expected command echo in PTY output, got:\n%s", out) + } + if !strings.Contains(out, "HELLO_PTY_TEST") { + t.Errorf("expected command output in PTY output, got:\n%s", out) + } +} + +func TestShellPTYTermEnv(t *testing.T) { + _, stdin, sub := startTestShell(t) + + out := waitForMarker(t, sub, stdin, "echo \"$TERM\"\n", "xterm-256color") + if !strings.Contains(out, "xterm-256color") { + t.Errorf("expected TERM=xterm-256color in output, got:\n%s", out) + } +}