fix: prevent Mosh character ghosting caused by overlapping diffs

- client.go: track last output oldNum, replace instead of append when same base
- transport.go: reject diffs that do not chain from latest acknowledged state
  (oldNum != ackNum), preventing stale ANSI sequences from corrupting output

README.md

@@ -10,6 +10,9 @@ Implements the full mosh wire protocol: AES-128-OCB3 authenticated
 encryption, SSP transport with sequencing and retransmission, datagram
 fragmentation, and the protobuf state sync layer.
 
+[![Watch the demo](demo.png)](https://www.youtube.com/watch?v=iWo9I2bjZkI)
+*[unixshells.com](https://unixshells.com) uses mosh-go compiled to WASM for the web terminal*
+
 ## Install
 
 ```
@@ -95,6 +98,16 @@ Standard mosh clients and servers ignore them (forward-compatible).
 - **Session control** (Host/UserInstruction field 9): list, switch,
   and create latch sessions over the mosh connection.
 
+## WASM build
+
+The mosh client compiles to WebAssembly for use in the web terminal:
+
+```
+GOOS=js GOARCH=wasm CGO_ENABLED=0 go build -tags osusergo -o mosh.wasm ./cmd/mosh-wasm/
+```
+
+The `-tags osusergo` flag is required to avoid `os/user` CGO issues with `GOOS=js`.
+
 ## Files
 
 - `server.go` -- server, PTY lifecycle, `ServeRW` bridge mode
@@ -104,10 +117,11 @@ Standard mosh clients and servers ignore them (forward-compatible).
 - `fragment.go` -- datagram fragmentation and reassembly
 - `pb.go` -- protobuf encoding (transport, host, user, latch extensions)
 - `framebuffer.go` -- terminal state tracking for diff protocol
+- `predict.go` -- local echo prediction engine (speculative keystroke rendering)
 
 ## Dependencies
 
-`creack/pty` and the Go standard library.
+`creack/pty`, `golang.org/x/crypto`, and the Go standard library.
 
 ## License
 

client.go

@@ -12,7 +12,7 @@ import (
 // Client is a mosh client. It connects to a mosh server over UDP,
 // handles the SSP transport, and provides send/recv for terminal I/O.
 type Client struct {
-	conn      *net.UDPConn
+	conn      Conn
 	transport *Transport
 	ocb       *OCB
 
@@ -20,12 +20,20 @@ type Client struct {
 	output  []byte // accumulated terminal output
 	outputC chan struct{}
 
+	// Action tracking for cumulative diffs.
+	actionsMu        sync.Mutex
+	actions          []UserInstruction
+	ackedActionCount int                // how many actions the server has
+	lastAcked        uint64
+	sentActionCounts map[uint64]int     // sentNum → action count at that state
+	dirty            bool               // true = new actions since last tick
+
 	done chan struct{}
 	wg   sync.WaitGroup
 }
 
-// Dial connects to a mosh server. The key is the base64-encoded mosh key
-// (with or without padding).
+// Dial connects to a mosh server over UDP. The key is the base64-encoded
+// mosh key (with or without padding).
 func Dial(host string, port int, key string) (*Client, error) {
 	// Pad key for base64 if needed.
 	for len(key)%4 != 0 {
@@ -49,41 +57,114 @@ func Dial(host string, port int, key string) (*Client, error) {
 		return nil, err
 	}
 
+	return DialConn(conn, ocb)
+}
+
+// DialConn creates a mosh client over an existing datagram connection.
+// Use this with WebTransport or other non-UDP transports.
+func DialConn(conn Conn, ocb *OCB) (*Client, error) {
 	c := &Client{
-		conn:      conn,
-		transport: NewTransport(ocb, false),
-		ocb:       ocb,
-		outputC:   make(chan struct{}, 1),
-		done:      make(chan struct{}),
+		conn:             conn,
+		transport:        NewTransport(ocb, false),
+		ocb:              ocb,
+		outputC:          make(chan struct{}, 1),
+		done:             make(chan struct{}),
+		sentActionCounts: make(map[uint64]int),
 	}
 
 	c.wg.Add(2)
 	go c.recvLoop()
 	go c.sendLoop()
 
-	// Send initial empty datagram to associate with the server.
+	// Send initial keepalive to associate with the server.
+	c.transport.ForceNextSend()
 	c.tick()
 
 	return c, nil
 }
 
+// DialConnManual creates a mosh client without an internal sendLoop.
+// The caller must call Tick() periodically (e.g., every 8-16ms) to flush
+// outgoing datagrams. Use this in WASM where Go timers are unreliable.
+func DialConnManual(conn Conn, ocb *OCB) (*Client, error) {
+	c := &Client{
+		conn:             conn,
+		transport:        NewTransport(ocb, false),
+		ocb:              ocb,
+		outputC:          make(chan struct{}, 1),
+		done:             make(chan struct{}),
+		sentActionCounts: make(map[uint64]int),
+	}
+
+	c.wg.Add(1)
+	go c.recvLoop()
+
+	// Send initial keepalive to associate with the server.
+	c.transport.ForceNextSend()
+	c.tick()
+
+	return c, nil
+}
+
+// DialConnRaw creates a mosh client with no internal goroutines.
+// The caller must call Tick() for sending AND RecvRaw() for receiving.
+// Use this when the caller needs raw transport diffs (e.g., for
+// framebuffer state tracking in the WASM client).
+func DialConnRaw(conn Conn, ocb *OCB) (*Client, error) {
+	c := &Client{
+		conn:             conn,
+		transport:        NewTransport(ocb, false),
+		ocb:              ocb,
+		outputC:          make(chan struct{}, 1),
+		done:             make(chan struct{}),
+		sentActionCounts: make(map[uint64]int),
+	}
+
+	c.transport.ForceNextSend()
+	c.tick()
+
+	return c, nil
+}
+
+// RecvRaw reads one datagram from the connection and processes it
+// through the transport. Returns the raw diff payload, or nil if
+// no complete message was received (timeout, fragment, replay).
+// The caller can use Transport().LastRecvOldNum()/LastRecvNewNum()
+// to get the state numbers for this diff.
+func (c *Client) RecvRaw(timeout time.Duration) []byte {
+	buf := make([]byte, maxPayload+64)
+
+	c.conn.SetReadDeadline(time.Now().Add(timeout))
+	n, err := c.conn.Read(buf)
+	if err != nil {
+		return nil
+	}
+	if n < minDatagram {
+		return nil
+	}
+
+	data := make([]byte, n)
+	copy(data, buf[:n])
+	return c.transport.Recv(data)
+}
+
 // Send sends keystrokes to the server.
 func (c *Client) Send(keys []byte) {
-	diff := MarshalUserMessage([]UserInstruction{{Keys: keys}})
-	c.transport.SetPending(diff)
-	c.tick()
+	c.actionsMu.Lock()
+	c.actions = append(c.actions, UserInstruction{Keys: append([]byte{}, keys...)})
+	c.dirty = true
+	c.actionsMu.Unlock()
 }
 
 // Resize sends a resize to the server.
 func (c *Client) Resize(cols, rows uint16) {
-	diff := MarshalUserMessage([]UserInstruction{{
-		Width:  int32(cols),
-		Height: int32(rows),
-	}})
-	c.transport.SetPending(diff)
-	c.tick()
+	c.actionsMu.Lock()
+	c.actions = append(c.actions, UserInstruction{Width: int32(cols), Height: int32(rows)})
+	c.dirty = true
+	c.actionsMu.Unlock()
 }
 
+
 // Recv reads accumulated terminal output, blocking until output is available
 // or the timeout expires. Returns nil on timeout.
 func (c *Client) Recv(timeout time.Duration) []byte {
@@ -131,16 +212,61 @@ func (c *Client) Transport() *Transport {
 	return c.transport
 }
 
+// Tick flushes pending actions as mosh datagrams. Call this periodically
+// from an external timer (e.g., JS setInterval in WASM) when the internal
+// sendLoop is not used.
+func (c *Client) Tick() {
+	c.tick()
+}
+
 func (c *Client) tick() {
+	// Only create a new state when the server has acked the previous one.
+	// This matches the Dart mosh client's behavior: at most one unacked
+	// state in flight. Keystrokes accumulate until the ack arrives, then
+	// the next tick sends them all as a single new state.
+	// Without this check, SetPending resets diffSent, causing Tick() to
+	// increment sentNum for the same cumulative payload — the server
+	// applies duplicate keystrokes.
+	c.actionsMu.Lock()
+	if c.dirty && c.transport.AckedByRemote() >= c.transport.SentNum() {
+		c.dirty = false
+		c.processAcksLocked()
+		newActions := c.actions[c.ackedActionCount:]
+		c.transport.SetPending(MarshalUserMessage(newActions))
+		c.sentActionCounts[c.transport.SentNum()+1] = len(c.actions)
+	}
+	c.actionsMu.Unlock()
+
 	for _, dg := range c.transport.Tick() {
 		c.conn.Write(dg)
 	}
 }
 
+func (c *Client) processAcksLocked() {
+	acked := c.transport.AckedByRemote()
+	if acked > c.lastAcked {
+		c.lastAcked = acked
+		// Only advance base when server caught up (no states in flight).
+		if acked >= c.transport.SentNum() {
+			if count, ok := c.sentActionCounts[acked]; ok && count > c.ackedActionCount {
+				c.ackedActionCount = count
+			}
+			c.sentActionCounts = make(map[uint64]int)
+		}
+	}
+}
+
 func (c *Client) recvLoop() {
 	defer c.wg.Done()
 	buf := make([]byte, maxPayload+64)
 
+	// Track the base state (oldNum) of the most recently buffered output.
+	// Mosh diffs are cumulative snapshots from a given base. When a new diff
+	// has the same oldNum as the previous one, it supersedes the older diff
+	// and we replace the output buffer instead of appending. This prevents
+	// character ghosting caused by concatenating overlapping diffs.
+	var lastOutputOldNum uint64
+
 	for {
 		select {
 		case <-c.done:
@@ -168,6 +294,9 @@ func (c *Client) recvLoop() {
 			continue
 		}
 
+		// Read oldNum AFTER transport.Recv() — it's set by the Recv call above.
+		currentOldNum := c.transport.LastRecvOldNum()
+
 		instrs, err := UnmarshalHostMessage(diff)
 		if err != nil || len(instrs) == 0 {
 			continue
@@ -182,7 +311,14 @@ func (c *Client) recvLoop() {
 		}
 
 		c.mu.Lock()
-		c.output = append(c.output, output...)
+		// Same base → newer diff supersedes older one (replace).
+		// Different base → new incremental content (append).
+		if currentOldNum == lastOutputOldNum {
+			c.output = output
+		} else {
+			c.output = append(c.output, output...)
+		}
+		lastOutputOldNum = currentOldNum
 		c.mu.Unlock()
 
 		select {

client_test.go

@@ -3,6 +3,7 @@ package mosh
 import (
 	"io"
 	"os/exec"
+	"regexp"
 	"strconv"
 	"strings"
 	"sync"
@@ -10,6 +11,12 @@ import (
 	"time"
 )
 
+var clientAnsiRe = regexp.MustCompile(`\x1b\[[0-9;?]*[A-Za-z]`)
+
+func clientStripANSI(s string) string {
+	return clientAnsiRe.ReplaceAllString(s, "")
+}
+
 // TestClientServerE2E tests the Go client against the Go server.
 func TestClientServerE2E(t *testing.T) {
 	srv, err := NewServer("/bin/sh", 0, 0)
@@ -48,13 +55,13 @@ func TestClientServerE2E(t *testing.T) {
 	for {
 		select {
 		case <-deadline:
-			t.Fatalf("marker not echoed. Got: %q", allOutput)
+			t.Fatalf("marker not echoed. Got: %q", clientStripANSI(allOutput))
 		default:
 		}
 		out := client.Recv(500 * time.Millisecond)
 		if out != nil {
 			allOutput += string(out)
-			if strings.Contains(allOutput, marker) {
+			if strings.Contains(clientStripANSI(allOutput), marker) {
 				t.Log("Go client → Go server E2E passed")
 				return
 			}
@@ -103,7 +110,8 @@ func TestClientResize(t *testing.T) {
 		out := client.Recv(500 * time.Millisecond)
 		if out != nil {
 			allOutput += string(out)
-			if strings.Contains(allOutput, "132") && strings.Contains(allOutput, "43") {
+			stripped := clientStripANSI(allOutput)
+			if strings.Contains(stripped, "132") && strings.Contains(stripped, "43") {
 				t.Log("Go client resize verified: 132x43")
 				return
 			}
@@ -165,7 +173,7 @@ func TestClientServeRW(t *testing.T) {
 		out := client.Recv(500 * time.Millisecond)
 		if out != nil {
 			allOutput += string(out)
-			if strings.Contains(allOutput, "hello from latch") {
+			if strings.Contains(clientStripANSI(allOutput), "hello from latch") {
 				break
 			}
 		}
@@ -194,6 +202,14 @@ func TestClientServeRW(t *testing.T) {
 		t.Fatal("timed out waiting for keystrokes on pipe")
 	}
 
+	// Drain any extra data from the pipe (cumulative diffs may resend keystrokes).
+	go func() {
+		buf := make([]byte, 4096)
+		for {
+			inR.Read(buf)
+		}
+	}()
+
 	// Resize.
 	client.Resize(200, 50)
 	time.Sleep(500 * time.Millisecond)
@@ -284,7 +300,7 @@ func TestClientAgainstRealMoshServer(t *testing.T) {
 		recv := client.Recv(500 * time.Millisecond)
 		if recv != nil {
 			allOutput += string(recv)
-			if strings.Contains(allOutput, marker) {
+			if strings.Contains(clientStripANSI(allOutput), marker) {
 				t.Log("Go client → real mosh-server E2E passed")
 				return
 			}
@@ -292,6 +308,61 @@ func TestClientAgainstRealMoshServer(t *testing.T) {
 	}
 }
 
+// TestFastTypingNoDuplication verifies that rapid keystrokes don't cause
+// character doubling due to overlapping diffs from the same base.
+func TestFastTypingNoDuplication(t *testing.T) {
+	srv, err := NewServer("/bin/sh", 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Serve()
+	defer srv.Close()
+	<-srv.started
+
+	client, err := Dial("127.0.0.1", srv.Port(), srv.KeyBase64())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	// Wait for shell prompt.
+	for i := 0; i < 40; i++ {
+		if out := client.Recv(500 * time.Millisecond); len(out) > 0 {
+			break
+		}
+	}
+
+	// Send "echo TESTMARKER\n" as rapid individual keystrokes with no delay.
+	cmd := "echo TESTMARKER\n"
+	for _, ch := range cmd {
+		client.Send([]byte{byte(ch)})
+	}
+
+	// Collect output until we see TESTMARKER in the echo output.
+	var allOutput string
+	deadline := time.After(10 * time.Second)
+	for {
+		select {
+		case <-deadline:
+			t.Fatalf("marker not echoed. Got: %q", clientStripANSI(allOutput))
+		default:
+		}
+		out := client.Recv(500 * time.Millisecond)
+		if out != nil {
+			allOutput += string(out)
+			stripped := clientStripANSI(allOutput)
+			if strings.Contains(stripped, "TESTMARKER") {
+				// Check that TESTMARKER appears but TTEESSTTMMAARRKKEERR does not.
+				if strings.Contains(stripped, "TTEESSTTMMAARRKKEERR") {
+					t.Fatalf("character doubling detected: %q", stripped)
+				}
+				t.Log("fast typing test passed: no character duplication")
+				return
+			}
+		}
+	}
+}
+
 // parseMoshConnect extracts port and key from mosh-server output.
 func parseMoshConnect(output string) (int, string) {
 	for _, line := range strings.Split(output, "\n") {

cmd/mosh-wasm/main.go

@@ -0,0 +1,191 @@
+//go:build js && wasm
+
+package main
+
+import (
+	"encoding/base64"
+	"fmt"
+	"sync"
+	"syscall/js"
+	"time"
+
+	mosh "github.com/unixshells/mosh-go"
+)
+
+func main() {
+	js.Global().Set("moshConnect", js.FuncOf(moshConnect))
+	select {} // keep alive
+}
+
+// moshConnect(url, key, cols, rows) → Promise<MoshSession>
+func moshConnect(this js.Value, args []js.Value) interface{} {
+	if len(args) < 4 {
+		return reject("moshConnect requires (url, key, cols, rows)")
+	}
+
+	url := args[0].String()
+	key := args[1].String()
+	cols := args[2].Int()
+	rows := args[3].Int()
+
+	handler := js.FuncOf(func(this js.Value, pargs []js.Value) interface{} {
+		resolve := pargs[0]
+		rejectFn := pargs[1]
+
+		go func() {
+			session, err := newSession(url, key, cols, rows)
+			if err != nil {
+				rejectFn.Invoke(err.Error())
+				return
+			}
+			resolve.Invoke(session.jsObject())
+		}()
+
+		return nil
+	})
+
+	return js.Global().Get("Promise").New(handler)
+}
+
+type session struct {
+	client  *mosh.Client
+	conn    *wtConn
+	tracker *stateTracker
+	mu      sync.Mutex
+	closed  bool
+}
+
+func newSession(url, key string, cols, rows int) (*session, error) {
+	for len(key)%4 != 0 {
+		key += "="
+	}
+	rawKey, err := base64.StdEncoding.DecodeString(key)
+	if err != nil {
+		return nil, fmt.Errorf("bad key: %w", err)
+	}
+	ocb, err := mosh.NewOCB(rawKey)
+	if err != nil {
+		return nil, err
+	}
+
+	conn, err := dialWebTransport(url)
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := mosh.DialConnRaw(conn, ocb)
+	if err != nil {
+		conn.Close()
+		return nil, err
+	}
+
+	s := &session{
+		client:  client,
+		conn:    conn,
+		tracker: newStateTracker(cols, rows),
+	}
+
+	// Background goroutine: read from mosh client, feed to state tracker.
+	go s.recvLoop()
+
+	return s, nil
+}
+
+func (s *session) jsObject() js.Value {
+	obj := js.Global().Get("Object").New()
+
+	obj.Set("send", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		if len(args) < 1 {
+			return nil
+		}
+		data := []byte(args[0].String())
+		s.client.Send(data)
+		s.client.Tick()
+
+		// Feed input to predictor — returns immediate overlay output.
+		if out := s.tracker.keystroke(data); len(out) > 0 {
+			return string(out)
+		}
+		return js.Null()
+	}))
+
+	obj.Set("resize", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		if len(args) < 2 {
+			return nil
+		}
+		cols := args[0].Int()
+		rows := args[1].Int()
+		s.client.Resize(uint16(cols), uint16(rows))
+		s.tracker.resize(cols, rows)
+		s.client.Tick()
+		return nil
+	}))
+
+	// tick() — flush pending keystrokes + keepalive/retransmit.
+	obj.Set("tick", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		s.client.Tick()
+		return nil
+	}))
+
+	// poll() — returns pre-diffed ANSI output from the state tracker.
+	obj.Set("poll", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		out := s.tracker.poll()
+		if out == nil {
+			return js.Null()
+		}
+		return string(out)
+	}))
+
+	// reconnect() — swap WebTransport connection using session token.
+	obj.Set("reconnect", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		handler := js.FuncOf(func(this js.Value, pargs []js.Value) interface{} {
+			resolve := pargs[0]
+			rejectFn := pargs[1]
+			go func() {
+				if err := s.conn.Reconnect(); err != nil {
+					rejectFn.Invoke(err.Error())
+					return
+				}
+				resolve.Invoke(true)
+			}()
+			return nil
+		})
+		return js.Global().Get("Promise").New(handler)
+	}))
+
+	obj.Set("close", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		s.mu.Lock()
+		s.closed = true
+		s.mu.Unlock()
+		s.client.Close()
+		return nil
+	}))
+
+	return obj
+}
+
+// recvLoop reads raw transport diffs and feeds them to the state
+// tracker for framebuffer-based processing.
+func (s *session) recvLoop() {
+	for {
+		diff := s.client.RecvRaw(1 * time.Second)
+		if diff == nil {
+			s.mu.Lock()
+			closed := s.closed
+			s.mu.Unlock()
+			if closed {
+				return
+			}
+			continue
+		}
+		t := s.client.Transport()
+		oldNum := t.LastRecvOldNum()
+		newNum := t.LastRecvNewNum()
+		throwawayNum := t.ThrowawayNum()
+		s.tracker.applyDiff(diff, oldNum, newNum, throwawayNum)
+	}
+}
+
+func reject(msg string) js.Value {
+	return js.Global().Get("Promise").Call("reject", msg)
+}

cmd/mosh-wasm/state.go

@@ -0,0 +1,225 @@
+//go:build js && wasm
+
+package main
+
+import (
+	"sync"
+	"time"
+
+	mosh "github.com/unixshells/mosh-go"
+	vt "github.com/unixshells/vt-go"
+)
+
+// stateTracker implements Dart-style received state management.
+// Each incoming diff is applied to a copy of the base state (looked up
+// by oldNum), producing a new state stored by newNum. The display
+// always shows the latest received state, diffed against what's
+// currently on screen. This eliminates garbled output from missed
+// diffs and overlapping state transitions.
+type stateTracker struct {
+	mu sync.Mutex
+
+	shadow      *vt.Emulator
+	shadowState uint64
+	cols, rows  int
+
+	// Received states: stateNum → framebuffer snapshot.
+	states       map[uint64]*mosh.Framebuffer
+	latestState  uint64
+	displayedFB  *mosh.Framebuffer
+
+	// Prediction engine for local echo.
+	predictor *mosh.Predictor
+
+	// Output buffer: pre-diffed ANSI ready for the terminal.
+	output []byte
+}
+
+func newStateTracker(cols, rows int) *stateTracker {
+	return &stateTracker{
+		shadow:    vt.NewEmulator(cols, rows),
+		cols:      cols,
+		rows:      rows,
+		states:    make(map[uint64]*mosh.Framebuffer),
+		predictor: mosh.NewPredictor(),
+	}
+}
+
+// applyDiff processes an incoming server diff using state tracking.
+// throwawayNum is the server's indication of which states it no longer references.
+func (st *stateTracker) applyDiff(diff []byte, oldNum, newNum, throwawayNum uint64) {
+	st.mu.Lock()
+	defer st.mu.Unlock()
+
+	// Already have this state — skip.
+	if _, ok := st.states[newNum]; ok {
+		return
+	}
+
+	// Snapshot current shadow as oldNum if we don't have it.
+	if oldNum == st.shadowState {
+		if _, ok := st.states[oldNum]; !ok {
+			st.states[oldNum] = mosh.SnapshotEmulator(st.shadow, true)
+		}
+	}
+
+	// Need the base state to apply the diff.
+	base, ok := st.states[oldNum]
+	if !ok {
+		// Unknown base state. Reset the shadow emulator and start fresh.
+		st.resetShadow()
+		st.shadowState = oldNum
+		base = mosh.SnapshotEmulator(st.shadow, true)
+		st.states[oldNum] = base
+	} else if st.shadowState != oldNum {
+		// Restore shadow to the base state.
+		st.resetShadow()
+		if base != nil {
+			st.shadow.Write(base.Diff(nil))
+		}
+	}
+
+	// Apply diff: feed hoststrings to shadow.
+	instrs, err := mosh.UnmarshalHostMessage(diff)
+	if err != nil {
+		return
+	}
+	for _, hi := range instrs {
+		if len(hi.Hoststring) > 0 {
+			st.shadow.Write(hi.Hoststring)
+		}
+	}
+	st.shadowState = newNum
+
+	// Snapshot result.
+	snap := mosh.SnapshotEmulator(st.shadow, true)
+	st.states[newNum] = snap
+	if newNum > st.latestState {
+		st.latestState = newNum
+	}
+
+	// Prune states the server no longer references.
+	if throwawayNum > 0 {
+		for n := range st.states {
+			if n < throwawayNum {
+				delete(st.states, n)
+			}
+		}
+	}
+
+	// Hard cap: keep at most 16 states to prevent unbounded memory growth.
+	if len(st.states) > 16 {
+		minKeep := st.latestState
+		if minKeep > 16 {
+			minKeep -= 16
+		} else {
+			minKeep = 0
+		}
+		for n := range st.states {
+			if n < minKeep {
+				delete(st.states, n)
+			}
+		}
+	}
+
+	// Confirm predictions against server state.
+	latest := st.states[st.latestState]
+	if latest != nil {
+		st.predictor.Confirm(latest)
+		st.predictor.ExpireStale(time.Now())
+
+		// Apply prediction overlay for display.
+		displayFB := latest
+		if st.predictor.Active() {
+			displayFB = st.cloneFB(latest)
+			st.predictor.Overlay(displayFB)
+		}
+
+		ansi := displayFB.Diff(st.displayedFB)
+		if len(ansi) > 0 {
+			st.output = append(st.output, ansi...)
+		}
+		st.displayedFB = displayFB
+	}
+}
+
+// poll returns accumulated ANSI output, or nil.
+func (st *stateTracker) poll() []byte {
+	st.mu.Lock()
+	defer st.mu.Unlock()
+	if len(st.output) == 0 {
+		return nil
+	}
+	out := st.output
+	st.output = nil
+	return out
+}
+
+// keystroke feeds user input to the predictor and returns any prediction
+// overlay output that should be displayed immediately.
+func (st *stateTracker) keystroke(data []byte) []byte {
+	st.mu.Lock()
+	defer st.mu.Unlock()
+
+	st.predictor.Keystroke(data)
+
+	if !st.predictor.Active() || st.displayedFB == nil {
+		return nil
+	}
+
+	// Overlay predictions onto the current display and diff.
+	overlaid := st.cloneFB(st.displayedFB)
+
+	// First undo previous prediction overlay by using latest server state.
+	var serverFB *mosh.Framebuffer
+	if st.latestState > 0 {
+		serverFB = st.states[st.latestState]
+	}
+	if serverFB == nil {
+		serverFB = st.displayedFB
+	}
+
+	displayFB := st.cloneFB(serverFB)
+	st.predictor.Overlay(displayFB)
+
+	ansi := displayFB.Diff(overlaid)
+	if len(ansi) > 0 {
+		st.displayedFB = displayFB
+	}
+	return ansi
+}
+
+// cloneFB creates a deep copy of a framebuffer.
+func (st *stateTracker) cloneFB(fb *mosh.Framebuffer) *mosh.Framebuffer {
+	clone := &mosh.Framebuffer{
+		W:      fb.W,
+		H:      fb.H,
+		CurX:   fb.CurX,
+		CurY:   fb.CurY,
+		CurVis: fb.CurVis,
+		Cells:  make([]mosh.Cell, len(fb.Cells)),
+	}
+	copy(clone.Cells, fb.Cells)
+	return clone
+}
+
+// resize resets the state tracker for new dimensions.
+func (st *stateTracker) resize(cols, rows int) {
+	st.mu.Lock()
+	defer st.mu.Unlock()
+	st.cols = cols
+	st.rows = rows
+	// Must create a new emulator for new dimensions.
+	st.shadow = vt.NewEmulator(cols, rows)
+	st.shadowState = 0
+	st.states = make(map[uint64]*mosh.Framebuffer)
+	st.displayedFB = nil
+	st.predictor.Reset()
+}
+
+// resetShadow clears the shadow emulator without reallocating it.
+func (st *stateTracker) resetShadow() {
+	st.shadow.Resize(st.cols, st.rows)
+	st.shadow.Write([]byte("\033[H\033[2J\033[m"))
+}
+

cmd/mosh-wasm/webtransport.go

@@ -0,0 +1,273 @@
+//go:build js && wasm
+
+package main
+
+import (
+	"errors"
+	"strings"
+	"sync"
+	"syscall/js"
+	"time"
+)
+
+// wtConn wraps a browser WebTransport connection as a mosh.Conn.
+// Supports transparent reconnection: when the underlying WebTransport
+// dies, reads return timeouts while the worker re-dials. Once a new
+// connection is swapped in via Reconnect(), reads resume.
+type wtConn struct {
+	mu        sync.Mutex
+	transport js.Value
+	writer    js.Value // persistent writable stream writer
+	incoming  chan []byte
+	done      chan struct{}
+	once      sync.Once
+	deadline  time.Time
+
+	url   string
+	token string // session token from relay, used for reconnect
+}
+
+func dialWebTransport(url string) (*wtConn, error) {
+	wt := js.Global().Get("WebTransport").New(url)
+
+	readyCh := make(chan error, 1)
+	onReady := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		readyCh <- nil
+		return nil
+	})
+	onFail := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		msg := "WebTransport connection failed"
+		if len(args) > 0 && !args[0].IsUndefined() && !args[0].IsNull() {
+			if m := args[0].Get("message"); !m.IsUndefined() {
+				msg = m.String()
+			}
+		}
+		readyCh <- errors.New(msg)
+		return nil
+	})
+	wt.Get("ready").Call("then", onReady, onFail)
+
+	select {
+	case err := <-readyCh:
+		if err != nil {
+			onReady.Release()
+			onFail.Release()
+			return nil, err
+		}
+	case <-time.After(10 * time.Second):
+		onReady.Release()
+		onFail.Release()
+		wt.Call("close")
+		return nil, errors.New("WebTransport connect timeout")
+	}
+	onReady.Release()
+	onFail.Release()
+
+	writer := wt.Get("datagrams").Get("writable").Call("getWriter")
+
+	c := &wtConn{
+		transport: wt,
+		writer:    writer,
+		incoming:  make(chan []byte, 256),
+		done:      make(chan struct{}),
+		url:       url,
+	}
+
+	// Start reading datagrams via a persistent JS callback.
+	c.startReader()
+
+	return c, nil
+}
+
+// Reconnect dials a new WebTransport connection using the session token
+// and swaps it into this conn. The mosh client keeps its SSP state and
+// resumes transparently.
+func (c *wtConn) Reconnect() error {
+	c.mu.Lock()
+	token := c.token
+	url := c.url
+	c.mu.Unlock()
+
+	if token == "" {
+		return errors.New("no session token")
+	}
+
+	// Add token to URL for reconnect.
+	sep := "&"
+	if !strings.Contains(url, "?") {
+		sep = "?"
+	}
+	reconnURL := url + sep + "token=" + token
+
+	wt := js.Global().Get("WebTransport").New(reconnURL)
+
+	readyCh := make(chan error, 1)
+	onReady := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		readyCh <- nil
+		return nil
+	})
+	onFail := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		msg := "reconnect failed"
+		if len(args) > 0 && !args[0].IsUndefined() && !args[0].IsNull() {
+			if m := args[0].Get("message"); !m.IsUndefined() {
+				msg = m.String()
+			}
+		}
+		readyCh <- errors.New(msg)
+		return nil
+	})
+	wt.Get("ready").Call("then", onReady, onFail)
+
+	select {
+	case err := <-readyCh:
+		if err != nil {
+			onReady.Release()
+			onFail.Release()
+			return err
+		}
+	case <-time.After(10 * time.Second):
+		onReady.Release()
+		onFail.Release()
+		wt.Call("close")
+		return errors.New("reconnect timeout")
+	}
+	onReady.Release()
+	onFail.Release()
+
+	writer := wt.Get("datagrams").Get("writable").Call("getWriter")
+
+	// Swap in the new connection.
+	c.mu.Lock()
+	oldTransport := c.transport
+	c.transport = wt
+	c.writer = writer
+	c.mu.Unlock()
+
+	// Close old transport (best-effort).
+	if !oldTransport.IsUndefined() && !oldTransport.IsNull() {
+		defer func() { recover() }()
+		oldTransport.Call("close")
+	}
+
+	// Start reading from the new connection.
+	c.startReader()
+
+	return nil
+}
+
+// Token returns the session token received from the relay.
+func (c *wtConn) Token() string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.token
+}
+
+// startReader sets up a self-chaining read loop using persistent callbacks.
+func (c *wtConn) startReader() {
+	c.mu.Lock()
+	reader := c.transport.Get("datagrams").Get("readable").Call("getReader")
+	c.mu.Unlock()
+
+	var onData, onErr js.Func
+	var readNext js.Func
+
+	readNext = js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		reader.Call("read").Call("then", onData, onErr)
+		return nil
+	})
+
+	onData = js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		result := args[0]
+		if result.Get("done").Bool() {
+			return nil
+		}
+		value := result.Get("value")
+		buf := make([]byte, value.Get("byteLength").Int())
+		js.CopyBytesToGo(buf, js.Global().Get("Uint8Array").New(value))
+
+		// Check for TOKEN: prefix (session token from relay).
+		if len(buf) > 6 && string(buf[:6]) == "TOKEN:" {
+			c.mu.Lock()
+			c.token = string(buf[6:])
+			c.mu.Unlock()
+		} else {
+			select {
+			case c.incoming <- buf:
+			default:
+			}
+		}
+
+		// Defer next read to next event loop tick so Go goroutines can process.
+		js.Global().Call("setTimeout", readNext, 0)
+		return nil
+	})
+
+	onErr = js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		// Stream error or closed — stop reading.
+		// The reconnect logic in the worker will handle this.
+		return nil
+	})
+
+	// Start first read.
+	readNext.Invoke()
+}
+
+func (c *wtConn) Read(b []byte) (int, error) {
+	c.mu.Lock()
+	deadline := c.deadline
+	c.mu.Unlock()
+
+	var timer <-chan time.Time
+	if !deadline.IsZero() {
+		d := time.Until(deadline)
+		if d <= 0 {
+			return 0, errors.New("i/o timeout")
+		}
+		timer = time.After(d)
+	}
+
+	select {
+	case <-c.done:
+		return 0, errors.New("connection closed")
+	case data := <-c.incoming:
+		n := copy(b, data)
+		return n, nil
+	case <-timer:
+		return 0, errors.New("i/o timeout")
+	}
+}
+
+func (c *wtConn) Write(b []byte) (int, error) {
+	c.mu.Lock()
+	writer := c.writer
+	c.mu.Unlock()
+
+	if writer.IsUndefined() || writer.IsNull() {
+		return 0, errors.New("not connected")
+	}
+
+	arr := js.Global().Get("Uint8Array").New(len(b))
+	js.CopyBytesToJS(arr, b)
+	writer.Call("write", arr)
+	return len(b), nil
+}
+
+func (c *wtConn) SetReadDeadline(t time.Time) error {
+	c.mu.Lock()
+	c.deadline = t
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *wtConn) Close() error {
+	c.once.Do(func() {
+		close(c.done)
+		c.mu.Lock()
+		t := c.transport
+		c.mu.Unlock()
+		if !t.IsUndefined() && !t.IsNull() {
+			t.Call("close")
+		}
+	})
+	return nil
+}

cmd/remote-test/main.go

@@ -0,0 +1,133 @@
+package main
+
+import (
+	"encoding/base64"
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+	"time"
+	"unicode/utf8"
+
+	mosh "github.com/unixshells/mosh-go"
+)
+
+func main() {
+	if len(os.Args) != 4 {
+		fmt.Fprintf(os.Stderr, "Usage: %s <host> <port> <base64key>\n", os.Args[0])
+		os.Exit(1)
+	}
+
+	host := os.Args[1]
+	port, _ := strconv.Atoi(os.Args[2])
+	key, err := base64.StdEncoding.DecodeString(os.Args[3])
+	if err != nil {
+		// Try raw base64 without padding.
+		key, err = base64.RawStdEncoding.DecodeString(os.Args[2+1])
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "bad key: %v\n", err)
+			os.Exit(1)
+		}
+	}
+
+	ocb, err := mosh.NewOCB(key)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "ocb: %v\n", err)
+		os.Exit(1)
+	}
+
+	transport := mosh.NewTransport(ocb, false) // client mode
+
+	// Resolve host.
+	addrs, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", host, port))
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "resolve: %v\n", err)
+		os.Exit(1)
+	}
+
+	conn, err := net.DialUDP("udp", nil, addrs)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "dial: %v\n", err)
+		os.Exit(1)
+	}
+	defer conn.Close()
+
+	flush := func() {
+		datagrams := transport.Tick()
+		for _, dg := range datagrams {
+			conn.Write(dg)
+		}
+	}
+
+	// Read incoming in background.
+	go func() {
+		buf := make([]byte, 65536)
+		for {
+			n, err := conn.Read(buf)
+			if err != nil {
+				return
+			}
+			diff := transport.Recv(buf[:n])
+			if diff == nil || len(diff) == 0 {
+				continue
+			}
+
+			instrs, err := mosh.UnmarshalHostMessage(diff)
+			if err != nil {
+				fmt.Printf("unmarshal error: %v\n", err)
+				continue
+			}
+			for _, hi := range instrs {
+				if len(hi.Hoststring) > 0 {
+					text := string(hi.Hoststring)
+					if !utf8.ValidString(text) {
+						text = fmt.Sprintf("(%d raw bytes)", len(hi.Hoststring))
+					}
+					fmt.Printf("=== hoststring (%d bytes) ===\n%s\n=== end ===\n", len(hi.Hoststring), text)
+				}
+				if hi.Width > 0 || hi.Height > 0 {
+					fmt.Printf("=== resize: %dx%d ===\n", hi.Width, hi.Height)
+				}
+			}
+		}
+	}()
+
+	// Tick timer.
+	ticker := time.NewTicker(20 * time.Millisecond)
+	go func() {
+		for range ticker.C {
+			flush()
+		}
+	}()
+
+	// Initial keepalive.
+	fmt.Println("Sending initial keepalive...")
+	transport.ForceNextSend()
+	flush()
+
+	// Resize.
+	fmt.Println("Sending resize 80x24...")
+	resize := mosh.MarshalUserMessage([]mosh.UserInstruction{
+		{Width: 80, Height: 24},
+	})
+	transport.SetPending(resize)
+	flush()
+
+	// Wait for initial output.
+	fmt.Println("Waiting 5 seconds for initial output...")
+	time.Sleep(5 * time.Second)
+
+	// Send ls + Enter.
+	fmt.Println("\nSending ls + Enter...")
+	keystroke := mosh.MarshalUserMessage([]mosh.UserInstruction{
+		{Keys: []byte("ls\r")},
+	})
+	transport.SetPending(keystroke)
+	flush()
+
+	// Wait for response.
+	time.Sleep(3 * time.Second)
+
+	fmt.Println("\nDone.")
+	ticker.Stop()
+}

conn.go

@@ -0,0 +1,36 @@
+package mosh
+
+import "time"
+
+// Conn is a datagram-oriented connection used by the mosh transport.
+// UDP, WebTransport, and other datagram transports implement this interface.
+type Conn interface {
+	// Read reads a datagram. Returns the number of bytes read.
+	// Must respect deadlines set by SetReadDeadline.
+	Read(b []byte) (int, error)
+
+	// Write writes a datagram.
+	Write(b []byte) (int, error)
+
+	// SetReadDeadline sets the deadline for Read calls.
+	SetReadDeadline(t time.Time) error
+
+	// Close closes the connection.
+	Close() error
+}
+
+const (
+	// Direction bits in the 64-bit nonce header.
+	dirToServer = uint64(0)          // client → server
+	dirToClient = uint64(1) << 63    // server → client
+	seqMask     = ^(uint64(1) << 63) // lower 63 bits
+
+	// Minimum wire datagram: 8 (nonce) + 16 (tag) = 24 bytes.
+	minDatagram = 24
+
+	// Maximum payload per UDP datagram.
+	maxPayload = 16384
+
+	// Tick interval matches mosh's SEND_MINDELAY (8ms) for keystroke batching.
+	tickInterval = 8 * time.Millisecond
+)

framebuffer.go

@@ -1,7 +1,24 @@
 package mosh
 
 import (
+	"image/color"
 	"strconv"
+	"unicode/utf8"
+
+	"github.com/charmbracelet/x/ansi"
+	"github.com/unixshells/vt-go"
+)
+
+// Cell attribute bitmasks (matching charmbracelet/ultraviolet values).
+const (
+	attrBold          = 1 << 0
+	attrFaint         = 1 << 1
+	attrItalic        = 1 << 2
+	attrBlink         = 1 << 3
+	attrRapidBlink    = 1 << 4
+	attrReverse       = 1 << 5
+	attrConceal       = 1 << 6
+	attrStrikethrough = 1 << 7
 )
 
 // Framebuffer represents a terminal screen state for mosh SSP.
@@ -157,11 +174,11 @@ func (fb *Framebuffer) fullRedraw() []byte {
 			buf = append(buf, "\r\n"...)
 		}
 		rowOff := y * fb.W
-		// Track trailing spaces to avoid sending them.
+		// Track trailing default-attr spaces to avoid sending them.
 		lastNonSpace := -1
 		for x := fb.W - 1; x >= 0; x-- {
 			c := &fb.Cells[rowOff+x]
-			if c.Rune != ' ' || c.Rune == 0 || c.Attr != (Attr{}) {
+			if (c.Rune != ' ' && c.Rune != 0) || c.Attr != (Attr{}) {
 				lastNonSpace = x
 				break
 			}
@@ -361,3 +378,77 @@ func encodeRune(buf []byte, r rune) int {
 		return 4
 	}
 }
+
+// SnapshotEmulator captures the VT emulator state into a Framebuffer.
+func SnapshotEmulator(emu *vt.Emulator, cursorVisible bool) *Framebuffer {
+	w := emu.Width()
+	h := emu.Height()
+	fb := NewFramebuffer(w, h)
+
+	for y := 0; y < h; y++ {
+		for x := 0; x < w; x++ {
+			c := emu.CellAt(x, y)
+			if c == nil {
+				continue
+			}
+			cell := &fb.Cells[y*w+x]
+
+			// Content → first rune.
+			if c.Content == "" || c.Content == " " {
+				cell.Rune = ' '
+			} else {
+				r, _ := utf8.DecodeRuneInString(c.Content)
+				if r == utf8.RuneError {
+					cell.Rune = ' '
+				} else {
+					cell.Rune = r
+				}
+			}
+
+			cell.Width = c.Width
+			if cell.Width == 0 && cell.Rune == ' ' {
+				// Continuation cell.
+				cell.Width = 0
+			}
+
+			// Attributes.
+			cell.Attr.Bold = c.Style.Attrs&attrBold != 0
+			cell.Attr.Dim = c.Style.Attrs&attrFaint != 0
+			cell.Attr.Italic = c.Style.Attrs&attrItalic != 0
+			cell.Attr.Blink = c.Style.Attrs&attrBlink != 0
+			cell.Attr.Reverse = c.Style.Attrs&attrReverse != 0
+			cell.Attr.Strike = c.Style.Attrs&attrStrikethrough != 0
+			cell.Attr.Under = c.Style.Underline != ansi.UnderlineNone
+
+			cell.Attr.FG = convertColor(c.Style.Fg)
+			cell.Attr.BG = convertColor(c.Style.Bg)
+		}
+	}
+
+	pos := emu.CursorPosition()
+	fb.CurX = pos.X
+	fb.CurY = pos.Y
+	fb.CurVis = cursorVisible
+
+	return fb
+}
+
+// convertColor converts a color.Color from the VT emulator to our Color type.
+func convertColor(c color.Color) Color {
+	if c == nil {
+		return Color{Type: ColorDefault}
+	}
+	switch v := c.(type) {
+	case ansi.BasicColor:
+		return Color{Type: ColorIndex, Value: uint32(v)}
+	case ansi.IndexedColor:
+		return Color{Type: ColorIndex, Value: uint32(v)}
+	case ansi.TrueColor:
+		return Color{Type: ColorRGB, Value: uint32(v)}
+	case color.RGBA:
+		return Color{Type: ColorRGB, Value: uint32(v.R)<<16 | uint32(v.G)<<8 | uint32(v.B)}
+	default:
+		r, g, b, _ := c.RGBA()
+		return Color{Type: ColorRGB, Value: uint32(r>>8)<<16 | uint32(g>>8)<<8 | uint32(b>>8)}
+	}
+}

go.mod

@@ -1,11 +1,30 @@
-module github.com/unixshells/mosh-go
+module github.com/ys-ll/mosh-go
 
 go 1.26.0
 
 require (
+	github.com/charmbracelet/x/ansi v0.11.6
 	github.com/creack/pty v1.1.24
+	github.com/unixshells/vt-go v0.1.0
 	golang.org/x/crypto v0.49.0
 	golang.org/x/term v0.41.0
 )
 
-require golang.org/x/sys v0.42.0 // indirect
+require (
+	github.com/charmbracelet/colorprofile v0.4.2 // indirect
+	github.com/charmbracelet/ultraviolet v0.0.0-20260303162955-0b88c25f3fff // indirect
+	github.com/charmbracelet/x/exp/ordered v0.1.0 // indirect
+	github.com/charmbracelet/x/term v0.2.2 // indirect
+	github.com/charmbracelet/x/termios v0.1.1 // indirect
+	github.com/charmbracelet/x/windows v0.2.2 // indirect
+	github.com/clipperhouse/displaywidth v0.9.0 // indirect
+	github.com/clipperhouse/stringish v0.1.1 // indirect
+	github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
+	github.com/lucasb-eyer/go-colorful v1.3.0 // indirect
+	github.com/mattn/go-runewidth v0.0.19 // indirect
+	github.com/muesli/cancelreader v0.2.2 // indirect
+	github.com/rivo/uniseg v0.4.7 // indirect
+	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
+	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sys v0.42.0 // indirect
+)

go.sum

@@ -1,7 +1,43 @@
+github.com/charmbracelet/colorprofile v0.4.2 h1:BdSNuMjRbotnxHSfxy+PCSa4xAmz7szw70ktAtWRYrY=
+github.com/charmbracelet/colorprofile v0.4.2/go.mod h1:0rTi81QpwDElInthtrQ6Ni7cG0sDtwAd4C4le060fT8=
+github.com/charmbracelet/ultraviolet v0.0.0-20260303162955-0b88c25f3fff h1:uY7A6hTokHPJBHfq7rj9Y/wm+IAjOghZTxKfVW6QLvw=
+github.com/charmbracelet/ultraviolet v0.0.0-20260303162955-0b88c25f3fff/go.mod h1:E6/0abq9uG2SnM8IbLB9Y5SW09uIgfaFETk8aRzgXUQ=
+github.com/charmbracelet/x/ansi v0.11.6 h1:GhV21SiDz/45W9AnV2R61xZMRri5NlLnl6CVF7ihZW8=
+github.com/charmbracelet/x/ansi v0.11.6/go.mod h1:2JNYLgQUsyqaiLovhU2Rv/pb8r6ydXKS3NIttu3VGZQ=
+github.com/charmbracelet/x/exp/ordered v0.1.0 h1:55/qLwjIh0gL0Vni+QAWk7T/qRVP6sBf+2agPBgnOFE=
+github.com/charmbracelet/x/exp/ordered v0.1.0/go.mod h1:5UHwmG+is5THxMyCJHNPCn2/ecI07aKNrW+LcResjJ8=
+github.com/charmbracelet/x/term v0.2.2 h1:xVRT/S2ZcKdhhOuSP4t5cLi5o+JxklsoEObBSgfgZRk=
+github.com/charmbracelet/x/term v0.2.2/go.mod h1:kF8CY5RddLWrsgVwpw4kAa6TESp6EB5y3uxGLeCqzAI=
+github.com/charmbracelet/x/termios v0.1.1 h1:o3Q2bT8eqzGnGPOYheoYS8eEleT5ZVNYNy8JawjaNZY=
+github.com/charmbracelet/x/termios v0.1.1/go.mod h1:rB7fnv1TgOPOyyKRJ9o+AsTU/vK5WHJ2ivHeut/Pcwo=
+github.com/charmbracelet/x/windows v0.2.2 h1:IofanmuvaxnKHuV04sC0eBy/smG6kIKrWG2/jYn2GuM=
+github.com/charmbracelet/x/windows v0.2.2/go.mod h1:/8XtdKZzedat74NQFn0NGlGL4soHB0YQZrETF96h75k=
+github.com/clipperhouse/displaywidth v0.9.0 h1:Qb4KOhYwRiN3viMv1v/3cTBlz3AcAZX3+y9OLhMtAtA=
+github.com/clipperhouse/displaywidth v0.9.0/go.mod h1:aCAAqTlh4GIVkhQnJpbL0T/WfcrJXHcj8C0yjYcjOZA=
+github.com/clipperhouse/stringish v0.1.1 h1:+NSqMOr3GR6k1FdRhhnXrLfztGzuG+VuFDfatpWHKCs=
+github.com/clipperhouse/stringish v0.1.1/go.mod h1:v/WhFtE1q0ovMta2+m+UbpZ+2/HEXNWYXQgCt4hdOzA=
+github.com/clipperhouse/uax29/v2 v2.7.0 h1:+gs4oBZ2gPfVrKPthwbMzWZDaAFPGYK72F0NJv2v7Vk=
+github.com/clipperhouse/uax29/v2 v2.7.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM=
 github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
 github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE=
+github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag=
+github.com/lucasb-eyer/go-colorful v1.3.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
+github.com/mattn/go-runewidth v0.0.19 h1:v++JhqYnZuu5jSKrk9RbgF5v4CGUjqRfBm05byFGLdw=
+github.com/mattn/go-runewidth v0.0.19/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs=
+github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA=
+github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
+github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
+github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+github.com/unixshells/vt-go v0.1.0 h1:HWILcExV8MHc6o2YqIRMIM/KrjH31kPzZrlvLS9BG6Y=
+github.com/unixshells/vt-go v0.1.0/go.mod h1:dZpOXxVyO7LG1uxkVMixX7HwUf2ZbUkEKhtHHJTTIBI=
+github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no=
+github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
 golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
 golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
+golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
+golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
+golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
+golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU=

predict.go

@@ -0,0 +1,207 @@
+package mosh
+
+import (
+	"time"
+	"unicode"
+	"unicode/utf8"
+)
+
+// Predictor implements mosh-style speculative local echo.
+//
+// When the user types a printable character, it is predicted to appear
+// at the current cursor position and the cursor advances. When the
+// server confirms the character (it appears in the server's framebuffer
+// at the expected position), the prediction is retired. If the server
+// diverges, all predictions are cleared.
+//
+// This is the core of what makes mosh feel local: keystrokes appear
+// instantly even over high-latency connections.
+type Predictor struct {
+	pending []prediction // unconfirmed predictions, in order typed
+	curX    int          // predicted cursor X (columns)
+	curY    int          // predicted cursor Y (rows)
+	epoch   int          // incremented on reset; stale predictions are ignored
+	active  bool         // false until first prediction; cleared on divergence
+
+	// Glitch: when the server confirms a prediction, we note it so the
+	// overlay can stop rendering that cell as "speculative".
+	confirmed int // count of predictions confirmed so far this epoch
+}
+
+type prediction struct {
+	r     rune
+	x, y  int // where we predicted it would land
+	epoch int
+	at    time.Time
+}
+
+const (
+	predictionTimeout = 500 * time.Millisecond
+)
+
+// NewPredictor creates a new predictor.
+func NewPredictor() *Predictor {
+	return &Predictor{}
+}
+
+// KeystrokeFrom processes user input and returns any printable runes
+// that should be predicted. Non-printable input (control chars, escape
+// sequences) clears predictions since we can't know what the server
+// will do.
+func (p *Predictor) Keystroke(input []byte) {
+	i := 0
+	for i < len(input) {
+		r, size := utf8.DecodeRune(input[i:])
+		i += size
+		if r == utf8.RuneError {
+			p.Reset()
+			return
+		}
+
+		if r < 0x20 || r == 0x7f {
+			// Control character — we can't predict what happens.
+			// Backspace, enter, tab, escape sequences, etc.
+			p.Reset()
+			return
+		}
+
+		if unicode.IsPrint(r) {
+			p.pending = append(p.pending, prediction{
+				r:     r,
+				x:     p.curX,
+				y:     p.curY,
+				epoch: p.epoch,
+				at:    time.Now(),
+			})
+			p.curX++
+			p.active = true
+		}
+	}
+}
+
+// Reset clears all predictions (e.g., on control characters or server divergence).
+func (p *Predictor) Reset() {
+	p.pending = p.pending[:0]
+	p.epoch++
+	p.active = false
+	p.confirmed = 0
+}
+
+// SetCursor updates the predicted cursor position from the server state.
+// Call this when processing server output so predictions know where
+// the cursor is.
+func (p *Predictor) SetCursor(x, y int) {
+	if !p.active {
+		p.curX = x
+		p.curY = y
+	}
+}
+
+// Active returns true if there are pending predictions.
+func (p *Predictor) Active() bool {
+	return p.active && len(p.pending) > 0
+}
+
+// ExpireStale removes predictions older than the timeout.
+func (p *Predictor) ExpireStale(now time.Time) {
+	cutoff := now.Add(-predictionTimeout)
+	changed := false
+	for len(p.pending) > 0 && p.pending[0].at.Before(cutoff) {
+		p.pending = p.pending[1:]
+		changed = true
+	}
+	if changed && len(p.pending) == 0 {
+		p.active = false
+	}
+}
+
+// Confirm checks the server framebuffer against pending predictions.
+// Confirmed predictions are removed. If the server shows something
+// different where we predicted, all predictions are cleared.
+func (p *Predictor) Confirm(fb *Framebuffer) {
+	if !p.active || len(p.pending) == 0 {
+		// No predictions, just track cursor.
+		p.curX = fb.CurX
+		p.curY = fb.CurY
+		return
+	}
+
+	// Walk through pending predictions and see if the server matches.
+	confirmed := 0
+	for confirmed < len(p.pending) {
+		pred := &p.pending[confirmed]
+
+		if pred.epoch != p.epoch {
+			// Stale prediction from before a reset.
+			confirmed++
+			continue
+		}
+
+		cell := fb.CellAt(pred.x, pred.y)
+		if cell == nil {
+			// Out of bounds — clear everything.
+			p.Reset()
+			p.curX = fb.CurX
+			p.curY = fb.CurY
+			return
+		}
+
+		if cell.Rune == pred.r {
+			// Server confirmed this prediction.
+			confirmed++
+		} else if (cell.Rune == ' ' || cell.Rune == 0) && pred.r != ' ' {
+			// Server hasn't caught up yet — stop checking.
+			// (But if we predicted a space, a space cell is a match, handled above.)
+			break
+		} else {
+			// Server diverged — something unexpected happened
+			// (tab completion, shell editing, etc.)
+			p.Reset()
+			p.curX = fb.CurX
+			p.curY = fb.CurY
+			return
+		}
+	}
+
+	if confirmed > 0 {
+		p.pending = p.pending[confirmed:]
+		p.confirmed += confirmed
+	}
+
+	if len(p.pending) == 0 {
+		// All predictions confirmed. Sync cursor to server.
+		p.active = false
+		p.curX = fb.CurX
+		p.curY = fb.CurY
+	}
+}
+
+// Overlay applies pending predictions onto a framebuffer for display.
+// The framebuffer is modified in-place — call this on a copy if needed.
+// Predicted characters get an underline attribute to indicate they're
+// speculative (matching C++ mosh behavior).
+func (p *Predictor) Overlay(fb *Framebuffer) {
+	if !p.active {
+		return
+	}
+
+	for i := range p.pending {
+		pred := &p.pending[i]
+		if pred.epoch != p.epoch {
+			continue
+		}
+		cell := fb.CellAt(pred.x, pred.y)
+		if cell == nil {
+			continue
+		}
+		cell.Rune = pred.r
+		cell.Width = 1
+		cell.Attr.Under = true // underline marks speculative characters
+	}
+
+	// Move cursor to predicted position.
+	if len(p.pending) > 0 {
+		fb.CurX = p.curX
+		fb.CurY = p.curY
+	}
+}

predict_test.go

@@ -0,0 +1,357 @@
+package mosh
+
+import (
+	"testing"
+	"time"
+	"unicode/utf8"
+)
+
+func TestPredictorBasicEcho(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+
+	// Type "abc"
+	p.Keystroke([]byte("a"))
+	p.Keystroke([]byte("b"))
+	p.Keystroke([]byte("c"))
+
+	if !p.Active() {
+		t.Fatal("predictor should be active after typing")
+	}
+	if len(p.pending) != 3 {
+		t.Fatalf("expected 3 pending, got %d", len(p.pending))
+	}
+	if p.pending[0].r != 'a' || p.pending[0].x != 0 || p.pending[0].y != 0 {
+		t.Fatalf("prediction 0: got %+v", p.pending[0])
+	}
+	if p.pending[1].r != 'b' || p.pending[1].x != 1 {
+		t.Fatalf("prediction 1: got %+v", p.pending[1])
+	}
+	if p.pending[2].r != 'c' || p.pending[2].x != 2 {
+		t.Fatalf("prediction 2: got %+v", p.pending[2])
+	}
+}
+
+func TestPredictorOverlay(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("hi"))
+
+	fb := NewFramebuffer(80, 24)
+	p.Overlay(fb)
+
+	c0 := fb.CellAt(0, 0)
+	c1 := fb.CellAt(1, 0)
+	if c0.Rune != 'h' || !c0.Attr.Under {
+		t.Fatalf("cell 0: got %+v, want 'h' underlined", c0)
+	}
+	if c1.Rune != 'i' || !c1.Attr.Under {
+		t.Fatalf("cell 1: got %+v, want 'i' underlined", c1)
+	}
+	if fb.CurX != 2 || fb.CurY != 0 {
+		t.Fatalf("cursor: got (%d,%d), want (2,0)", fb.CurX, fb.CurY)
+	}
+}
+
+func TestPredictorConfirmAll(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("ab"))
+
+	// Simulate server confirming "ab" at positions 0,1
+	fb := NewFramebuffer(80, 24)
+	fb.CellAt(0, 0).Rune = 'a'
+	fb.CellAt(1, 0).Rune = 'b'
+	fb.CurX = 2
+	fb.CurY = 0
+
+	p.Confirm(fb)
+	if p.Active() {
+		t.Fatal("predictor should be inactive after full confirmation")
+	}
+	if len(p.pending) != 0 {
+		t.Fatalf("expected 0 pending, got %d", len(p.pending))
+	}
+}
+
+func TestPredictorPartialConfirm(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("abc"))
+
+	// Server has confirmed "a" but not "b" or "c" yet
+	fb := NewFramebuffer(80, 24)
+	fb.CellAt(0, 0).Rune = 'a'
+	fb.CurX = 1
+	fb.CurY = 0
+
+	p.Confirm(fb)
+	if !p.Active() {
+		t.Fatal("predictor should still be active with pending predictions")
+	}
+	if len(p.pending) != 2 {
+		t.Fatalf("expected 2 pending, got %d", len(p.pending))
+	}
+	if p.pending[0].r != 'b' {
+		t.Fatalf("first remaining prediction: got %c, want b", p.pending[0].r)
+	}
+}
+
+func TestPredictorDivergence(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("abc"))
+
+	// Server shows 'x' where we predicted 'a' — tab completion, etc.
+	fb := NewFramebuffer(80, 24)
+	fb.CellAt(0, 0).Rune = 'x'
+	fb.CurX = 5
+	fb.CurY = 0
+
+	p.Confirm(fb)
+	if p.Active() {
+		t.Fatal("predictor should reset on divergence")
+	}
+	if p.curX != 5 {
+		t.Fatalf("cursor should sync to server: got %d, want 5", p.curX)
+	}
+}
+
+func TestPredictorControlCharResets(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("ab"))
+
+	if !p.Active() {
+		t.Fatal("should be active")
+	}
+
+	// Send a control character (enter)
+	p.Keystroke([]byte("\n"))
+	if p.Active() {
+		t.Fatal("control char should reset predictions")
+	}
+}
+
+func TestPredictorEscapeSequenceResets(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("ab"))
+
+	// Escape byte
+	p.Keystroke([]byte{0x1b})
+	if p.Active() {
+		t.Fatal("escape should reset predictions")
+	}
+}
+
+func TestPredictorExpireStale(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("a"))
+
+	// Manually backdate the prediction
+	p.pending[0].at = time.Now().Add(-600 * time.Millisecond)
+
+	p.ExpireStale(time.Now())
+	if p.Active() {
+		t.Fatal("stale prediction should be expired")
+	}
+	if len(p.pending) != 0 {
+		t.Fatalf("expected 0 pending, got %d", len(p.pending))
+	}
+}
+
+func TestPredictorSpaceConfirm(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("hi there"))
+
+	if len(p.pending) != 8 {
+		t.Fatalf("expected 8 predictions, got %d", len(p.pending))
+	}
+
+	// Server shows "hi " — space at position 2 should confirm, not stall
+	fb := NewFramebuffer(80, 24)
+	fb.CellAt(0, 0).Rune = 'h'
+	fb.CellAt(1, 0).Rune = 'i'
+	fb.CellAt(2, 0).Rune = ' '
+	fb.CurX = 3
+	fb.CurY = 0
+
+	p.Confirm(fb)
+	if len(p.pending) != 5 {
+		t.Fatalf("expected 5 pending after confirming 'hi ', got %d", len(p.pending))
+	}
+	if p.pending[0].r != 't' {
+		t.Fatalf("first remaining should be 't', got %c", p.pending[0].r)
+	}
+}
+
+func TestPredictorNoActivityNoCursorOverride(t *testing.T) {
+	p := NewPredictor()
+
+	// Without any predictions, SetCursor should track server
+	p.SetCursor(10, 5)
+	if p.curX != 10 || p.curY != 5 {
+		t.Fatalf("cursor not tracking: got (%d,%d)", p.curX, p.curY)
+	}
+
+	// After typing, SetCursor should not override predicted position
+	p.Keystroke([]byte("x"))
+	p.SetCursor(0, 0)
+	if p.curX != 11 {
+		t.Fatalf("predicted cursor should not be overridden: got %d, want 11", p.curX)
+	}
+}
+
+func TestPredictorMultibyteUTF8(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(0, 0)
+	p.Keystroke([]byte("é")) // 2 bytes
+
+	if len(p.pending) != 1 {
+		t.Fatalf("expected 1 prediction, got %d", len(p.pending))
+	}
+	if p.pending[0].r != 'é' {
+		t.Fatalf("got %c, want é", p.pending[0].r)
+	}
+}
+
+func TestPredictorOverlayDoesNotAffectUnpredictedCells(t *testing.T) {
+	p := NewPredictor()
+	p.SetCursor(5, 0)
+	p.Keystroke([]byte("x"))
+
+	fb := NewFramebuffer(80, 24)
+	fb.CellAt(0, 0).Rune = 'A'
+	fb.CellAt(1, 0).Rune = 'B'
+
+	p.Overlay(fb)
+
+	// Unpredicted cells should be untouched
+	if fb.CellAt(0, 0).Rune != 'A' {
+		t.Fatal("cell 0 should be unchanged")
+	}
+	if fb.CellAt(1, 0).Rune != 'B' {
+		t.Fatal("cell 1 should be unchanged")
+	}
+	// Predicted cell
+	if fb.CellAt(5, 0).Rune != 'x' {
+		t.Fatal("cell 5 should be 'x'")
+	}
+}
+
+// TestPredictorE2EGoServer tests prediction with real mosh transport.
+func TestPredictorE2EGoServer(t *testing.T) {
+	srv, err := NewServer("/bin/sh", 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Serve()
+	defer srv.Close()
+	<-srv.started
+
+	client, err := Dial("127.0.0.1", srv.Port(), srv.KeyBase64())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	// Wait for shell prompt.
+	for i := 0; i < 40; i++ {
+		if out := client.Recv(500 * time.Millisecond); len(out) > 0 {
+			break
+		}
+	}
+
+	pred := NewPredictor()
+	pred.SetCursor(0, 0)
+
+	// Type printable chars — predictions should appear immediately.
+	keys := []byte("echo hello")
+	pred.Keystroke(keys)
+	client.Send(keys)
+
+	if !pred.Active() {
+		t.Fatal("predictor should be active after typing")
+	}
+	if len(pred.pending) != 10 {
+		t.Fatalf("expected 10 predictions, got %d", len(pred.pending))
+	}
+
+	// The predictions can be overlaid on a framebuffer immediately —
+	// this is the "instant echo" the user sees.
+	fb := NewFramebuffer(80, 24)
+	pred.Overlay(fb)
+	if fb.CellAt(0, 0).Rune != 'e' || fb.CellAt(1, 0).Rune != 'c' {
+		t.Fatal("overlay should show predicted characters immediately")
+	}
+
+	// Wait for server to confirm. The server will echo the characters
+	// and predictions should get confirmed.
+	deadline := time.After(10 * time.Second)
+	for pred.Active() {
+		select {
+		case <-deadline:
+			t.Fatalf("predictions not confirmed. %d still pending", len(pred.pending))
+		default:
+		}
+		out := client.Recv(200 * time.Millisecond)
+		if out != nil {
+			// Parse server output into a framebuffer to confirm predictions.
+			// For simplicity, build a minimal framebuffer from ANSI output.
+			serverFB := NewFramebuffer(80, 24)
+			applyANSIToFB(serverFB, out)
+			pred.Confirm(serverFB)
+		}
+		pred.ExpireStale(time.Now())
+	}
+
+	t.Log("prediction E2E with Go server passed")
+}
+
+// applyANSIToFB is a minimal helper that writes characters from ANSI
+// output into a framebuffer. It doesn't parse escape sequences — just
+// extracts printable characters at sequential positions. Good enough
+// for testing prediction confirmation.
+func applyANSIToFB(fb *Framebuffer, data []byte) {
+	x, y := 0, 0
+	i := 0
+	for i < len(data) {
+		if data[i] == 0x1b {
+			// Skip escape sequence
+			i++
+			if i < len(data) && data[i] == '[' {
+				i++
+				for i < len(data) && !((data[i] >= 'A' && data[i] <= 'Z') || (data[i] >= 'a' && data[i] <= 'z')) {
+					i++
+				}
+				if i < len(data) {
+					i++ // skip final byte
+				}
+			}
+			continue
+		}
+		if data[i] == '\r' {
+			x = 0
+			i++
+			continue
+		}
+		if data[i] == '\n' {
+			y++
+			i++
+			continue
+		}
+		r, size := utf8.DecodeRune(data[i:])
+		i += size
+		if r >= 0x20 && x < fb.W && y < fb.H {
+			cell := fb.CellAt(x, y)
+			cell.Rune = r
+			cell.Width = 1
+			x++
+		}
+	}
+	fb.CurX = x
+	fb.CurY = y
+}

server.go

@@ -1,3 +1,5 @@
+//go:build !js
+
 package mosh
 
 import (
@@ -10,26 +12,14 @@ import (
 	"os"
 	"os/exec"
 	"sync"
+	"sync/atomic"
 	"time"
 
+	"github.com/unixshells/vt-go"
 	"github.com/creack/pty"
 )
 
 const (
-	// Direction bits in the 64-bit nonce header.
-	dirToServer = uint64(0)          // client → server
-	dirToClient = uint64(1) << 63    // server → client
-	seqMask     = ^(uint64(1) << 63) // lower 63 bits
-
-	// Minimum wire datagram: 8 (nonce) + 16 (tag) = 24 bytes.
-	minDatagram = 24
-
-	// Maximum payload per UDP datagram.
-	maxPayload = 16384
-
-	// Heartbeat / tick interval for the SSP send loop.
-	tickInterval = 20 * time.Millisecond
-
 	// Server gives up waiting for client after this.
 	associationTimeout = 60 * time.Second
 
@@ -59,6 +49,12 @@ type Server struct {
 	// Transport handles SSP sequencing, fragmentation, and crypto.
 	transport *Transport
 
+	// VT emulator and framebuffer state for CUP-based diffing.
+	emu        *vt.Emulator
+	baseFB     *Framebuffer // what client has (last-acked)
+	sentFB     *Framebuffer // what we last sent (pending ack)
+	curVisible atomic.Bool
+
 	// Remote client address — set on first received datagram.
 	mu         sync.Mutex
 	clientAddr *net.UDPAddr
@@ -154,6 +150,13 @@ func (s *Server) Serve() error {
 		Cols: uint16(s.cols),
 	})
 
+	s.emu = vt.NewEmulator(s.cols, s.rows)
+	s.curVisible.Store(true)
+	s.emu.SetCallbacks(vt.Callbacks{
+		CursorVisibility: func(visible bool) { s.curVisible.Store(visible) },
+	})
+	s.baseFB = NewFramebuffer(s.cols, s.rows)
+
 	var wg sync.WaitGroup
 	wg.Add(3)
 
@@ -196,6 +199,13 @@ func (s *Server) Done() <-chan struct{} {
 func (s *Server) ServeRW(rw io.ReadWriteCloser, resize func(cols, rows uint16)) error {
 	close(s.started)
 
+	s.emu = vt.NewEmulator(s.cols, s.rows)
+	s.curVisible.Store(true)
+	s.emu.SetCallbacks(vt.Callbacks{
+		CursorVisibility: func(visible bool) { s.curVisible.Store(visible) },
+	})
+	s.baseFB = NewFramebuffer(s.cols, s.rows)
+
 	var wg sync.WaitGroup
 	wg.Add(3)
 
@@ -248,6 +258,9 @@ func (s *Server) readIO(r io.Reader, out chan<- []byte) {
 }
 
 func (s *Server) mainLoopRW(rw io.Writer, resize func(cols, rows uint16), ioOutput <-chan []byte, userInput <-chan UserInstruction) {
+	// Wait for the first client datagram. If no client connects within
+	// 60 seconds, this was a failed setup — shut down to release resources.
+	deadline := time.After(associationTimeout)
 	for {
 		s.mu.Lock()
 		addr := s.clientAddr
@@ -258,6 +271,9 @@ func (s *Server) mainLoopRW(rw io.Writer, resize func(cols, rows uint16), ioOutp
 		select {
 		case <-s.done:
 			return
+		case <-deadline:
+			close(s.done)
+			return
 		case <-time.After(100 * time.Millisecond):
 		}
 	}
@@ -265,7 +281,7 @@ func (s *Server) mainLoopRW(rw io.Writer, resize func(cols, rows uint16), ioOutp
 	ticker := time.NewTicker(tickInterval)
 	defer ticker.Stop()
 
-	var pending []byte
+	dirty := false
 
 	for {
 		select {
@@ -276,7 +292,8 @@ func (s *Server) mainLoopRW(rw io.Writer, resize func(cols, rows uint16), ioOutp
 			if !ok {
 				return
 			}
-			pending = append(pending, data...)
+			s.emu.Write(data)
+			dirty = true
 
 		case ui := <-userInput:
 			if len(ui.Keys) > 0 {
@@ -290,18 +307,29 @@ func (s *Server) mainLoopRW(rw io.Writer, resize func(cols, rows uint16), ioOutp
 				if resize != nil {
 					resize(uint16(ui.Width), uint16(ui.Height))
 				}
+				s.emu.Resize(int(ui.Width), int(ui.Height))
+				s.baseFB = NewFramebuffer(int(ui.Width), int(ui.Height))
+				s.sentFB = nil
+				dirty = true
 			}
 
 		case <-ticker.C:
-			if len(pending) > 0 {
-				hi := HostInstruction{
-					Hoststring: pending,
-					EchoAckNum: -1,
-				}
-				diff := marshalHostMessage([]HostInstruction{hi})
-				s.transport.SetPending(diff)
-				pending = nil
+			if s.sentFB != nil && s.transport.AckedByRemote() >= s.transport.SentNum() {
+				s.baseFB = s.sentFB
+				s.sentFB = nil
 			}
+
+			if dirty {
+				currentFB := SnapshotEmulator(s.emu, s.curVisible.Load())
+				diffBytes := currentFB.Diff(s.baseFB)
+				if len(diffBytes) > 0 {
+					hi := HostInstruction{Hoststring: diffBytes, EchoAckNum: -1}
+					s.transport.SetPending(marshalHostMessage([]HostInstruction{hi}))
+					s.sentFB = currentFB
+				}
+				dirty = false
+			}
+
 			datagrams := s.transport.Tick()
 			s.sendDatagrams(datagrams)
 		}
@@ -331,10 +359,13 @@ func (s *Server) Close() {
 	}
 }
 
-// mainLoop is the SSP event loop. It maintains the terminal framebuffer,
-// processes user input, diffs the screen, and drives the transport.
+// mainLoop is the SSP event loop. It feeds PTY output to a VT emulator,
+// snapshots the framebuffer, diffs against the last-acked state, and
+// sends CUP-based updates via the transport.
 func (s *Server) mainLoop(ptyOutput <-chan []byte, userInput <-chan UserInstruction) {
-	// Wait for client to connect before processing.
+	// Wait for the first client datagram. If no client connects within
+	// 60 seconds, this was a failed setup — shut down to release resources.
+	deadline := time.After(associationTimeout)
 	for {
 		s.mu.Lock()
 		addr := s.clientAddr
@@ -345,18 +376,17 @@ func (s *Server) mainLoop(ptyOutput <-chan []byte, userInput <-chan UserInstruct
 		select {
 		case <-s.done:
 			return
+		case <-deadline:
+			close(s.done)
+			return
 		case <-time.After(100 * time.Millisecond):
 		}
 	}
 
-	// Terminal state: we maintain a simple passthrough buffer.
-	// PTY output is accumulated and sent as HostBytes diffs.
-	// For full mosh compatibility, the diff is the raw terminal output
-	// wrapped in the HostMessage protobuf.
 	ticker := time.NewTicker(tickInterval)
 	defer ticker.Stop()
 
-	var pending []byte
+	dirty := false
 
 	for {
 		select {
@@ -367,14 +397,13 @@ func (s *Server) mainLoop(ptyOutput <-chan []byte, userInput <-chan UserInstruct
 			if !ok {
 				return
 			}
-			pending = append(pending, data...)
+			s.emu.Write(data)
+			dirty = true
 
 		case ui := <-userInput:
-			// Handle keystrokes.
 			if len(ui.Keys) > 0 {
 				s.ptmx.Write(ui.Keys)
 			}
-			// Handle resize.
 			if ui.Width > 0 && ui.Height > 0 {
 				s.mu.Lock()
 				s.cols = int(ui.Width)
@@ -384,20 +413,28 @@ func (s *Server) mainLoop(ptyOutput <-chan []byte, userInput <-chan UserInstruct
 					Cols: uint16(ui.Width),
 					Rows: uint16(ui.Height),
 				})
+				s.emu.Resize(int(ui.Width), int(ui.Height))
+				s.baseFB = NewFramebuffer(int(ui.Width), int(ui.Height))
+				s.sentFB = nil
+				dirty = true
 			}
 
 		case <-ticker.C:
-			// Build diff and send via transport.
-			if len(pending) > 0 {
-				hi := HostInstruction{
-					Hoststring: pending,
-					EchoAckNum: -1,
-				}
-				diff := marshalHostMessage([]HostInstruction{hi})
+			// Advance base when client acks all pending.
+			if s.sentFB != nil && s.transport.AckedByRemote() >= s.transport.SentNum() {
+				s.baseFB = s.sentFB
+				s.sentFB = nil
+			}
 
-				// Wrap in TransportInstruction via transport.
-				s.transport.SetPending(diff)
-				pending = nil
+			if dirty {
+				currentFB := SnapshotEmulator(s.emu, s.curVisible.Load())
+				diffBytes := currentFB.Diff(s.baseFB)
+				if len(diffBytes) > 0 {
+					hi := HostInstruction{Hoststring: diffBytes, EchoAckNum: -1}
+					s.transport.SetPending(marshalHostMessage([]HostInstruction{hi}))
+					s.sentFB = currentFB
+				}
+				dirty = false
 			}
 
 			datagrams := s.transport.Tick()

server_test.go

@@ -1,3 +1,5 @@
+//go:build !js
+
 package mosh
 
 import (
@@ -8,6 +10,7 @@ import (
 	"net"
 	"os"
 	"os/exec"
+	"regexp"
 	"strconv"
 	"strings"
 	"testing"
@@ -16,6 +19,13 @@ import (
 	"github.com/creack/pty"
 )
 
+var ansiRe = regexp.MustCompile(`\x1b\[[0-9;?]*[A-Za-z]`)
+
+// stripANSI removes CSI escape sequences from s.
+func stripANSI(s string) string {
+	return ansiRe.ReplaceAllString(s, "")
+}
+
 func TestGenerateKey(t *testing.T) {
 	key, b64, err := GenerateKey()
 	if err != nil {
@@ -175,13 +185,13 @@ func TestServerE2E(t *testing.T) {
 	for !found {
 		select {
 		case <-deadline:
-			t.Fatalf("marker not echoed. Got: %q", allOutput)
+			t.Fatalf("marker not echoed. Got: %q", stripANSI(allOutput))
 		default:
 		}
 		payload := recvClient(500 * time.Millisecond)
 		if payload != nil {
 			allOutput += string(payload)
-			if strings.Contains(allOutput, marker) {
+			if strings.Contains(stripANSI(allOutput), marker) {
 				found = true
 			}
 		}
@@ -504,14 +514,14 @@ func TestServeRW(t *testing.T) {
 	for {
 		select {
 		case <-deadline:
-			t.Fatalf("did not receive terminal output via ServeRW, got: %q", allOutput)
+			t.Fatalf("did not receive terminal output via ServeRW, got: %q", stripANSI(allOutput))
 		default:
 		}
 		payload := recvClient(500 * time.Millisecond)
 		if payload != nil {
 			allOutput += string(payload)
 		}
-		if strings.Contains(allOutput, "HELLO_RW") {
+		if strings.Contains(stripANSI(allOutput), "HELLO_RW") {
 			break
 		}
 		sendClient(nil)
@@ -625,14 +635,14 @@ func TestServeRW(t *testing.T) {
 	for {
 		select {
 		case <-deadline2:
-			t.Fatalf("did not receive terminal output, got: %q", allOutput2)
+			t.Fatalf("did not receive terminal output, got: %q", stripANSI(allOutput2))
 		default:
 		}
 		payload := recvClient2(500 * time.Millisecond)
 		if payload != nil {
 			allOutput2 += string(payload)
 		}
-		if strings.Contains(allOutput2, "OUTPUT_OK") {
+		if strings.Contains(stripANSI(allOutput2), "OUTPUT_OK") {
 			break
 		}
 		sendClient2(nil)

transport.go

@@ -26,13 +26,20 @@ type Transport struct {
 
 	// Outgoing state (SSP §3).
 	sentNum      uint64 // newest state we've sent (new_num)
-	ackedNum     uint64 // newest state the remote has acknowledged (ack from remote)
-	pendingDiff  []byte // diff payload waiting to be sent
+	ackedByRemote uint64 // newest state the remote has acknowledged
+	pendingDiff    []byte // diff payload waiting to be sent
+	diffSent       bool   // true = pendingDiff has been sent at least once
+	diffOldNum     uint64 // locked oldNum for all diffs until base advances
+	hasPendingBase bool   // true = diffOldNum is locked
+	pendingDataAck bool   // true = send ack ASAP (received data, not just ack)
 
-	// Incoming state.
-	recvNum      uint64 // newest state we've received from remote (their new_num)
-	recvAck      uint64 // what we've told the remote we received (our ack_num)
-	throwawayNum uint64 // oldest state we still hold
+	// Incoming state — list of received state nums for old_num validation.
+	receivedNums   []uint64 // ordered list of state nums we have
+	ackNum         uint64   // latest received state num
+	sentAckNum     uint64   // last ackNum we actually sent on wire
+	throwawayNum   uint64   // oldest state we still hold
+	lastRecvOldNum uint64   // oldNum from most recently received diff
+	lastRecvNewNum uint64   // newNum from most recently received diff
 
 	// Sequence counter for the crypto layer (independent of SSP state numbering).
 	seqOut      uint64
@@ -67,10 +74,11 @@ const (
 // NewTransport creates a transport. isServer determines direction bits.
 func NewTransport(ocb *OCB, isServer bool) *Transport {
 	t := &Transport{
-		ocb:      ocb,
-		rto:      initialRTO,
-		lastSend: time.Now(),
-		lastRecv: time.Now(),
+		ocb:          ocb,
+		rto:          initialRTO,
+		lastSend:     time.Now(),
+		lastRecv:     time.Now(),
+		receivedNums: []uint64{0}, // start with state 0
 	}
 	if isServer {
 		t.toRemote = dirToClient
@@ -107,9 +115,19 @@ func (t *Transport) HasCap(bit byte) bool {
 	return (t.localCaps[idx] & t.remoteCaps[idx] & bit) != 0
 }
 
+// ForceNextSend forces the next tick to send, even with no pending diff.
+func (t *Transport) ForceNextSend() {
+	t.mu.Lock()
+	t.lastSend = time.Time{} // zero time, always expired
+	t.mu.Unlock()
+}
+
 // SetPending sets the diff payload to send on the next tick.
 func (t *Transport) SetPending(diff []byte) {
 	t.mu.Lock()
+	if len(diff) > 0 {
+		t.diffSent = false
+	}
 	t.pendingDiff = diff
 	t.mu.Unlock()
 }
@@ -124,29 +142,44 @@ func (t *Transport) Tick() [][]byte {
 
 	// Decide if we should send.
 	haveDiff := len(t.pendingDiff) > 0
-	needAck := t.recvNum > t.recvAck
-	expired := now.Sub(t.lastSend) >= t.rto
+	haveNewDiff := haveDiff && !t.diffSent
+	needAck := t.ackNum > t.sentAckNum
+	sinceLastSend := now.Sub(t.lastSend)
+	expired := sinceLastSend >= t.rto
+	urgentAck := t.pendingDataAck
 
-	if !haveDiff && !needAck && !expired {
+	shouldSend := haveNewDiff || needAck || expired || urgentAck
+	if !shouldSend {
 		return nil
 	}
 
 	// Build TransportInstruction.
-	if haveDiff {
+	if haveNewDiff {
 		t.sentNum++
+		t.diffSent = true
+		if !t.hasPendingBase {
+			t.diffOldNum = t.ackedByRemote
+			t.hasPendingBase = true
+		}
+	}
+	t.pendingDataAck = false
+
+	oldNum := t.ackedByRemote
+	if haveDiff {
+		oldNum = t.diffOldNum // use locked oldNum for diff retransmissions
 	}
 
 	ti := TransportInstruction{
-		ProtocolVersion: 2, // mosh protocol version
-		OldNum:          t.ackedNum,
+		ProtocolVersion: 2,
+		OldNum:          oldNum,
 		NewNum:          t.sentNum,
-		AckNum:          t.recvNum,
-		ThrowawayNum:    t.throwawayNum,
+		AckNum:          t.ackNum,
+		ThrowawayNum:    0, // client doesn't tell server to throw away states
 		Diff:            t.pendingDiff,
 		LatchCaps:       t.localCaps,
 	}
-	t.recvAck = t.recvNum
-	t.pendingDiff = nil
+	t.sentAckNum = t.ackNum
+	// Do NOT nil pendingDiff — keep for retransmission until server acks.
 
 	// Marshal → compress → fragment → encrypt.
 	pbData := ti.Marshal()
@@ -250,22 +283,107 @@ func (t *Transport) Recv(wire []byte) []byte {
 		t.mu.Unlock()
 	}
 
-	// Process SSP fields.
+	// Process SSP fields (matching upstream mosh recv logic).
 	t.mu.Lock()
-	if ti.AckNum > t.ackedNum {
-		t.ackedNum = ti.AckNum
+	defer t.mu.Unlock()
+
+	// Process ack from remote.
+	if ti.AckNum > t.ackedByRemote {
+		t.ackedByRemote = ti.AckNum
+		if t.ackedByRemote >= t.sentNum && t.pendingDiff != nil {
+			t.pendingDiff = nil
+			t.diffSent = false
+			t.hasPendingBase = false
+		}
 	}
-	if ti.NewNum > t.recvNum {
-		t.recvNum = ti.NewNum
+
+	// Check if we already have new_num (dedup).
+	for _, n := range t.receivedNums {
+		if n == ti.NewNum {
+			return nil
+		}
 	}
+
+	// Reject diffs that do not chain from our latest acknowledged state.
+	// Mosh diffs MUST be sequential: each diff's oldNum must equal the
+	// newNum of the most recently accepted diff. Accepting diffs from
+	// stale bases (oldNum != ackNum) causes overlapping ANSI sequences
+	// that produce character ghosting and garbled output.
+	if ti.OldNum != t.ackNum {
+		return nil
+	}
+
+	// Process throwaway.
 	if ti.ThrowawayNum > t.throwawayNum {
 		t.throwawayNum = ti.ThrowawayNum
+		filtered := t.receivedNums[:0]
+		for _, n := range t.receivedNums {
+			if n >= t.throwawayNum {
+				filtered = append(filtered, n)
+			}
+		}
+		t.receivedNums = filtered
+	}
+
+	// Track oldNum/newNum for state management.
+	t.lastRecvOldNum = ti.OldNum
+	t.lastRecvNewNum = ti.NewNum
+
+	// Add new state.
+	t.receivedNums = append(t.receivedNums, ti.NewNum)
+
+	// Bound the list.
+	if len(t.receivedNums) > 128 {
+		t.receivedNums = t.receivedNums[1:]
+	}
+
+	// Update ack num to latest received state.
+	t.ackNum = ti.NewNum
+
+	// Trigger immediate ack when we receive data.
+	if len(ti.Diff) > 0 {
+		t.pendingDataAck = true
 	}
-	t.mu.Unlock()
 
 	return ti.Diff
 }
 
+// AckedByRemote returns the highest state number the remote has acked.
+func (t *Transport) AckedByRemote() uint64 {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.ackedByRemote
+}
+
+// SentNum returns the current sent state number.
+func (t *Transport) SentNum() uint64 {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.sentNum
+}
+
+// LastRecvOldNum returns the oldNum from the most recently received diff.
+func (t *Transport) LastRecvOldNum() uint64 {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.lastRecvOldNum
+}
+
+// LastRecvNewNum returns the newNum from the most recently received diff.
+func (t *Transport) LastRecvNewNum() uint64 {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.lastRecvNewNum
+}
+
+// ThrowawayNum returns the server's throwaway number — states below this
+// are no longer referenced by the server and can be safely pruned.
+func (t *Transport) ThrowawayNum() uint64 {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.throwawayNum
+}
+
 // LastRecv returns the time of the last received datagram.
 func (t *Transport) LastRecv() time.Time {
 	t.mu.Lock()