fix key repeat: only create new state when previous is acked (match Dart client)

The Dart mosh client only calls sendNew() when hasPendingState is false.
The Go client was calling SetPending on every tick with dirty=true, which
reset diffSent and caused new sequence numbers for the same cumulative
payload. Now tick() checks AckedByRemote >= SentNum before calling
SetPending — at most one unacked state in flight, matching the Dart
client's behavior.

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,19 +57,50 @@ 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
@@ -69,21 +108,21 @@ func Dial(host string, port int, key string) (*Client, error) {
 
 // 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,12 +170,50 @@ 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)

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,135 @@
+//go:build js && wasm
+
+package main
+
+import (
+	"encoding/base64"
+	"fmt"
+	"sync"
+	"syscall/js"
+
+	mosh "github.com/unixshells/mosh-go"
+)
+
+func main() {
+	js.Global().Set("moshConnect", js.FuncOf(moshConnect))
+	select {} // keep alive
+}
+
+// moshConnect(url, key) → Promise<MoshSession>
+// url: WebTransport URL (e.g., "https://relay.example.com/mosh/user/device")
+// key: base64-encoded mosh key
+func moshConnect(this js.Value, args []js.Value) interface{} {
+	if len(args) < 2 {
+		return reject("moshConnect requires (url, key)")
+	}
+
+	url := args[0].String()
+	key := args[1].String()
+
+	handler := js.FuncOf(func(this js.Value, pargs []js.Value) interface{} {
+		resolve := pargs[0]
+		rejectFn := pargs[1]
+
+		go func() {
+			session, err := newSession(url, key)
+			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
+	mu     sync.Mutex
+	closed bool
+}
+
+func newSession(url, key string) (*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
+	}
+
+	// Use DialConnManual — no internal sendLoop. JS drives Tick() via setInterval.
+	client, err := mosh.DialConnManual(conn, ocb)
+	if err != nil {
+		conn.Close()
+		return nil, err
+	}
+
+	return &session{client: client, conn: conn}, 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
+		}
+		s.client.Send([]byte(args[0].String()))
+		s.client.Tick()
+		return nil
+	}))
+
+	obj.Set("resize", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		if len(args) < 2 {
+			return nil
+		}
+		s.client.Resize(uint16(args[0].Int()), uint16(args[1].Int()))
+		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() — check for output. Called by JS setInterval.
+	// Returns string if output available, null otherwise.
+	// Using JS-driven polling avoids Go goroutine scheduler starvation.
+	obj.Set("poll", js.FuncOf(func(this js.Value, args []js.Value) interface{} {
+		out := s.client.Recv(0)
+		if out == nil {
+			return js.Null()
+		}
+		return string(out)
+	}))
+
+	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
+}
+
+func reject(msg string) js.Value {
+	return js.Global().Get("Promise").Call("reject", msg)
+}

cmd/mosh-wasm/webtransport.go

@@ -0,0 +1,158 @@
+//go:build js && wasm
+
+package main
+
+import (
+	"errors"
+	"sync"
+	"syscall/js"
+	"time"
+)
+
+// wtConn wraps a browser WebTransport connection as a mosh.Conn.
+type wtConn struct {
+	transport js.Value
+	writer    js.Value // persistent writable stream writer
+	incoming  chan []byte
+	done      chan struct{}
+	once      sync.Once
+
+	mu       sync.Mutex
+	deadline time.Time
+}
+
+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{}),
+	}
+
+	// Start reading datagrams via a persistent JS callback.
+	c.startReader()
+
+	return c, nil
+}
+
+// startReader sets up a self-chaining read loop using persistent callbacks.
+func (c *wtConn) startReader() {
+	reader := c.transport.Get("datagrams").Get("readable").Call("getReader")
+
+	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))
+		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.
+		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) {
+	arr := js.Global().Get("Uint8Array").New(len(b))
+	js.CopyBytesToJS(arr, b)
+	c.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.transport.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,15 @@
+//go:build !js
+
 package mosh
 
 import (
+	"image/color"
 	"strconv"
+	"unicode/utf8"
+
+	uv "github.com/charmbracelet/ultraviolet"
+	"github.com/charmbracelet/x/ansi"
+	"github.com/charmbracelet/x/vt"
 )
 
 // Framebuffer represents a terminal screen state for mosh SSP.
@@ -361,3 +369,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&uv.AttrBold != 0
+			cell.Attr.Dim = c.Style.Attrs&uv.AttrFaint != 0
+			cell.Attr.Italic = c.Style.Attrs&uv.AttrItalic != 0
+			cell.Attr.Blink = c.Style.Attrs&uv.AttrBlink != 0
+			cell.Attr.Reverse = c.Style.Attrs&uv.AttrReverse != 0
+			cell.Attr.Strike = c.Style.Attrs&uv.AttrStrikethrough != 0
+			cell.Attr.Under = c.Style.Underline != uv.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

@@ -3,9 +3,28 @@ module github.com/unixshells/mosh-go
 go 1.26.0
 
 require (
+	github.com/charmbracelet/ultraviolet v0.0.0-20260303162955-0b88c25f3fff
+	github.com/charmbracelet/x/ansi v0.11.6
+	github.com/charmbracelet/x/vt v0.0.0-20260305213658-fe36e8c10185
 	github.com/creack/pty v1.1.24
 	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/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/vt v0.0.0-20260305213658-fe36e8c10185 h1:KYS9lWBROkdkuyc0bdQuS8C86Me3OAgjUAKcB4zArlg=
+github.com/charmbracelet/x/vt v0.0.0-20260305213658-fe36e8c10185/go.mod h1:2djWW5EeP5QlpztEYYbbyyzUyyxVD71mWVQguNISHcM=
+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/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=

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/charmbracelet/x/vt"
 	"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,18 @@ 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
 
 	// Sequence counter for the crypto layer (independent of SSP state numbering).
 	seqOut      uint64
@@ -67,10 +72,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 +113,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 +140,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 +281,84 @@ 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
+		}
 	}
+
+	// Check if we have old_num (required to apply diff).
+	hasOld := false
+	for _, n := range t.receivedNums {
+		if n == ti.OldNum {
+			hasOld = true
+			break
+		}
+	}
+	if !hasOld {
+		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
+	}
+
+	// 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
+}
+
 // LastRecv returns the time of the last received datagram.
 func (t *Transport) LastRecv() time.Time {
 	t.mu.Lock()