fix(terminal): stop replayed output from echoing escape-sequence replies into the shell

server/internal/handler/terminal.go

@@ -16,8 +16,13 @@ import (
 // between the client (xterm.js) and a PTY session attached to devpod ssh.
 //
 // Wire protocol:
-//   - 0x00 + data: terminal I/O (stdin from client, stdout to client)
+//   - 0x00 + data: live terminal I/O (stdin from client, stdout to client)
 //   - 0x01 + JSON: control messages (e.g., {"cols":80,"rows":24} for resize)
+//   - 0x02 + data: replayed historical output (server → client only)
+//   - 0x03:        replay complete (server → client only, empty payload)
+//
+// 0x02/0x03 exist so the client can render scrollback without answering
+// terminal queries captured in it. See terminal.Client for why.
 func (h *Handler) HandleTerminalWebSocket(w http.ResponseWriter, r *http.Request) {
 	sessionIDStr := chi.URLParam(r, "sessionID")
 	sessionID, err := uuid.Parse(sessionIDStr)
@@ -115,20 +120,35 @@ func (h *Handler) HandleTerminalWebSocket(w http.ResponseWriter, r *http.Request
 	}
 }
 
-// wsWriter wraps a WebSocket connection as an io.Writer, prepending
-// the 0x00 data prefix to each write.
+// wsWriter adapts a WebSocket connection to terminal.Client, prefixing
+// each payload with the frame byte that identifies its kind.
 type wsWriter struct {
 	conn *websocket.Conn
 	ctx  context.Context
 }
 
-func (w *wsWriter) Write(p []byte) (int, error) {
+// frame sends a single prefixed binary message.
+func (w *wsWriter) frame(prefix byte, p []byte) error {
 	msg := make([]byte, 1+len(p))
-	msg[0] = 0x00
+	msg[0] = prefix
 	copy(msg[1:], p)
-	err := w.conn.Write(w.ctx, websocket.MessageBinary, msg)
-	if err != nil {
+	return w.conn.Write(w.ctx, websocket.MessageBinary, msg)
+}
+
+// Write sends live PTY output.
+func (w *wsWriter) Write(p []byte) (int, error) {
+	if err := w.frame(0x00, p); err != nil {
 		return 0, err
 	}
 	return len(p), nil
 }
+
+// WriteReplay sends buffered historical output.
+func (w *wsWriter) WriteReplay(p []byte) error {
+	return w.frame(0x02, p)
+}
+
+// ReplayComplete tells the client it may resume responding to the terminal.
+func (w *wsWriter) ReplayComplete() error {
+	return w.frame(0x03, nil)
+}

server/internal/terminal/manager.go

@@ -15,13 +15,32 @@ import (
 // keeping memory per terminal session bounded.
 const replayBufferSize = 100 * 1024
 
+// Client receives PTY output. Replayed (historical) output is delivered
+// separately from live output so the client can tell the two apart.
+//
+// This distinction is load-bearing, not cosmetic. The replay buffer stores
+// raw PTY bytes, which may include terminal *queries* the remote shell
+// emitted earlier (OSC 11 background-colour, DA, DSR, ...). A terminal
+// emulator receiving those queries answers them, and the answer would be
+// written back to PTY stdin — landing in the shell's line buffer as garbage.
+// Framing replay distinctly lets the client suppress responses until the
+// replay boundary has passed.
+type Client interface {
+	// Write delivers live PTY output.
+	io.Writer
+	// WriteReplay delivers buffered historical output.
+	WriteReplay(p []byte) error
+	// ReplayComplete signals that no further replayed output will arrive.
+	ReplayComplete() error
+}
+
 // Session represents a single PTY session attached to a devpod ssh process.
 type Session struct {
 	ptmx *os.File
 	cmd  *exec.Cmd
 
 	mu      sync.Mutex
-	clients map[io.Writer]bool
+	clients map[Client]bool
 	replay  []byte        // recent PTY output, replayed to new clients
 	done    chan struct{} // closed when the reader goroutine exits
 }
@@ -59,7 +78,7 @@ func (m *Manager) GetOrCreate(sessionID string, cmdFactory func() *exec.Cmd) (*S
 	s := &Session{
 		ptmx:    ptmx,
 		cmd:     cmd,
-		clients: make(map[io.Writer]bool),
+		clients: make(map[Client]bool),
 		done:    make(chan struct{}),
 	}
 	m.sessions[sessionID] = s
@@ -127,26 +146,37 @@ func (s *Session) appendReplay(data []byte) {
 	}
 }
 
-// AddClient registers a writer to receive PTY output and replays the
-// recent buffer so the client doesn't land on a blank terminal.
-// The replay write happens under the session lock to keep ordering
-// consistent with concurrent readLoop fan-out.
-func (s *Session) AddClient(w io.Writer) {
+// AddClient registers a client to receive PTY output and replays the
+// recent buffer so it doesn't land on a blank terminal.
+//
+// The replay write, the replay-complete marker, and the registration all
+// happen under a single hold of the session lock. That ordering matters:
+// if the marker were sent after the lock was released, live output could
+// interleave ahead of it, and the client would suppress the response to a
+// genuinely live query.
+//
+// ReplayComplete is sent unconditionally, including when the buffer is
+// empty — a client that never receives it has no way to know replay is
+// over, and would stay muted until its own fallback timer expires.
+func (s *Session) AddClient(c Client) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if len(s.replay) > 0 {
-		if _, err := w.Write(s.replay); err != nil {
+		if err := c.WriteReplay(s.replay); err != nil {
 			return
 		}
 	}
-	s.clients[w] = true
+	if err := c.ReplayComplete(); err != nil {
+		return
+	}
+	s.clients[c] = true
 }
 
-// RemoveClient unregisters a writer from PTY output.
-func (s *Session) RemoveClient(w io.Writer) {
+// RemoveClient unregisters a client from PTY output.
+func (s *Session) RemoveClient(c Client) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	delete(s.clients, w)
+	delete(s.clients, c)
 }
 
 // Done returns a channel that is closed when the PTY process exits.

server/internal/terminal/manager_test.go

@@ -0,0 +1,251 @@
+package terminal
+
+import (
+	"bytes"
+	"errors"
+	"os"
+	"sync"
+	"testing"
+	"time"
+)
+
+// fakeClient records the calls a Session makes on it, in order, so tests can
+// assert on both the content and the sequencing of replay vs. live delivery.
+type fakeClient struct {
+	mu sync.Mutex
+
+	live     [][]byte
+	replay   [][]byte
+	complete int
+	calls    []string // ordered call log: "replay", "complete", "live"
+
+	replayErr   error
+	completeErr error
+}
+
+func (c *fakeClient) Write(p []byte) (int, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.live = append(c.live, append([]byte(nil), p...))
+	c.calls = append(c.calls, "live")
+	return len(p), nil
+}
+
+func (c *fakeClient) WriteReplay(p []byte) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.replayErr != nil {
+		return c.replayErr
+	}
+	c.replay = append(c.replay, append([]byte(nil), p...))
+	c.calls = append(c.calls, "replay")
+	return nil
+}
+
+func (c *fakeClient) ReplayComplete() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.completeErr != nil {
+		return c.completeErr
+	}
+	c.complete++
+	c.calls = append(c.calls, "complete")
+	return nil
+}
+
+func (c *fakeClient) snapshot() ([][]byte, [][]byte, int, []string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.live, c.replay, c.complete, append([]string(nil), c.calls...)
+}
+
+// newTestSession wires a Session to an os.Pipe standing in for the PTY, so
+// tests exercise the real readLoop fan-out rather than a reimplementation.
+// Writing to the returned *os.File simulates shell output.
+func newTestSession(t *testing.T) (*Session, *os.File) {
+	t.Helper()
+
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("os.Pipe: %v", err)
+	}
+
+	s := &Session{
+		ptmx:    r,
+		clients: make(map[Client]bool),
+		done:    make(chan struct{}),
+	}
+	go s.readLoop("test")
+
+	t.Cleanup(func() {
+		w.Close()
+		<-s.done
+		r.Close()
+	})
+	return s, w
+}
+
+// emit writes to the fake PTY and waits for readLoop to fan it out.
+func emit(t *testing.T, s *Session, w *os.File, data string) {
+	t.Helper()
+	if _, err := w.Write([]byte(data)); err != nil {
+		t.Fatalf("write to fake pty: %v", err)
+	}
+	waitForReplay(t, s, len(data))
+}
+
+// waitForReplay blocks until the session's replay buffer has grown to at
+// least n bytes, so tests don't race the readLoop goroutine.
+func waitForReplay(t *testing.T, s *Session, n int) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		s.mu.Lock()
+		got := len(s.replay)
+		s.mu.Unlock()
+		if got >= n {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %d replay bytes", n)
+}
+
+func TestAddClientReplaysThenMarksComplete(t *testing.T) {
+	s, w := newTestSession(t)
+	emit(t, s, w, "hello from the shell")
+
+	c := &fakeClient{}
+	s.AddClient(c)
+
+	_, replay, complete, calls := c.snapshot()
+	if len(replay) != 1 || !bytes.Equal(replay[0], []byte("hello from the shell")) {
+		t.Errorf("replay = %q, want one chunk of the buffered output", replay)
+	}
+	if complete != 1 {
+		t.Errorf("ReplayComplete called %d times, want 1", complete)
+	}
+	// Ordering is the whole point: history, then the boundary, then live.
+	if len(calls) != 2 || calls[0] != "replay" || calls[1] != "complete" {
+		t.Errorf("call order = %v, want [replay complete]", calls)
+	}
+}
+
+func TestAddClientMarksCompleteWithEmptyReplayBuffer(t *testing.T) {
+	// The first client on a fresh PTY has nothing to replay, but still needs
+	// the boundary marker — otherwise it stays muted until its fallback fires.
+	s, _ := newTestSession(t)
+
+	c := &fakeClient{}
+	s.AddClient(c)
+
+	_, replay, complete, _ := c.snapshot()
+	if len(replay) != 0 {
+		t.Errorf("WriteReplay called with empty buffer: %q", replay)
+	}
+	if complete != 1 {
+		t.Errorf("ReplayComplete called %d times, want 1", complete)
+	}
+}
+
+func TestOutputAfterRegistrationGoesToLiveWrite(t *testing.T) {
+	s, w := newTestSession(t)
+	emit(t, s, w, "old")
+
+	c := &fakeClient{}
+	s.AddClient(c)
+	emit(t, s, w, "new")
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if live, _, _, _ := c.snapshot(); len(live) > 0 {
+			break
+		}
+		time.Sleep(time.Millisecond)
+	}
+
+	live, replay, _, calls := c.snapshot()
+	if len(live) != 1 || !bytes.Equal(live[0], []byte("new")) {
+		t.Errorf("live = %q, want one chunk %q", live, "new")
+	}
+	if len(replay) != 1 || !bytes.Equal(replay[0], []byte("old")) {
+		t.Errorf("replay = %q, want only the pre-registration output", replay)
+	}
+	if len(calls) != 3 || calls[2] != "live" {
+		t.Errorf("call order = %v, want live delivery last", calls)
+	}
+}
+
+func TestAddClientSkipsRegistrationWhenReplayFails(t *testing.T) {
+	s, w := newTestSession(t)
+	emit(t, s, w, "old")
+
+	c := &fakeClient{replayErr: errors.New("connection gone")}
+	s.AddClient(c)
+
+	if _, _, complete, _ := c.snapshot(); complete != 0 {
+		t.Errorf("ReplayComplete called after replay failure")
+	}
+	emit(t, s, w, "new")
+
+	if live, _, _, _ := c.snapshot(); len(live) != 0 {
+		t.Errorf("live output delivered to unregistered client: %q", live)
+	}
+}
+
+func TestAddClientSkipsRegistrationWhenCompleteFails(t *testing.T) {
+	s, w := newTestSession(t)
+
+	c := &fakeClient{completeErr: errors.New("connection gone")}
+	s.AddClient(c)
+	emit(t, s, w, "new")
+
+	if live, _, _, _ := c.snapshot(); len(live) != 0 {
+		t.Errorf("live output delivered to unregistered client: %q", live)
+	}
+}
+
+func TestSecondClientReplayIncludesOutputSinceFirstAttached(t *testing.T) {
+	s, w := newTestSession(t)
+	emit(t, s, w, "first")
+
+	c1 := &fakeClient{}
+	s.AddClient(c1)
+	emit(t, s, w, "second")
+
+	c2 := &fakeClient{}
+	s.AddClient(c2)
+
+	_, replay, complete, _ := c2.snapshot()
+	if len(replay) != 1 || !bytes.Equal(replay[0], []byte("firstsecond")) {
+		t.Errorf("second client replay = %q, want the full buffer to date", replay)
+	}
+	if complete != 1 {
+		t.Errorf("second client ReplayComplete called %d times, want 1", complete)
+	}
+	// The first client must not be re-replayed when a second one attaches.
+	if _, r1, n1, _ := c1.snapshot(); len(r1) != 1 || n1 != 1 {
+		t.Errorf("first client saw replay=%d complete=%d, want 1 and 1", len(r1), n1)
+	}
+}
+
+func TestAppendReplayTrimsToMostRecentBytes(t *testing.T) {
+	s := &Session{clients: make(map[Client]bool), done: make(chan struct{})}
+
+	// Marker bytes are disjoint from the filler so counting stays unambiguous.
+	const marker = "END!"
+	s.appendReplay(bytes.Repeat([]byte("a"), replayBufferSize))
+	s.appendReplay([]byte(marker))
+
+	if len(s.replay) != replayBufferSize {
+		t.Fatalf("replay length = %d, want %d", len(s.replay), replayBufferSize)
+	}
+	if !bytes.HasSuffix(s.replay, []byte(marker)) {
+		t.Errorf("replay lost the most recent bytes")
+	}
+	// The trim must drop from the front: exactly len(marker) of the original
+	// filler should be gone, not an equivalent amount from the recent end.
+	if got := bytes.Count(s.replay, []byte("a")); got != replayBufferSize-len(marker) {
+		t.Errorf("filler retained = %d bytes, want %d", got, replayBufferSize-len(marker))
+	}
+}