fix(remote): evict stale persistent SSH bridge leases

daemon/remote/README.md

@@ -62,6 +62,12 @@ Remote slot files:
 3. `~/.cmux/daemon/<version>/<slot>/daemon.lock` single-owner lock.
 4. `~/.cmux/daemon/<version>/<slot>/daemon.log` startup and crash diagnostics.
 
+Each `serve --stdio --persistent` bridge includes a fresh `bridge_lease_id` in
+its authenticated socket handshake. The persistent server tracks authenticated
+bridge connections and lets the newest authenticated bridge take over the slot,
+closing older connections (including half-open SSH bridges) without touching
+the persistent PTY sessions.
+
 PTY lifecycle:
 1. A local attach creates or reuses a named `pty.*` session in the persistent daemon.
 2. If the local surface closes, the stdio proxy disconnects and its attachment detaches, but the PTY process and bounded scrollback remain in the daemon.

daemon/remote/cmd/cmuxd-remote/main.go

@@ -924,7 +924,16 @@ func runPersistentStdioProxy(stdin io.Reader, stdout, stderr io.Writer, slot str
 	if err := ensurePersistentDaemonRunning(paths, token, leasePort, stderr); err != nil {
 		return err
 	}
-	conn, err := dialPersistentDaemon(paths.socket, token)
+	bridgeLeaseID, err := newPersistentDaemonBridgeLeaseID()
+	if err != nil {
+		logPersistentDaemonEvent(
+			stderr,
+			"bridge_lease_generation_failed",
+			"error_category", persistentDaemonErrorCategory(err),
+		)
+		return errors.New(persistentDaemonBridgeLeaseError)
+	}
+	conn, err := dialPersistentDaemonWithBridgeLease(paths.socket, token, bridgeLeaseID)
 	if err != nil {
 		return err
 	}
@@ -1230,6 +1239,7 @@ func servePersistentDaemonWithVerifierConfig(
 ) error {
 	hub := newWebSocketPTYHub(wsPTYServerConfig{}, stderr)
 	defer hub.closeAll()
+	bridgeLeases := newPersistentDaemonBridgeLeaseRegistry()
 	var activeConnections int64
 	var idleSince time.Time
 	var slotLeaseObserved bool
@@ -1315,7 +1325,7 @@ func servePersistentDaemonWithVerifierConfig(
 		atomic.AddInt64(&activeConnections, 1)
 		go func() {
 			defer atomic.AddInt64(&activeConnections, -1)
-			handlePersistentDaemonConn(conn, verifier, hub, stderr, requestShutdown)
+			handlePersistentDaemonConn(conn, verifier, hub, stderr, requestShutdown, bridgeLeases)
 		}()
 	}
 }
@@ -1365,6 +1375,7 @@ func handlePersistentDaemonConn(
 	hub *wsPTYHub,
 	stderr io.Writer,
 	requestShutdown func(),
+	bridgeLeases ...*persistentDaemonBridgeLeaseRegistry,
 ) {
 	handlePersistentDaemonConnWithAuthTimeout(
 		conn,
@@ -1373,6 +1384,7 @@ func handlePersistentDaemonConn(
 		stderr,
 		persistentDaemonAuthTimeout,
 		requestShutdown,
+		bridgeLeases...,
 	)
 }
 
@@ -1383,9 +1395,16 @@ func handlePersistentDaemonConnWithAuthTimeout(
 	stderr io.Writer,
 	timeout time.Duration,
 	requestShutdown func(),
+	bridgeLeases ...*persistentDaemonBridgeLeaseRegistry,
 ) {
+	var bridgeLeaseRegistry *persistentDaemonBridgeLeaseRegistry
+	if len(bridgeLeases) > 0 {
+		bridgeLeaseRegistry = bridgeLeases[0]
+	}
 	defer conn.Close()
+	defer bridgeLeaseRegistry.release(conn)
 	defer logPersistentDaemonEvent(stderr, "connection_closed")
+	bridgeLeaseRegistry.register(conn)
 	logPersistentDaemonEvent(stderr, "connection_accepted")
 	if timeout > 0 {
 		if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil {
@@ -1400,14 +1419,34 @@ func handlePersistentDaemonConnWithAuthTimeout(
 	}
 	reader := bufio.NewReaderSize(conn, 64*1024)
 	writer := &stdioFrameWriter{writer: bufio.NewWriter(conn)}
-	if err := authenticatePersistentDaemonConn(reader, writer, verifier); err != nil {
+	evictedBridgeConnections := 0
+	if err := authenticatePersistentDaemonConnWithLease(
+		reader,
+		writer,
+		verifier,
+		func(leaseID string) error {
+			if bridgeLeaseRegistry == nil {
+				return nil
+			}
+			evicted, claimErr := bridgeLeaseRegistry.claim(conn, leaseID)
+			evictedBridgeConnections = evicted
+			return claimErr
+		},
+	); err != nil {
 		logPersistentDaemonEvent(
 			stderr,
 			"connection_rejected",
 			"reason", persistentDaemonAuthenticationFailureReason(err),
 		)
 		return
 	}
+	if evictedBridgeConnections > 0 {
+		logPersistentDaemonEvent(
+			stderr,
+			"bridge_lease_takeover",
+			"evicted_connections", strconv.Itoa(evictedBridgeConnections),
+		)
+	}
 	logPersistentDaemonEvent(stderr, "connection_authenticated")
 	if timeout > 0 {
 		if err := conn.SetDeadline(time.Time{}); err != nil {
@@ -1446,12 +1485,23 @@ func persistentDaemonAuthenticationFailureReason(err error) string {
 		"authentication method is invalid",
 		"authentication token is invalid":
 		return err.Error()
+	case "persistent daemon bridge lease rejected":
+		return "bridge_lease_rejected"
 	default:
 		return persistentDaemonErrorCategory(err)
 	}
 }
 
 func authenticatePersistentDaemonConn(reader *bufio.Reader, writer *stdioFrameWriter, verifier persistentDaemonTokenVerifier) error {
+	return authenticatePersistentDaemonConnWithLease(reader, writer, verifier, nil)
+}
+
+func authenticatePersistentDaemonConnWithLease(
+	reader *bufio.Reader,
+	writer *stdioFrameWriter,
+	verifier persistentDaemonTokenVerifier,
+	onBridgeLease func(string) error,
+) error {
 	line, oversized, err := readRPCFrame(reader, maxRPCFrameBytes)
 	if err != nil || oversized {
 		rejection := fmt.Errorf("authentication frame read failed: %w", err)
@@ -1503,6 +1553,20 @@ func authenticatePersistentDaemonConn(reader *bufio.Reader, writer *stdioFrameWr
 			},
 		}, errors.New("authentication token is invalid"))
 	}
+	bridgeLeaseID, _ := getStringParam(req.Params, persistentDaemonBridgeLeaseParam)
+	bridgeLeaseID = strings.TrimSpace(bridgeLeaseID)
+	if bridgeLeaseID != "" && onBridgeLease != nil {
+		if err := onBridgeLease(bridgeLeaseID); err != nil {
+			return writePersistentDaemonAuthRejection(writer, rpcResponse{
+				ID: req.ID,
+				OK: false,
+				Error: &rpcError{
+					Code:    "unauthorized",
+					Message: persistentDaemonBridgeLeaseError,
+				},
+			}, errors.New("persistent daemon bridge lease rejected"))
+		}
+	}
 	if err := writer.writeResponse(rpcResponse{
 		ID: req.ID,
 		OK: true,
@@ -1615,22 +1679,41 @@ func runRPCServerWithReader(
 }
 
 func dialPersistentDaemon(socketPath string, token string) (net.Conn, error) {
-	conn, err := net.DialTimeout("unix", socketPath, 2*time.Second)
+	return dialPersistentDaemonWithBridgeLease(socketPath, token, "")
+}
+
+func dialPersistentDaemonWithBridgeLease(socketPath string, token string, bridgeLeaseID string) (net.Conn, error) {
+	dialer := net.Dialer{Timeout: 2 * time.Second}
+	conn, err := dialer.DialContext(context.Background(), "unix", socketPath)
 	if err != nil {
 		return nil, err
 	}
-	if err := authenticatePersistentDaemonClient(conn, token); err != nil {
+	if err := authenticatePersistentDaemonClientWithBridgeLease(conn, token, bridgeLeaseID); err != nil {
 		_ = conn.Close()
 		return nil, err
 	}
 	return conn, nil
 }
 
-func authenticatePersistentDaemonClient(conn net.Conn, token string) error {
-	return authenticatePersistentDaemonClientWithTimeout(conn, token, persistentDaemonAuthTimeout)
+func authenticatePersistentDaemonClientWithTimeout(conn net.Conn, token string, timeout time.Duration) error {
+	return authenticatePersistentDaemonClientWithTimeoutAndBridgeLease(conn, token, timeout, "")
 }
 
-func authenticatePersistentDaemonClientWithTimeout(conn net.Conn, token string, timeout time.Duration) error {
+func authenticatePersistentDaemonClientWithBridgeLease(conn net.Conn, token string, bridgeLeaseID string) error {
+	return authenticatePersistentDaemonClientWithTimeoutAndBridgeLease(
+		conn,
+		token,
+		persistentDaemonAuthTimeout,
+		bridgeLeaseID,
+	)
+}
+
+func authenticatePersistentDaemonClientWithTimeoutAndBridgeLease(
+	conn net.Conn,
+	token string,
+	timeout time.Duration,
+	bridgeLeaseID string,
+) error {
 	if timeout > 0 {
 		if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil {
 			return err
@@ -1639,12 +1722,16 @@ func authenticatePersistentDaemonClientWithTimeout(conn net.Conn, token string,
 	}
 
 	writer := bufio.NewWriter(conn)
+	params := map[string]any{
+		"token": token,
+	}
+	if normalizedLeaseID := strings.TrimSpace(bridgeLeaseID); normalizedLeaseID != "" {
+		params[persistentDaemonBridgeLeaseParam] = normalizedLeaseID
+	}
 	request := rpcRequest{
 		ID:     "auth",
 		Method: persistentDaemonAuthMethod,
-		Params: map[string]any{
-			"token": token,
-		},
+		Params: params,
 	}
 	data, err := json.Marshal(request)
 	if err != nil {
@@ -1989,10 +2076,11 @@ func (s *rpcServer) handleProxyOpen(req rpcRequest) rpcResponse {
 		timeoutMs = parsed
 	}
 
-	conn, err := net.DialTimeout(
+	dialer := net.Dialer{Timeout: time.Duration(timeoutMs) * time.Millisecond}
+	conn, err := dialer.DialContext(
+		context.Background(),
 		"tcp",
 		net.JoinHostPort(host, strconv.Itoa(port)),
-		time.Duration(timeoutMs)*time.Millisecond,
 	)
 	if err != nil {
 		return rpcResponse{

daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"net"
+	"strings"
+	"sync"
+)
+
+const (
+	persistentDaemonBridgeLeaseParam = "bridge_lease_id"
+	persistentDaemonBridgeLeaseMax   = 128
+	persistentDaemonBridgeLeaseError = "persistent daemon connection could not be established; reconnect and try again"
+)
+
+func newPersistentDaemonBridgeLeaseID() (string, error) {
+	raw := make([]byte, 16)
+	if _, err := rand.Read(raw); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(raw), nil
+}
+
+// persistentDaemonBridgeLeaseRegistry tracks every accepted stdio connection
+// and the one connection that most recently claimed the slot. A new claimed
+// bridge is authoritative: it closes all older connections, which makes
+// takeover work even when an older binary did not send a lease id.
+type persistentDaemonBridgeLeaseRegistry struct {
+	mu          sync.Mutex
+	connections map[net.Conn]struct{}
+	holder      net.Conn
+}
+
+func newPersistentDaemonBridgeLeaseRegistry() *persistentDaemonBridgeLeaseRegistry {
+	return &persistentDaemonBridgeLeaseRegistry{
+		connections: make(map[net.Conn]struct{}),
+	}
+}
+
+func (r *persistentDaemonBridgeLeaseRegistry) register(conn net.Conn) {
+	if r == nil || conn == nil {
+		return
+	}
+	r.mu.Lock()
+	if r.connections == nil {
+		r.connections = make(map[net.Conn]struct{})
+	}
+	r.connections[conn] = struct{}{}
+	r.mu.Unlock()
+}
+
+func (r *persistentDaemonBridgeLeaseRegistry) claim(conn net.Conn, leaseID string) (int, error) {
+	if r == nil || conn == nil {
+		return 0, errors.New("persistent daemon bridge lease registry is unavailable")
+	}
+	leaseID = strings.TrimSpace(leaseID)
+	if leaseID == "" {
+		return 0, errors.New("persistent daemon bridge lease id is required")
+	}
+	if len(leaseID) > persistentDaemonBridgeLeaseMax {
+		return 0, errors.New("persistent daemon bridge lease id is too long")
+	}
+
+	r.mu.Lock()
+	if r.connections == nil {
+		r.connections = make(map[net.Conn]struct{})
+	}
+	r.connections[conn] = struct{}{}
+	evicted := make([]net.Conn, 0, len(r.connections)-1)
+	for candidate := range r.connections {
+		if candidate == conn {
+			continue
+		}
+		evicted = append(evicted, candidate)
+		delete(r.connections, candidate)
+	}
+	r.holder = conn
+	r.mu.Unlock()
+
+	for _, candidate := range evicted {
+		_ = candidate.Close()
+	}
+	return len(evicted), nil
+}
+
+func (r *persistentDaemonBridgeLeaseRegistry) release(conn net.Conn) {
+	if r == nil || conn == nil {
+		return
+	}
+	r.mu.Lock()
+	delete(r.connections, conn)
+	if r.holder == conn {
+		r.holder = nil
+	}
+	r.mu.Unlock()
+}