save work

* Use IsEmpty rather than nullcheck for scbus types

src/app/common/modals/viewremoteconndetail.tsx

@@ -259,8 +259,10 @@ class ViewRemoteConnDetailModal extends React.Component<{}, {}> {
             message = "Connected and ready to run commands.";
         } else if (remote.status == "connecting") {
             message = remote.waitingforpassword ? "Connecting, waiting for user-input..." : "Connecting...";
-            let connectTimeout = remote.connecttimeout ?? 0;
-            message = message + " (" + connectTimeout + "s)";
+            if (remote.countdownactive) {
+                let connectTimeout = remote.connecttimeout ?? 0;
+                message = message + " (" + connectTimeout + "s)";
+            }
         } else if (remote.status == "disconnected") {
             message = "Disconnected";
         } else if (remote.status == "error") {

src/models/model.ts

@@ -715,7 +715,7 @@ class Model {
         return this.ws.open.get();
     }
 
-    runUpdate(genUpdate: UpdateMessage, interactive: boolean) {
+    runUpdate(genUpdate: UpdatePacket, interactive: boolean) {
         mobx.action(() => {
             const oldContext = this.getUIContext();
             try {
@@ -727,8 +727,9 @@ class Model {
             const newContext = this.getUIContext();
             if (oldContext.sessionid != newContext.sessionid || oldContext.screenid != newContext.screenid) {
                 this.inputModel.resetInput();
-                if (!("ptydata64" in genUpdate)) {
-                    const reversedGenUpdate = genUpdate.slice().reverse();
+                if (genUpdate.type == "model") {
+                    const modelUpdate = genUpdate as ModelUpdatePacket;
+                    const reversedGenUpdate = modelUpdate.data.slice().reverse();
                     const lastCmdLine = reversedGenUpdate.find((update) => "cmdline" in update);
                     if (lastCmdLine) {
                         // TODO a bit of a hack since this update gets applied in runUpdate_internal.
@@ -768,20 +769,12 @@ class Model {
     }
 
     updateActiveSession(sessionId: string): void {
-        const [oldActiveSessionId, oldActiveScreenId] = this.getActiveIds();
-
         if (sessionId != null) {
             const newSessionId = sessionId;
             if (this.activeSessionId.get() != newSessionId) {
                 this.activeSessionId.set(newSessionId);
             }
         }
-        const [newActiveSessionId, newActiveScreenId] = this.getActiveIds();
-        if (oldActiveSessionId != newActiveSessionId || oldActiveScreenId != newActiveScreenId) {
-            this.activeMainView.set("session");
-            this.deactivateScreenLines();
-            this.ws.watchScreen(newActiveSessionId, newActiveScreenId);
-        }
     }
 
     updateScreenNumRunningCommands(numRunningCommandUpdates: ScreenNumRunningCommandsUpdateType[]) {
@@ -796,9 +789,9 @@ class Model {
         }
     }
 
-    runUpdate_internal(genUpdate: UpdateMessage, uiContext: UIContextType, interactive: boolean) {
-        if ("ptydata64" in genUpdate) {
-            const ptyMsg: PtyDataUpdateType = genUpdate;
+    runUpdate_internal(genUpdate: UpdatePacket, uiContext: UIContextType, interactive: boolean) {
+        if (genUpdate.type == "pty") {
+            const ptyMsg = genUpdate.data as PtyDataUpdateType;
             if (isBlank(ptyMsg.remoteid)) {
                 // regular update
                 this.updatePtyData(ptyMsg);
@@ -807,125 +800,138 @@ class Model {
                 const ptyData = base64ToArray(ptyMsg.ptydata64);
                 this.remotesModel.receiveData(ptyMsg.remoteid, ptyMsg.ptypos, ptyData);
             }
-            return;
-        }
-        let showedRemotesModal = false;
-        genUpdate.forEach((update) => {
-            if (update.connect != null) {
-                if (update.connect.screens != null) {
-                    this.screenMap.clear();
-                    this.updateScreens(update.connect.screens);
-                }
-                if (update.connect.sessions != null) {
-                    this.sessionList.clear();
-                    this.updateSessions(update.connect.sessions);
-                }
-                if (update.connect.remotes != null) {
-                    this.remotes.clear();
-                    this.updateRemotes(update.connect.remotes);
-                }
-                if (update.connect.activesessionid != null) {
-                    this.updateActiveSession(update.connect.activesessionid);
-                }
-                if (update.connect.screennumrunningcommands != null) {
-                    this.updateScreenNumRunningCommands(update.connect.screennumrunningcommands);
-                }
-                if (update.connect.screenstatusindicators != null) {
-                    this.updateScreenStatusIndicators(update.connect.screenstatusindicators);
-                }
+        } else if (genUpdate.type == "model") {
+            const modelUpdateItems = genUpdate.data as ModelUpdateItemType[];
 
-                this.sessionListLoaded.set(true);
-                this.remotesLoaded.set(true);
-            } else if (update.screen != null) {
-                this.updateScreens([update.screen]);
-            } else if (update.session != null) {
-                this.updateSessions([update.session]);
-            } else if (update.activesessionid != null) {
-                this.updateActiveSession(update.activesessionid);
-            } else if (update.line != null) {
-                this.addLineCmd(update.line.line, update.line.cmd, interactive);
-            } else if (update.cmd != null) {
-                this.updateCmd(update.cmd);
-            } else if (update.screenlines != null) {
-                this.updateScreenLines(update.screenlines, false);
-            } else if (update.remote != null) {
-                this.updateRemotes([update.remote]);
-                // This code's purpose is to show view remote connection modal when a new connection is added
-                if (!showedRemotesModal && this.remotesModel.recentConnAddedState.get()) {
-                    showedRemotesModal = true;
-                    this.remotesModel.openReadModal(update.remote.remoteid);
-                }
-            } else if (update.mainview != null) {
-                switch (update.mainview.mainview) {
-                    case "session":
-                        this.activeMainView.set("session");
-                        break;
-                    case "history":
-                        if (update.mainview.historyview != null) {
-                            this.historyViewModel.showHistoryView(update.mainview.historyview);
-                        } else {
-                            console.warn("invalid historyview in update:", update.mainview);
-                        }
-                        break;
-                    case "bookmarks":
-                        if (update.mainview.bookmarksview != null) {
-                            this.bookmarksModel.showBookmarksView(
-                                update.mainview.bookmarksview?.bookmarks ?? [],
-                                update.mainview.bookmarksview?.selectedbookmark
-                            );
-                        } else {
-                            console.warn("invalid bookmarksview in update:", update.mainview);
-                        }
-                        break;
-                    case "plugins":
-                        this.pluginsModel.showPluginsView();
-                        break;
-                    default:
-                        console.warn("invalid mainview in update:", update.mainview);
-                }
-            } else if (update.bookmarks != null) {
-                if (update.bookmarks.bookmarks != null) {
-                    this.bookmarksModel.mergeBookmarks(update.bookmarks.bookmarks);
-                }
-            } else if (update.clientdata != null) {
-                this.setClientData(update.clientdata);
-            } else if (update.cmdline != null) {
-                this.inputModel.updateCmdLine(update.cmdline);
-            } else if (update.openaicmdinfochat != null) {
-                this.inputModel.setOpenAICmdInfoChat(update.openaicmdinfochat);
-            } else if (update.screenstatusindicator != null) {
-                this.updateScreenStatusIndicators([update.screenstatusindicator]);
-            } else if (update.screennumrunningcommands != null) {
-                this.updateScreenNumRunningCommands([update.screennumrunningcommands]);
-            } else if (update.userinputrequest != null) {
-                let userInputRequest: UserInputRequest = update.userinputrequest;
-                this.modalsModel.pushModal(appconst.USER_INPUT, userInputRequest);
-            } else if (interactive) {
-                if (update.info != null) {
-                    const info: InfoType = update.info;
-                    this.inputModel.flashInfoMsg(info, info.timeoutms);
-                } else if (update.remoteview != null) {
-                    const rview: RemoteViewType = update.remoteview;
-                    if (rview.remoteedit != null) {
-                        this.remotesModel.openEditModal({ ...rview.remoteedit });
+            let showedRemotesModal = false;
+            const [oldActiveSessionId, oldActiveScreenId] = this.getActiveIds();
+            modelUpdateItems.forEach((update) => {
+                if (update.connect != null) {
+                    if (update.connect.screens != null) {
+                        this.screenMap.clear();
+                        this.updateScreens(update.connect.screens);
                     }
-                } else if (update.alertmessage != null) {
-                    const alertMessage: AlertMessageType = update.alertmessage;
-                    this.showAlert(alertMessage);
-                } else if (update.history != null) {
-                    if (
-                        uiContext.sessionid == update.history.sessionid &&
-                        uiContext.screenid == update.history.screenid
-                    ) {
-                        this.inputModel.setHistoryInfo(update.history);
+                    if (update.connect.sessions != null) {
+                        this.sessionList.clear();
+                        this.updateSessions(update.connect.sessions);
+                    }
+                    if (update.connect.remotes != null) {
+                        this.remotes.clear();
+                        this.updateRemotes(update.connect.remotes);
+                    }
+                    if (update.connect.activesessionid != null) {
+                        this.updateActiveSession(update.connect.activesessionid);
+                    }
+                    if (update.connect.screennumrunningcommands != null) {
+                        this.updateScreenNumRunningCommands(update.connect.screennumrunningcommands);
+                    }
+                    if (update.connect.screenstatusindicators != null) {
+                        this.updateScreenStatusIndicators(update.connect.screenstatusindicators);
+                    }
+
+                    this.sessionListLoaded.set(true);
+                    this.remotesLoaded.set(true);
+                } else if (update.screen != null) {
+                    this.updateScreens([update.screen]);
+                } else if (update.session != null) {
+                    this.updateSessions([update.session]);
+                } else if (update.activesessionid != null) {
+                    this.updateActiveSession(update.activesessionid);
+                } else if (update.line != null) {
+                    this.addLineCmd(update.line.line, update.line.cmd, interactive);
+                } else if (update.cmd != null) {
+                    this.updateCmd(update.cmd);
+                } else if (update.screenlines != null) {
+                    this.updateScreenLines(update.screenlines, false);
+                } else if (update.remote != null) {
+                    this.updateRemotes([update.remote]);
+                    // This code's purpose is to show view remote connection modal when a new connection is added
+                    if (!showedRemotesModal && this.remotesModel.recentConnAddedState.get()) {
+                        showedRemotesModal = true;
+                        this.remotesModel.openReadModal(update.remote.remoteid);
+                    }
+                } else if (update.mainview != null) {
+                    switch (update.mainview.mainview) {
+                        case "session":
+                            this.activeMainView.set("session");
+                            break;
+                        case "history":
+                            if (update.mainview.historyview != null) {
+                                this.historyViewModel.showHistoryView(update.mainview.historyview);
+                            } else {
+                                console.warn("invalid historyview in update:", update.mainview);
+                            }
+                            break;
+                        case "bookmarks":
+                            if (update.mainview.bookmarksview != null) {
+                                this.bookmarksModel.showBookmarksView(
+                                    update.mainview.bookmarksview?.bookmarks ?? [],
+                                    update.mainview.bookmarksview?.selectedbookmark
+                                );
+                            } else {
+                                console.warn("invalid bookmarksview in update:", update.mainview);
+                            }
+                            break;
+                        case "plugins":
+                            this.pluginsModel.showPluginsView();
+                            break;
+                        default:
+                            console.warn("invalid mainview in update:", update.mainview);
+                    }
+                } else if (update.bookmarks != null) {
+                    if (update.bookmarks.bookmarks != null) {
+                        this.bookmarksModel.mergeBookmarks(update.bookmarks.bookmarks);
+                    }
+                } else if (update.clientdata != null) {
+                    this.setClientData(update.clientdata);
+                } else if (update.cmdline != null) {
+                    this.inputModel.updateCmdLine(update.cmdline);
+                } else if (update.openaicmdinfochat != null) {
+                    this.inputModel.setOpenAICmdInfoChat(update.openaicmdinfochat);
+                } else if (update.screenstatusindicator != null) {
+                    this.updateScreenStatusIndicators([update.screenstatusindicator]);
+                } else if (update.screennumrunningcommands != null) {
+                    this.updateScreenNumRunningCommands([update.screennumrunningcommands]);
+                } else if (update.userinputrequest != null) {
+                    const userInputRequest: UserInputRequest = update.userinputrequest;
+                    this.modalsModel.pushModal(appconst.USER_INPUT, userInputRequest);
+                } else if (interactive) {
+                    if (update.info != null) {
+                        const info: InfoType = update.info;
+                        this.inputModel.flashInfoMsg(info, info.timeoutms);
+                    } else if (update.remoteview != null) {
+                        const rview: RemoteViewType = update.remoteview;
+                        if (rview.remoteedit != null) {
+                            this.remotesModel.openEditModal({ ...rview.remoteedit });
+                        }
+                    } else if (update.alertmessage != null) {
+                        const alertMessage: AlertMessageType = update.alertmessage;
+                        this.showAlert(alertMessage);
+                    } else if (update.history != null) {
+                        if (
+                            uiContext.sessionid == update.history.sessionid &&
+                            uiContext.screenid == update.history.screenid
+                        ) {
+                            this.inputModel.setHistoryInfo(update.history);
+                        }
+                    } else if (this.isDev) {
+                        console.log("did not match update", update);
                     }
                 } else if (this.isDev) {
                     console.log("did not match update", update);
                 }
-            } else if (this.isDev) {
-                console.log("did not match update", update);
+            });
+
+            // Check if the active session or screen has changed, and if so, watch the new screen
+            const [newActiveSessionId, newActiveScreenId] = this.getActiveIds();
+            if (oldActiveSessionId != newActiveSessionId || oldActiveScreenId != newActiveScreenId) {
+                this.activeMainView.set("session");
+                this.deactivateScreenLines();
+                this.ws.watchScreen(newActiveSessionId, newActiveScreenId);
             }
-        });
+        } else {
+            console.warn("unknown update", genUpdate);
+        }
     }
 
     updateRemotes(remotes: RemoteType[]): void {
@@ -1064,11 +1070,13 @@ class Model {
         this.handleCmdRestart(cmd);
     }
 
-    isInfoUpdate(update: UpdateMessage): boolean {
-        if (update == null || "ptydata64" in update) {
+    isInfoUpdate(update: UpdatePacket): boolean {
+        if (update.type == "model") {
+            const modelUpdate = update as ModelUpdatePacket;
+            return modelUpdate.data.some((u) => u.info != null || u.history != null);
+        } else {
             return false;
         }
-        return update.some((u) => u.info != null || u.history != null);
     }
 
     getClientDataLoop(loopNum: number): void {

src/types/custom.d.ts

@@ -338,6 +338,10 @@ declare global {
     };
 
     type ModelUpdateType = {
+        items?: ModelUpdateItemType[];
+    };
+
+    type ModelUpdateItemType = {
         interactive: boolean;
         session?: SessionDataType;
         activesessionid?: string;
@@ -440,7 +444,17 @@ declare global {
         showCut?: boolean;
     };
 
-    type UpdateMessage = PtyDataUpdateType | ModelUpdateType[];
+    type ModelUpdatePacket = {
+        type: "model";
+        data: ModelUpdateItemType[];
+    };
+
+    type PtyDataUpdatePacket = {
+        type: "pty";
+        data: PtyDataUpdateType;
+    };
+
+    type UpdatePacket = ModelUpdatePacket | PtyDataUpdatePacket;
 
     type RendererContext = {
         screenId: string;

waveshell/main-waveshell.go

@@ -14,6 +14,7 @@ import (
 	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/server"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/shexec"
+	"github.com/wavetermdev/waveterm/waveshell/pkg/wlog"
 )
 
 var BuildTime = "0"
@@ -39,6 +40,7 @@ func handleSingle() {
 		sender.Close()
 		sender.WaitForDone()
 	}()
+	wlog.LogConsumer = sender.SendLogPacket
 	initPacket := shexec.MakeInitPacket()
 	sender.SendPacket(initPacket)
 	if len(os.Args) >= 3 && os.Args[2] == "--version" {
@@ -133,11 +135,13 @@ func main() {
 		return
 	} else if firstArg == "--single" || firstArg == "--single-from-server" {
 		base.ProcessType = base.ProcessType_WaveShellSingle
+		wlog.GlobalSubsystem = base.ProcessType_WaveShellSingle
 		base.InitDebugLog("single")
 		handleSingle()
 		return
 	} else if firstArg == "--server" {
 		base.ProcessType = base.ProcessType_WaveShellServer
+		wlog.GlobalSubsystem = base.ProcessType_WaveShellServer
 		base.InitDebugLog("server")
 		rtnCode, err := server.RunServer()
 		if err != nil {

waveshell/pkg/base/base.go

@@ -42,8 +42,8 @@ const LogRcFileName = "debug.rcfile"
 const (
 	ProcessType_Unknown         = "unknown"
 	ProcessType_WaveSrv         = "wavesrv"
-	ProcessType_WaveShellSingle = "waveshell-single"
-	ProcessType_WaveShellServer = "waveshell-server"
+	ProcessType_WaveShellSingle = "wsh-1"
+	ProcessType_WaveShellServer = "wsh-s"
 )
 
 // keys are sessionids (also the key RcFilesDirBaseName)

waveshell/pkg/packet/packet.go

@@ -14,9 +14,9 @@ import (
 	"os"
 	"reflect"
 	"sync"
-	"time"
 
 	"github.com/wavetermdev/waveterm/waveshell/pkg/base"
+	"github.com/wavetermdev/waveterm/waveshell/pkg/wlog"
 )
 
 // single          : <init, >run, >cmddata, >cmddone, <cmdstart, <>data, <>dataack, <cmddone
@@ -556,11 +556,8 @@ func MakeRawPacket(val string) *RawPacketType {
 }
 
 type LogPacketType struct {
-	Type     string `json:"type"`
-	Ts       int64  `json:"ts"`                 // log timestamp
-	ReqId    string `json:"reqid,omitempty"`    // if this log line is related to an rpc request
-	ProcInfo string `json:"procinfo,omitempty"` // server/single
-	LogLine  string `json:"logline"`            // the logline data
+	Type  string        `json:"type"`
+	Entry wlog.LogEntry `json:"entry"`
 }
 
 func (*LogPacketType) GetType() string {
@@ -571,8 +568,8 @@ func (p *LogPacketType) String() string {
 	return "log"
 }
 
-func MakeLogPacket() *LogPacketType {
-	return &LogPacketType{Type: LogPacketStr, Ts: time.Now().UnixMilli()}
+func MakeLogPacket(entry wlog.LogEntry) *LogPacketType {
+	return &LogPacketType{Type: LogPacketStr, Entry: entry}
 }
 
 type ShellStatePacketType struct {
@@ -975,6 +972,11 @@ type CommandPacketType interface {
 	GetCK() base.CommandKey
 }
 
+type ModelUpdatePacketType struct {
+	Type    string `json:"type"`
+	Updates []any  `json:"updates"`
+}
+
 func AsExtType(pk PacketType) string {
 	if rpcPacket, ok := pk.(RpcPacketType); ok {
 		return fmt.Sprintf("%s[%s]", rpcPacket.GetType(), rpcPacket.GetReqId())
@@ -1103,6 +1105,10 @@ func MakePacketSender(output io.Writer, errHandler func(*PacketSender, PacketTyp
 	return sender
 }
 
+func (sender *PacketSender) SendLogPacket(entry wlog.LogEntry) {
+	sender.SendPacket(MakeLogPacket(entry))
+}
+
 func (sender *PacketSender) goHandleError(pk PacketType, err error) {
 	sender.Lock.Lock()
 	defer sender.Lock.Unlock()

waveshell/pkg/packet/parser.go

@@ -10,6 +10,8 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+
+	"github.com/wavetermdev/waveterm/waveshell/pkg/wlog"
 )
 
 type PacketParser struct {
@@ -243,6 +245,11 @@ func MakePacketParser(input io.Reader, opts *PacketParserOpts) *PacketParser {
 			if pk.GetType() == PingPacketStr {
 				continue
 			}
+			if pk.GetType() == LogPacketStr {
+				logPk := pk.(*LogPacketType)
+				wlog.LogLogEntry(logPk.Entry)
+				continue
+			}
 			if parser.RpcHandler {
 				sent := parser.trySendRpcResponse(pk)
 				if sent {

waveshell/pkg/server/server.go

@@ -23,6 +23,7 @@ import (
 	"github.com/wavetermdev/waveterm/waveshell/pkg/shellapi"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/shexec"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/utilfn"
+	"github.com/wavetermdev/waveterm/waveshell/pkg/wlog"
 )
 
 const MaxFileDataPacketSize = 16 * 1024
@@ -741,6 +742,13 @@ func RunServer() (int, error) {
 		WriteErrorChOnce:    &sync.Once{},
 		WriteFileContextMap: make(map[string]*WriteFileContext),
 	}
+	if debug {
+		packet.GlobalDebug = true
+	}
+	server.MainInput = packet.MakePacketParser(os.Stdin, nil)
+	server.Sender = packet.MakePacketSender(os.Stdout, server.packetSenderErrorHandler)
+	defer server.Close()
+	wlog.LogConsumer = server.Sender.SendLogPacket
 	go func() {
 		for {
 			if server.checkDone() {
@@ -750,12 +758,6 @@ func RunServer() (int, error) {
 			server.cleanWriteFileContexts()
 		}
 	}()
-	if debug {
-		packet.GlobalDebug = true
-	}
-	server.MainInput = packet.MakePacketParser(os.Stdin, nil)
-	server.Sender = packet.MakePacketSender(os.Stdout, server.packetSenderErrorHandler)
-	defer server.Close()
 	var err error
 	initPacket, err := shexec.MakeServerInitPacket()
 	if err != nil {

waveshell/pkg/shexec/shexec.go

@@ -1091,7 +1091,6 @@ func (cmd *ShExecType) DetachedWait(startPacket *packet.CmdStartPacketType) {
 	cmd.DetachedOutput.SendPacket(donePacket)
 	<-ptyCopyDone
 	cmd.Close()
-	return
 }
 
 func RunCommandDetached(pk *packet.RunPacketType, sender *packet.PacketSender) (*ShExecType, *packet.CmdStartPacketType, error) {

waveshell/pkg/wlog/wlog.go

@@ -0,0 +1,77 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+// implements distributed logging for waveshell processes
+package wlog
+
+import (
+	"fmt"
+	"log"
+)
+
+// wlog will send logs back to the controlling wavesrv process
+// note that these logs end up on your local machine where the main Wave Terminal process is running
+// wlog has no ability to send logs to a cloud service or Command Line Inc servers
+
+// this code is written a bit strange (with globals getting set from other packages)
+// because we want no dependencies so any package (including base) can use wlog
+
+// this should match base.ProcessType (set by main)
+var GlobalSubsystem string
+
+// if not set, Logf is a no-op.  will be set by main to hook up to
+// the main packet.PacketSender
+var LogConsumer func(LogEntry)
+
+type LogEntry struct {
+	LogLine   string `json:"logline"`
+	ReqId     string `json:"reqid"`
+	SubSystem string `json:"subsystem"`
+}
+
+func LogLogEntry(entry LogEntry) {
+	if LogConsumer == nil {
+		return
+	}
+	LogConsumer(entry)
+}
+
+// log with a request id (if related to an rpc request)
+func LogfRpc(reqId string, format string, args ...interface{}) {
+	if LogConsumer == nil {
+		return
+	}
+	logEntry := LogEntry{
+		LogLine:   fmt.Sprintf(format, args...),
+		ReqId:     reqId,
+		SubSystem: GlobalSubsystem,
+	}
+	LogConsumer(logEntry)
+}
+
+func LogfSS(subsystem string, format string, args ...interface{}) {
+	if LogConsumer == nil {
+		return
+	}
+	logEntry := LogEntry{
+		LogLine:   fmt.Sprintf(format, args...),
+		ReqId:     "",
+		SubSystem: subsystem,
+	}
+	LogConsumer(logEntry)
+}
+
+func Logf(format string, args ...interface{}) {
+	LogfSS(GlobalSubsystem, format, args...)
+}
+
+func LogWithLogger(entry LogEntry) {
+	if entry.SubSystem == "" {
+		entry.SubSystem = "unknown"
+	}
+	if entry.ReqId != "" {
+		log.Printf("[%s] reqid=%s %s", entry.SubSystem, entry.ReqId, entry.LogLine)
+	} else {
+		log.Printf("[%s] %s", entry.SubSystem, entry.LogLine)
+	}
+}

wavesrv/cmd/main-server.go

@@ -33,6 +33,7 @@ import (
 	"github.com/wavetermdev/waveterm/waveshell/pkg/base"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/server"
+	"github.com/wavetermdev/waveterm/waveshell/pkg/wlog"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/cmdrunner"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/pcloud"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/releasechecker"
@@ -804,6 +805,8 @@ func doShutdown(reason string) {
 func main() {
 	scbase.BuildTime = BuildTime
 	base.ProcessType = base.ProcessType_WaveSrv
+	wlog.GlobalSubsystem = base.ProcessType_WaveSrv
+	wlog.LogConsumer = wlog.LogWithLogger
 
 	if len(os.Args) >= 2 && os.Args[1] == "--test" {
 		log.Printf("running test fn\n")

wavesrv/pkg/releasechecker/releasechecker.go

@@ -1,3 +1,6 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
 package releasechecker
 
 import (
@@ -8,6 +11,7 @@ import (
 	"golang.org/x/mod/semver"
 
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/scbase"
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/scbus"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/sstore"
 )
 
@@ -66,9 +70,9 @@ func CheckNewRelease(ctx context.Context, force bool) (ReleaseCheckResult, error
 		return Failure, fmt.Errorf("error getting updated client data: %w", err)
 	}
 
-	update := &sstore.ModelUpdate{}
-	sstore.AddUpdate(update, *clientData)
-	sstore.MainBus.SendUpdate(update)
+	update := scbus.MakeUpdatePacket()
+	update.AddUpdate(clientData)
+	scbus.MainUpdateBus.DoUpdate(update)
 
 	return Success, nil
 }

wavesrv/pkg/remote/remote.go

@@ -34,8 +34,10 @@ import (
 	"github.com/wavetermdev/waveterm/waveshell/pkg/statediff"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/utilfn"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/scbase"
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/scbus"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/sstore"
+
 	"golang.org/x/crypto/ssh"
 	"golang.org/x/mod/semver"
 )
@@ -554,6 +556,9 @@ func (msh *MShellProc) GetShellPref() string {
 	if msh.Remote.ShellPref == sstore.ShellTypePref_Detect {
 		return msh.InitPkShellType
 	}
+	if msh.Remote.ShellPref == "" {
+		return packet.ShellType_bash
+	}
 	return msh.Remote.ShellPref
 }
 
@@ -601,6 +606,9 @@ func (msh *MShellProc) GetRemoteRuntimeState() RemoteRuntimeState {
 			if state.ConnectTimeout < 0 {
 				state.ConnectTimeout = 0
 			}
+			state.CountdownActive = true
+		} else {
+			state.CountdownActive = false
 		}
 	}
 	vars := msh.Remote.StateVars
@@ -678,9 +686,9 @@ func (msh *MShellProc) GetRemoteRuntimeState() RemoteRuntimeState {
 
 func (msh *MShellProc) NotifyRemoteUpdate() {
 	rstate := msh.GetRemoteRuntimeState()
-	update := &sstore.ModelUpdate{}
-	sstore.AddUpdate(update, rstate)
-	sstore.MainBus.SendUpdate(update)
+	update := scbus.MakeUpdatePacket()
+	update.AddUpdate(rstate)
+	scbus.MainUpdateBus.DoUpdate(update)
 }
 
 func GetAllRemoteRuntimeState() []*RemoteRuntimeState {
@@ -940,13 +948,13 @@ func (msh *MShellProc) writeToPtyBuffer_nolock(strFmt string, args ...interface{
 
 func sendRemotePtyUpdate(remoteId string, dataOffset int64, data []byte) {
 	data64 := base64.StdEncoding.EncodeToString(data)
-	update := &sstore.PtyDataUpdate{
+	update := scbus.MakePtyDataUpdate(&scbus.PtyDataUpdate{
 		RemoteId:   remoteId,
 		PtyPos:     dataOffset,
 		PtyData64:  data64,
 		PtyDataLen: int64(len(data)),
-	}
-	sstore.MainBus.SendUpdate(update)
+	})
+	scbus.MainUpdateBus.DoUpdate(update)
 }
 
 func (msh *MShellProc) isWaitingForPassword_nolock() bool {
@@ -1318,23 +1326,22 @@ func (NewLauncher) Launch(msh *MShellProc, interactive bool) {
 	if remoteCopy.ConnectMode != sstore.ConnectModeManual && remoteCopy.SSHOpts.SSHPassword == "" && !interactive {
 		sshOpts.BatchMode = true
 	}
-	makeClientCtx, makeClientCancelFn := context.WithCancel(context.Background())
-	defer makeClientCancelFn()
-	msh.WithLock(func() {
-		msh.Err = nil
-		msh.ErrNoInitPk = false
-		msh.Status = StatusConnecting
-		msh.MakeClientCancelFn = makeClientCancelFn
-		deadlineTime := time.Now().Add(RemoteConnectTimeout)
-		msh.MakeClientDeadline = &deadlineTime
-		go msh.NotifyRemoteUpdate()
-	})
-	go msh.watchClientDeadlineTime()
-	var cmdStr string
 	var cproc *shexec.ClientProc
 	var initPk *packet.InitPacketType
 	if sshOpts.SSHHost == "" && remoteCopy.Local {
-		cmdStr, err = MakeLocalMShellCommandStr(remoteCopy.IsSudo())
+		makeClientCtx, makeClientCancelFn := context.WithCancel(context.Background())
+		defer makeClientCancelFn()
+		msh.WithLock(func() {
+			msh.Err = nil
+			msh.ErrNoInitPk = false
+			msh.Status = StatusConnecting
+			msh.MakeClientCancelFn = makeClientCancelFn
+			deadlineTime := time.Now().Add(RemoteConnectTimeout)
+			msh.MakeClientDeadline = &deadlineTime
+			go msh.NotifyRemoteUpdate()
+		})
+		go msh.watchClientDeadlineTime()
+		cmdStr, err := MakeLocalMShellCommandStr(remoteCopy.IsSudo())
 		if err != nil {
 			msh.WriteToPtyBuffer("*error, cannot find local mshell binary: %v\n", err)
 			return
@@ -1359,6 +1366,13 @@ func (NewLauncher) Launch(msh *MShellProc, interactive bool) {
 		}
 		cproc, initPk, err = shexec.MakeClientProc(makeClientCtx, shexec.CmdWrap{Cmd: ecmd})
 	} else {
+		msh.WithLock(func() {
+			msh.Err = nil
+			msh.ErrNoInitPk = false
+			msh.Status = StatusConnecting
+			msh.MakeClientDeadline = nil
+			go msh.NotifyRemoteUpdate()
+		})
 		var client *ssh.Client
 		client, err = ConnectToClient(remoteCopy.SSHOpts)
 		if err != nil {
@@ -1375,6 +1389,15 @@ func (NewLauncher) Launch(msh *MShellProc, interactive bool) {
 			msh.setErrorStatus(statusErr)
 			return
 		}
+		makeClientCtx, makeClientCancelFn := context.WithCancel(context.Background())
+		defer makeClientCancelFn()
+		msh.WithLock(func() {
+			msh.MakeClientCancelFn = makeClientCancelFn
+			deadlineTime := time.Now().Add(RemoteConnectTimeout)
+			msh.MakeClientDeadline = &deadlineTime
+			go msh.NotifyRemoteUpdate()
+		})
+		go msh.watchClientDeadlineTime()
 		cproc, initPk, err = shexec.MakeClientProc(makeClientCtx, shexec.SessionWrap{Session: session, StartCmd: MakeServerRunOnlyCommandStr()})
 	}
 	// TODO check if initPk.State is not nil
@@ -1998,9 +2021,9 @@ func (msh *MShellProc) notifyHangups_nolock() {
 		if err != nil {
 			continue
 		}
-		update := &sstore.ModelUpdate{}
-		sstore.AddUpdate(update, *cmd)
-		sstore.MainBus.SendScreenUpdate(ck.GetGroupId(), update)
+		update := scbus.MakeUpdatePacket()
+		update.AddUpdate(*cmd)
+		scbus.MainUpdateBus.DoScreenUpdate(ck.GetGroupId(), update)
 		go pushNumRunningCmdsUpdate(&ck, -1)
 	}
 	msh.RunningCmds = make(map[base.CommandKey]RunCmdType)
@@ -2029,7 +2052,7 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
 		// fall-through (nothing to do)
 	}
 	if screen != nil {
-		sstore.AddUpdate(update, *screen)
+		update.AddUpdate(*screen)
 	}
 	rct := msh.GetRunningCmd(donePk.CK)
 	var statePtr *sstore.ShellStatePtr
@@ -2041,7 +2064,7 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
 			// fall-through (nothing to do)
 		}
 		if remoteInst != nil {
-			sstore.AddUpdate(update, sstore.MakeSessionUpdateForRemote(rct.SessionId, remoteInst))
+			update.AddUpdate(sstore.MakeSessionUpdateForRemote(rct.SessionId, remoteInst))
 		}
 		statePtr = &sstore.ShellStatePtr{BaseHash: donePk.FinalState.GetHashVal(false)}
 	} else if donePk.FinalStateDiff != nil && rct != nil {
@@ -2061,7 +2084,7 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
 				// fall-through (nothing to do)
 			}
 			if remoteInst != nil {
-				sstore.AddUpdate(update, sstore.MakeSessionUpdateForRemote(rct.SessionId, remoteInst))
+				update.AddUpdate(sstore.MakeSessionUpdateForRemote(rct.SessionId, remoteInst))
 			}
 			diffHashArr := append(([]string)(nil), donePk.FinalStateDiff.DiffHashArr...)
 			diffHashArr = append(diffHashArr, donePk.FinalStateDiff.GetHashVal(false))
@@ -2075,7 +2098,7 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
 			// fall-through (nothing to do)
 		}
 	}
-	sstore.MainBus.SendUpdate(update)
+	scbus.MainUpdateBus.DoUpdate(update)
 	return
 }
 
@@ -2104,13 +2127,13 @@ func (msh *MShellProc) handleCmdFinalPacket(finalPk *packet.CmdFinalPacketType)
 		log.Printf("error getting cmd(2) in handleCmdFinalPacket (not found)\n")
 		return
 	}
-	update := &sstore.ModelUpdate{}
-	sstore.AddUpdate(update, *rtnCmd)
+	update := scbus.MakeUpdatePacket()
+	update.AddUpdate(*rtnCmd)
 	if screen != nil {
-		sstore.AddUpdate(update, *screen)
+		update.AddUpdate(*screen)
 	}
 	go pushNumRunningCmdsUpdate(&finalPk.CK, -1)
-	sstore.MainBus.SendUpdate(update)
+	scbus.MainUpdateBus.DoUpdate(update)
 }
 
 // TODO notify FE about cmd errors
@@ -2146,7 +2169,7 @@ func (msh *MShellProc) handleDataPacket(dataPk *packet.DataPacketType, dataPosMa
 		}
 		utilfn.IncSyncMap(dataPosMap, dataPk.CK, int64(len(realData)))
 		if update != nil {
-			sstore.MainBus.SendScreenUpdate(dataPk.CK.GetGroupId(), update)
+			scbus.MainUpdateBus.DoScreenUpdate(dataPk.CK.GetGroupId(), update)
 		}
 	}
 	if ack != nil {
@@ -2175,9 +2198,9 @@ func (msh *MShellProc) makeHandleCmdFinalPacketClosure(finalPk *packet.CmdFinalP
 
 func sendScreenUpdates(screens []*sstore.ScreenType) {
 	for _, screen := range screens {
-		update := &sstore.ModelUpdate{}
-		sstore.AddUpdate(update, *screen)
-		sstore.MainBus.SendUpdate(update)
+		update := scbus.MakeUpdatePacket()
+		update.AddUpdate(*screen)
+		scbus.MainUpdateBus.DoUpdate(update)
 	}
 }
 

wavesrv/pkg/remote/sshclient.go

@@ -6,10 +6,11 @@ package remote
 import (
 	"bytes"
 	"context"
+	"crypto/rand"
+	"crypto/rsa"
 	"crypto/x509"
 	"encoding/base64"
 	"fmt"
-	"log"
 	"net"
 	"os"
 	"os/user"
@@ -21,8 +22,9 @@ import (
 
 	"github.com/kevinburke/ssh_config"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/base"
-	"github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket"
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/scbus"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/sstore"
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/userinput"
 	"golang.org/x/crypto/ssh"
 	"golang.org/x/crypto/ssh/knownhosts"
 )
@@ -35,39 +37,93 @@ func (uice UserInputCancelError) Error() string {
 	return uice.Err.Error()
 }
 
-func createPublicKeyAuth(identityFile string, passphrase string) (ssh.Signer, error) {
-	privateKey, err := os.ReadFile(base.ExpandHomeDir(identityFile))
+// This exists to trick the ssh library into continuing to try
+// different public keys even when the current key cannot be
+// properly parsed
+func createDummySigner() ([]ssh.Signer, error) {
+	dummyKey, err := rsa.GenerateKey(rand.Reader, 2048)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read ssh key file. err: %+v", err)
+		return nil, err
 	}
-	signer, err := ssh.ParsePrivateKey(privateKey)
-	if err == nil {
-		return signer, err
-	}
-	if _, ok := err.(*ssh.PassphraseMissingError); !ok {
-		return nil, fmt.Errorf("failed to parse private ssh key. err: %+v", err)
+	dummySigner, err := ssh.NewSignerFromKey(dummyKey)
+	if err != nil {
+		return nil, err
 	}
+	return []ssh.Signer{dummySigner}, nil
 
-	signer, err = ssh.ParsePrivateKeyWithPassphrase(privateKey, []byte(passphrase))
-	if err == nil {
-		return signer, err
+}
+
+// This is a workaround to only process one identity file at a time,
+// even if they have passphrases. It must be combined with retryable
+// authentication to work properly
+//
+// Despite returning an array of signers, we only ever provide one since
+// it allows proper user interaction in between attempts
+//
+// A significant number of errors end up returning dummy values as if
+// they were successes. An error in this function prevents any other
+// keys from being attempted. But if there's an error because of a dummy
+// file, the library can still try again with a new key.
+func createPublicKeyCallback(sshKeywords *SshKeywords, passphrase string) func() ([]ssh.Signer, error) {
+	identityFiles := make([]string, len(sshKeywords.IdentityFile))
+	copy(identityFiles, sshKeywords.IdentityFile)
+	identityFilesPtr := &identityFiles
+
+	return func() ([]ssh.Signer, error) {
+		if len(*identityFilesPtr) == 0 {
+			// skip this key and try with the next
+			return createDummySigner()
+		}
+		identityFile := (*identityFilesPtr)[0]
+		*identityFilesPtr = (*identityFilesPtr)[1:]
+		privateKey, err := os.ReadFile(base.ExpandHomeDir(identityFile))
+		if err != nil {
+			// skip this key and try with the next
+			return createDummySigner()
+		}
+		signer, err := ssh.ParsePrivateKey(privateKey)
+		if err == nil {
+			return []ssh.Signer{signer}, err
+		}
+		if _, ok := err.(*ssh.PassphraseMissingError); !ok {
+			// skip this key and try with the next
+			return createDummySigner()
+		}
+
+		signer, err = ssh.ParsePrivateKeyWithPassphrase(privateKey, []byte(passphrase))
+		if err == nil {
+			return []ssh.Signer{signer}, err
+		}
+		if err != x509.IncorrectPasswordError && err.Error() != "bcrypt_pbkdf: empty password" {
+			// skip this key and try with the next
+			return createDummySigner()
+		}
+
+		// batch mode deactivates user input
+		if sshKeywords.BatchMode {
+			// skip this key and try with the next
+			return createDummySigner()
+		}
+
+		request := &userinput.UserInputRequestType{
+			ResponseType: "text",
+			QueryText:    fmt.Sprintf("Enter passphrase for the SSH key: %s", identityFile),
+			Title:        "Publickey Auth + Passphrase",
+		}
+		ctx, _ := context.WithTimeout(context.Background(), 60*time.Second)
+		response, err := userinput.GetUserInput(ctx, scbus.MainRpcBus, request)
+		if err != nil {
+			// this is an error where we actually do want to stop
+			// trying keys
+			return nil, UserInputCancelError{Err: err}
+		}
+		signer, err = ssh.ParsePrivateKeyWithPassphrase(privateKey, []byte(response.Text))
+		if err != nil {
+			// skip this key and try with the next
+			return createDummySigner()
+		}
+		return []ssh.Signer{signer}, err
 	}
-	if err != x509.IncorrectPasswordError && err.Error() != "bcrypt_pbkdf: empty password" {
-		log.Printf("qwerty: %+v", err)
-		return nil, fmt.Errorf("failed to parse private ssh key. err: %+v", err)
-	}
-	request := &sstore.UserInputRequestType{
-		ResponseType: "text",
-		QueryText:    fmt.Sprintf("Enter passphrase for the SSH key: %s", identityFile),
-		Title:        "Publickey Auth + Passphrase",
-	}
-	ctx, cancelFn := context.WithTimeout(context.Background(), 15*time.Second)
-	defer cancelFn()
-	response, err := sstore.MainBus.GetUserInput(ctx, request)
-	if err != nil {
-		return nil, UserInputCancelError{Err: err}
-	}
-	return ssh.ParsePrivateKeyWithPassphrase(privateKey, []byte(response.Text))
 }
 
 func createDefaultPasswordCallbackPrompt(password string) func() (secret string, err error) {
@@ -83,14 +139,14 @@ func createInteractivePasswordCallbackPrompt() func() (secret string, err error)
 	return func() (secret string, err error) {
 		// limited to 15 seconds for some reason. this should be investigated more
 		// in the future
-		ctx, cancelFn := context.WithTimeout(context.Background(), 15*time.Second)
+		ctx, cancelFn := context.WithTimeout(context.Background(), 60*time.Second)
 		defer cancelFn()
-		request := &sstore.UserInputRequestType{
+		request := &userinput.UserInputRequestType{
 			ResponseType: "text",
 			QueryText:    "Password:",
 			Title:        "Password Authentication",
 		}
-		response, err := sstore.MainBus.GetUserInput(ctx, request)
+		response, err := userinput.GetUserInput(ctx, scbus.MainRpcBus, request)
 		if err != nil {
 			return "", err
 		}
@@ -143,14 +199,14 @@ func createInteractiveKbdInteractiveChallenge() func(name, instruction string, q
 func promptChallengeQuestion(question string, echo bool) (answer string, err error) {
 	// limited to 15 seconds for some reason. this should be investigated more
 	// in the future
-	ctx, cancelFn := context.WithTimeout(context.Background(), 15*time.Second)
+	ctx, cancelFn := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancelFn()
-	request := &sstore.UserInputRequestType{
+	request := &userinput.UserInputRequestType{
 		ResponseType: "text",
 		QueryText:    question,
 		Title:        "Keyboard Interactive Authentication",
 	}
-	response, err := sstore.MainBus.GetUserInput(ctx, request)
+	response, err := userinput.GetUserInput(ctx, scbus.MainRpcBus, request)
 	if err != nil {
 		return "", err
 	}
@@ -178,10 +234,10 @@ func openKnownHostsForEdit(knownHostsFilename string) (*os.File, error) {
 	return os.OpenFile(knownHostsFilename, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0644)
 }
 
-func writeToKnownHosts(knownHostsFile string, newLine string, getUserVerification func() (*scpacket.UserInputResponsePacketType, error)) error {
+func writeToKnownHosts(knownHostsFile string, newLine string, getUserVerification func() (*userinput.UserInputResponsePacketType, error)) error {
 	if getUserVerification == nil {
-		getUserVerification = func() (*scpacket.UserInputResponsePacketType, error) {
-			return &scpacket.UserInputResponsePacketType{
+		getUserVerification = func() (*userinput.UserInputResponsePacketType, error) {
+			return &userinput.UserInputResponsePacketType{
 				Type:    "confirm",
 				Confirm: true,
 			}, nil
@@ -214,7 +270,7 @@ func writeToKnownHosts(knownHostsFile string, newLine string, getUserVerificatio
 	return f.Close()
 }
 
-func createUnknownKeyVerifier(knownHostsFile string, hostname string, remote string, key ssh.PublicKey) func() (*scpacket.UserInputResponsePacketType, error) {
+func createUnknownKeyVerifier(knownHostsFile string, hostname string, remote string, key ssh.PublicKey) func() (*userinput.UserInputResponsePacketType, error) {
 	base64Key := base64.StdEncoding.EncodeToString(key.Marshal())
 	queryText := fmt.Sprintf(
 		"The authenticity of host '%s (%s)' can't be established "+
@@ -224,20 +280,20 @@ func createUnknownKeyVerifier(knownHostsFile string, hostname string, remote str
 			"**Would you like to continue connecting?** If so, the key will be permanently "+
 			"added to the file %s "+
 			"to protect from future man-in-the-middle attacks.", hostname, remote, key.Type(), base64Key, knownHostsFile)
-	request := &sstore.UserInputRequestType{
+	request := &userinput.UserInputRequestType{
 		ResponseType: "confirm",
 		QueryText:    queryText,
 		Markdown:     true,
 		Title:        "Known Hosts Key Missing",
 	}
-	return func() (*scpacket.UserInputResponsePacketType, error) {
+	return func() (*userinput.UserInputResponsePacketType, error) {
 		ctx, cancelFn := context.WithTimeout(context.Background(), 60*time.Second)
 		defer cancelFn()
-		return sstore.MainBus.GetUserInput(ctx, request)
+		return userinput.GetUserInput(ctx, scbus.MainRpcBus, request)
 	}
 }
 
-func createMissingKnownHostsVerifier(knownHostsFile string, hostname string, remote string, key ssh.PublicKey) func() (*scpacket.UserInputResponsePacketType, error) {
+func createMissingKnownHostsVerifier(knownHostsFile string, hostname string, remote string, key ssh.PublicKey) func() (*userinput.UserInputResponsePacketType, error) {
 	base64Key := base64.StdEncoding.EncodeToString(key.Marshal())
 	queryText := fmt.Sprintf(
 		"The authenticity of host '%s (%s)' can't be established "+
@@ -248,16 +304,16 @@ func createMissingKnownHostsVerifier(knownHostsFile string, hostname string, rem
 			"- %s will be created  \n"+
 			"- the key will be added to %s\n\n"+
 			"This will protect from future man-in-the-middle attacks.", hostname, remote, key.Type(), base64Key, knownHostsFile, knownHostsFile)
-	request := &sstore.UserInputRequestType{
+	request := &userinput.UserInputRequestType{
 		ResponseType: "confirm",
 		QueryText:    queryText,
 		Markdown:     true,
 		Title:        "Known Hosts File Missing",
 	}
-	return func() (*scpacket.UserInputResponsePacketType, error) {
+	return func() (*userinput.UserInputResponsePacketType, error) {
 		ctx, cancelFn := context.WithTimeout(context.Background(), 60*time.Second)
 		defer cancelFn()
-		return sstore.MainBus.GetUserInput(ctx, request)
+		return userinput.GetUserInput(ctx, scbus.MainRpcBus, request)
 	}
 }
 
@@ -388,13 +444,13 @@ func createHostKeyCallback(opts *sstore.SSHOpts) (ssh.HostKeyCallback, error) {
 				"%s\n\n"+
 				"**Offending Keys**  \n"+
 				"%s", key.Type(), correctKeyFingerprint, strings.Join(bulletListKnownHosts, "  \n"), strings.Join(offendingKeysFmt, "  \n"))
-			update := &sstore.ModelUpdate{}
-			sstore.AddUpdate(update, sstore.AlertMessageType{
+			update := scbus.MakeUpdatePacket()
+			update.AddUpdate(sstore.AlertMessageType{
 				Markdown: true,
 				Title:    "Known Hosts Key Changed",
 				Message:  alertText,
 			})
-			sstore.MainBus.SendUpdate(update)
+			scbus.MainUpdateBus.DoUpdate(update)
 			return fmt.Errorf("remote host identification has changed")
 		}
 
@@ -410,61 +466,180 @@ func createHostKeyCallback(opts *sstore.SSHOpts) (ssh.HostKeyCallback, error) {
 }
 
 func ConnectToClient(opts *sstore.SSHOpts) (*ssh.Client, error) {
-	ssh_config.ReloadConfigs()
-	configIdentity, _ := ssh_config.GetStrict(opts.SSHHost, "IdentityFile")
-	var identityFile string
-	if opts.SSHIdentity != "" {
-		identityFile = opts.SSHIdentity
+	sshConfigKeywords, err := findSshConfigKeywords(opts.SSHHost)
+	if err != nil {
+		return nil, err
+	}
+
+	sshKeywords, err := combineSshKeywords(opts, sshConfigKeywords)
+	if err != nil {
+		return nil, err
+	}
+
+	publicKeyCallback := ssh.PublicKeysCallback(createPublicKeyCallback(sshKeywords, opts.SSHPassword))
+	keyboardInteractive := ssh.KeyboardInteractive(createCombinedKbdInteractiveChallenge(opts.SSHPassword))
+	passwordCallback := ssh.PasswordCallback(createCombinedPasswordCallbackPrompt(opts.SSHPassword))
+
+	// batch mode turns off interactive input. this means the number of
+	// attemtps must drop to 1 with this setup
+	var attemptsAllowed int
+	if sshKeywords.BatchMode {
+		attemptsAllowed = 1
 	} else {
-		identityFile = configIdentity
+		attemptsAllowed = 2
+	}
+
+	// exclude gssapi-with-mic and hostbased until implemented
+	authMethodMap := map[string]ssh.AuthMethod{
+		"publickey":            ssh.RetryableAuthMethod(publicKeyCallback, len(sshKeywords.IdentityFile)),
+		"keyboard-interactive": ssh.RetryableAuthMethod(keyboardInteractive, attemptsAllowed),
+		"password":             ssh.RetryableAuthMethod(passwordCallback, attemptsAllowed),
+	}
+
+	authMethodActiveMap := map[string]bool{
+		"publickey":            sshKeywords.PubkeyAuthentication,
+		"keyboard-interactive": sshKeywords.KbdInteractiveAuthentication,
+		"password":             sshKeywords.PasswordAuthentication,
+	}
+
+	var authMethods []ssh.AuthMethod
+	for _, authMethodName := range sshKeywords.PreferredAuthentications {
+		authMethodActive, ok := authMethodActiveMap[authMethodName]
+		if !ok || !authMethodActive {
+			continue
+		}
+		authMethod, ok := authMethodMap[authMethodName]
+		if !ok {
+			continue
+		}
+		authMethods = append(authMethods, authMethod)
 	}
 
 	hostKeyCallback, err := createHostKeyCallback(opts)
 	if err != nil {
 		return nil, err
 	}
-	var authMethods []ssh.AuthMethod
-	publicKeySigner, err := createPublicKeyAuth(identityFile, opts.SSHPassword)
-	if err == nil {
-		authMethods = append(authMethods, ssh.PublicKeys(publicKeySigner))
-	}
-	authMethods = append(authMethods, ssh.RetryableAuthMethod(ssh.KeyboardInteractive(createCombinedKbdInteractiveChallenge(opts.SSHPassword)), 2))
-	authMethods = append(authMethods, ssh.RetryableAuthMethod(ssh.PasswordCallback(createCombinedPasswordCallbackPrompt(opts.SSHPassword)), 2))
 
-	configUser, _ := ssh_config.GetStrict(opts.SSHHost, "User")
-	configHostName, _ := ssh_config.GetStrict(opts.SSHHost, "HostName")
-	configPort, _ := ssh_config.GetStrict(opts.SSHHost, "Port")
-	var username string
+	clientConfig := &ssh.ClientConfig{
+		User:            sshKeywords.User,
+		Auth:            authMethods,
+		HostKeyCallback: hostKeyCallback,
+	}
+	networkAddr := sshKeywords.HostName + ":" + sshKeywords.Port
+	return ssh.Dial("tcp", networkAddr, clientConfig)
+}
+
+type SshKeywords struct {
+	User                         string
+	HostName                     string
+	Port                         string
+	IdentityFile                 []string
+	BatchMode                    bool
+	PubkeyAuthentication         bool
+	PasswordAuthentication       bool
+	KbdInteractiveAuthentication bool
+	PreferredAuthentications     []string
+}
+
+func combineSshKeywords(opts *sstore.SSHOpts, configKeywords *SshKeywords) (*SshKeywords, error) {
+	sshKeywords := &SshKeywords{}
+
 	if opts.SSHUser != "" {
-		username = opts.SSHUser
-	} else if configUser != "" {
-		username = configUser
+		sshKeywords.User = opts.SSHUser
+	} else if configKeywords.User != "" {
+		sshKeywords.User = configKeywords.User
 	} else {
 		user, err := user.Current()
 		if err != nil {
 			return nil, fmt.Errorf("failed to get user for ssh: %+v", err)
 		}
-		username = user.Username
+		sshKeywords.User = user.Username
 	}
-	var hostName string
-	if configHostName != "" {
-		hostName = configHostName
+
+	// we have to check the host value because of the weird way
+	// we store the pattern as the hostname for imported remotes
+	if configKeywords.HostName != "" {
+		sshKeywords.HostName = configKeywords.HostName
 	} else {
-		hostName = opts.SSHHost
+		sshKeywords.HostName = opts.SSHHost
 	}
-	clientConfig := &ssh.ClientConfig{
-		User:            username,
-		Auth:            authMethods,
-		HostKeyCallback: hostKeyCallback,
-	}
-	var port string
+
 	if opts.SSHPort != 0 && opts.SSHPort != 22 {
-		port = strconv.Itoa(opts.SSHPort)
-	} else if configPort != "" && configPort != "22" {
-		port = configPort
+		sshKeywords.Port = strconv.Itoa(opts.SSHPort)
+	} else if configKeywords.Port != "" && configKeywords.Port != "22" {
+		sshKeywords.Port = configKeywords.Port
 	} else {
-		port = "22"
+		sshKeywords.Port = "22"
 	}
-	networkAddr := hostName + ":" + port
-	return ssh.Dial("tcp", networkAddr, clientConfig)
+
+	sshKeywords.IdentityFile = []string{opts.SSHIdentity}
+	sshKeywords.IdentityFile = append(sshKeywords.IdentityFile, configKeywords.IdentityFile...)
+
+	// these are not officially supported in the waveterm frontend but can be configured
+	// in ssh config files
+	sshKeywords.BatchMode = configKeywords.BatchMode
+	sshKeywords.PubkeyAuthentication = configKeywords.PubkeyAuthentication
+	sshKeywords.PasswordAuthentication = configKeywords.PasswordAuthentication
+	sshKeywords.KbdInteractiveAuthentication = configKeywords.KbdInteractiveAuthentication
+	sshKeywords.PreferredAuthentications = configKeywords.PreferredAuthentications
+
+	return sshKeywords, nil
+}
+
+// note that a `var == "yes"` will default to false
+// but `var != "no"` will default to true
+// when given unexpected strings
+func findSshConfigKeywords(hostPattern string) (*SshKeywords, error) {
+	ssh_config.ReloadConfigs()
+	sshKeywords := &SshKeywords{}
+	var err error
+
+	sshKeywords.User, err = ssh_config.GetStrict(hostPattern, "User")
+	if err != nil {
+		return nil, err
+	}
+
+	sshKeywords.HostName, err = ssh_config.GetStrict(hostPattern, "HostName")
+	if err != nil {
+		return nil, err
+	}
+
+	sshKeywords.Port, err = ssh_config.GetStrict(hostPattern, "Port")
+	if err != nil {
+		return nil, err
+	}
+
+	sshKeywords.IdentityFile = ssh_config.GetAll(hostPattern, "IdentityFile")
+
+	batchModeRaw, err := ssh_config.GetStrict(hostPattern, "BatchMode")
+	if err != nil {
+		return nil, err
+	}
+	sshKeywords.BatchMode = (strings.ToLower(batchModeRaw) == "yes")
+
+	// we currently do not support host-bound or unbound but will use yes when they are selected
+	pubkeyAuthenticationRaw, err := ssh_config.GetStrict(hostPattern, "PubkeyAuthentication")
+	if err != nil {
+		return nil, err
+	}
+	sshKeywords.PubkeyAuthentication = (strings.ToLower(pubkeyAuthenticationRaw) != "no")
+
+	passwordAuthenticationRaw, err := ssh_config.GetStrict(hostPattern, "PasswordAuthentication")
+	if err != nil {
+		return nil, err
+	}
+	sshKeywords.PasswordAuthentication = (strings.ToLower(passwordAuthenticationRaw) != "no")
+
+	kbdInteractiveAuthenticationRaw, err := ssh_config.GetStrict(hostPattern, "KbdInteractiveAuthentication")
+	if err != nil {
+		return nil, err
+	}
+	sshKeywords.KbdInteractiveAuthentication = (strings.ToLower(kbdInteractiveAuthenticationRaw) != "no")
+
+	// these are parsed as a single string and must be separated
+	// these are case sensitive in openssh so they are here too
+	preferredAuthenticationsRaw, err := ssh_config.GetStrict(hostPattern, "PreferredAuthentications")
+	sshKeywords.PreferredAuthentications = strings.Split(preferredAuthenticationsRaw, ",")
+
+	return sshKeywords, nil
 }

wavesrv/pkg/scbus/modelupdate.go

@@ -0,0 +1,132 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package scbus
+
+import (
+	"encoding/json"
+	"reflect"
+
+	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
+)
+
+const ModelUpdateStr = "model"
+
+// A channel for sending model updates to the client
+type ModelUpdateChannel[J any] struct {
+	ScreenId string
+	ClientId string
+	ch       chan J
+}
+
+func (uch *ModelUpdateChannel[J]) GetChannel() chan J {
+	return uch.ch
+}
+
+func (uch *ModelUpdateChannel[J]) SetChannel(ch chan J) {
+	uch.ch = ch
+}
+
+// Match the screenId to the channel
+func (sch *ModelUpdateChannel[J]) Match(screenId string) bool {
+	if screenId == "" {
+		return true
+	}
+	return screenId == sch.ScreenId
+}
+
+// An interface for all model updates
+type ModelUpdateItem interface {
+	// The key to use when marshalling to JSON and interpreting in the client
+	GetType() string
+}
+
+// An inner data type for the ModelUpdatePacketType. Stores a collection of model updates to be sent to the client.
+type ModelUpdate []ModelUpdateItem
+
+func (mu *ModelUpdate) IsEmpty() bool {
+	if mu == nil {
+		return true
+	}
+	muArr := []ModelUpdateItem(*mu)
+	return len(muArr) == 0
+}
+
+func (mu *ModelUpdate) MarshalJSON() ([]byte, error) {
+	rtn := make([]map[string]any, 0)
+	for _, u := range *mu {
+		m := make(map[string]any)
+		m[(u).GetType()] = u
+		rtn = append(rtn, m)
+	}
+	return json.Marshal(rtn)
+}
+
+// An UpdatePacket for sending model updates to the client
+type ModelUpdatePacketType struct {
+	Type string       `json:"type"`
+	Data *ModelUpdate `json:"data"`
+}
+
+func (*ModelUpdatePacketType) GetType() string {
+	return ModelUpdateStr
+}
+
+func (upk *ModelUpdatePacketType) IsEmpty() bool {
+	if upk == nil {
+		return true
+	}
+	return upk.Data.IsEmpty()
+}
+
+// Clean the ClientData in an update, if present
+func (upk *ModelUpdatePacketType) Clean() {
+	if upk.IsEmpty() {
+		return
+	}
+	for _, item := range *(upk.Data) {
+		if i, ok := (item).(CleanableUpdateItem); ok {
+			i.Clean()
+		}
+	}
+}
+
+// Add a collection of model updates to the update
+func (upk *ModelUpdatePacketType) AddUpdate(items ...ModelUpdateItem) {
+	*(upk.Data) = append(*(upk.Data), items...)
+}
+
+// Create a new model update packet
+func MakeUpdatePacket() *ModelUpdatePacketType {
+	return &ModelUpdatePacketType{
+		Type: ModelUpdateStr,
+		Data: &ModelUpdate{},
+	}
+}
+
+// Returns the items in the update that are of type I
+func GetUpdateItems[I ModelUpdateItem](upk *ModelUpdatePacketType) []*I {
+	if upk.IsEmpty() {
+		return nil
+	}
+	ret := make([]*I, 0)
+	for _, item := range *(upk.Data) {
+		if i, ok := (item).(I); ok {
+			ret = append(ret, &i)
+		}
+	}
+	return ret
+}
+
+// An interface for model updates that can be cleaned
+type CleanableUpdateItem interface {
+	Clean()
+}
+
+func init() {
+	// Register the model update packet type
+	packet.RegisterPacketType(ModelUpdateStr, reflect.TypeOf(ModelUpdatePacketType{}))
+
+	// Enforce the UpdatePacket interface
+	var _ UpdatePacket = (*ModelUpdatePacketType)(nil)
+}

wavesrv/pkg/scbus/ptydataupdate.go

@@ -0,0 +1,53 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package scbus
+
+import (
+	"reflect"
+
+	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
+)
+
+const PtyDataUpdateStr = "pty"
+
+// The inner data type for the PtyDataUpdatePacketType. Stores the pty data to be sent to the client.
+type PtyDataUpdate struct {
+	ScreenId   string `json:"screenid,omitempty"`
+	LineId     string `json:"lineid,omitempty"`
+	RemoteId   string `json:"remoteid,omitempty"`
+	PtyPos     int64  `json:"ptypos"`
+	PtyData64  string `json:"ptydata64"`
+	PtyDataLen int64  `json:"ptydatalen"`
+}
+
+// An UpdatePacket for sending pty data to the client
+type PtyDataUpdatePacketType struct {
+	Type string         `json:"type"`
+	Data *PtyDataUpdate `json:"data"`
+}
+
+func (*PtyDataUpdatePacketType) GetType() string {
+	return PtyDataUpdateStr
+}
+
+func (pdu *PtyDataUpdatePacketType) Clean() {
+	// This is a no-op for PtyDataUpdatePacketType, but it is required to satisfy the UpdatePacket interface
+}
+
+func (pdu *PtyDataUpdatePacketType) IsEmpty() bool {
+	return pdu == nil || pdu.Data == nil || pdu.Data.PtyDataLen == 0
+}
+
+// Create a new PtyDataUpdatePacketType
+func MakePtyDataUpdate(update *PtyDataUpdate) *PtyDataUpdatePacketType {
+	return &PtyDataUpdatePacketType{Type: PtyDataUpdateStr, Data: update}
+}
+
+func init() {
+	// Register the PtyDataUpdatePacketType with the packet package
+	packet.RegisterPacketType(PtyDataUpdateStr, reflect.TypeOf(PtyDataUpdatePacketType{}))
+
+	// Enforce the UpdatePacket interface
+	var _ UpdatePacket = (*PtyDataUpdatePacketType)(nil)
+}

wavesrv/pkg/scbus/scbus.go

@@ -0,0 +1,248 @@
+// Copyright 2024, Command Line Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+// Defines interfaces for creating communciation channels between server and clients
+package scbus
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"reflect"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
+)
+
+var MainUpdateBus *UpdateBus = MakeUpdateBus()
+var MainRpcBus *RpcBus = MakeRpcBus()
+
+// The default channel size
+const ChSize = 100
+
+type Channel[I packet.PacketType] interface {
+	GetChannel() chan I
+	SetChannel(chan I)
+	Match(string) bool
+}
+
+// A concurrent bus for registering and managing channels
+type Bus[I packet.PacketType] struct {
+	Lock     *sync.Mutex
+	Channels map[string]Channel[I]
+}
+
+// Opens new channel and registers it with the bus. If a channel exists, it is closed and replaced.
+func (bus *Bus[I]) RegisterChannel(key string, channelEntry Channel[I]) chan I {
+	bus.Lock.Lock()
+	defer bus.Lock.Unlock()
+	uch, found := bus.Channels[key]
+	ch := make(chan I, ChSize)
+	channelEntry.SetChannel(ch)
+	if found {
+		close(uch.GetChannel())
+	}
+	bus.Channels[key] = channelEntry
+	return channelEntry.GetChannel()
+}
+
+// Closes the channel matching the provided key and removes it from the bus
+func (bus *Bus[I]) UnregisterChannel(key string) {
+	bus.Lock.Lock()
+	defer bus.Lock.Unlock()
+	uch, found := bus.Channels[key]
+	if found {
+		close(uch.GetChannel())
+		delete(bus.Channels, key)
+	}
+}
+
+// An interface for updates to be sent over an UpdateChannel
+type UpdatePacket interface {
+	// The key to use when marshalling to JSON and interpreting in the client
+	GetType() string
+	Clean()
+	IsEmpty() bool
+}
+
+// Enforce that UpdatePacket is a packet.PacketType
+var _ packet.PacketType = (UpdatePacket)(nil)
+
+// A channel for sending model updates to the client
+type UpdateChannel struct {
+	ScreenId string
+	ch       chan UpdatePacket
+}
+
+func (uch *UpdateChannel) GetChannel() chan UpdatePacket {
+	return uch.ch
+}
+
+func (uch *UpdateChannel) SetChannel(ch chan UpdatePacket) {
+	uch.ch = ch
+}
+
+// Match the screenId to the channel
+func (sch *UpdateChannel) Match(screenId string) bool {
+	if screenId == "" {
+		return true
+	}
+	return screenId == sch.ScreenId
+}
+
+// Enforce that UpdateChannel is a Channel
+var _ Channel[UpdatePacket] = (*UpdateChannel)(nil)
+
+// A collection of channels that can transmit updates
+type UpdateBus struct {
+	Bus[UpdatePacket]
+}
+
+func (bus *UpdateBus) GetLock() *sync.Mutex {
+	return bus.Lock
+}
+
+// Create a new UpdateBus
+func MakeUpdateBus() *UpdateBus {
+	return &UpdateBus{
+		Bus[UpdatePacket]{
+			Lock:     &sync.Mutex{},
+			Channels: make(map[string]Channel[UpdatePacket]),
+		},
+	}
+}
+
+// Send an update to all channels in the collection
+func (bus *UpdateBus) DoUpdate(update UpdatePacket) {
+	if update.IsEmpty() {
+		return
+	}
+	update.Clean()
+	bus.Lock.Lock()
+	defer bus.Lock.Unlock()
+	for key, uch := range bus.Channels {
+		select {
+		case uch.GetChannel() <- update:
+
+		default:
+			log.Printf("[error] dropped update on %s updatebus uch key=%s\n", reflect.TypeOf(uch), key)
+		}
+	}
+}
+
+// Send a model update to only clients that are subscribed to the given screenId
+func (bus *UpdateBus) DoScreenUpdate(screenId string, update UpdatePacket) {
+	if update.IsEmpty() {
+		return
+	}
+	update.Clean()
+	bus.Lock.Lock()
+	defer bus.Lock.Unlock()
+	for id, uch := range bus.Channels {
+		if uch.Match(screenId) {
+			select {
+			case uch.GetChannel() <- update:
+
+			default:
+				log.Printf("[error] dropped update on updatebus uch id=%s\n", id)
+			}
+		}
+	}
+}
+
+// An interface for rpc requests
+// This is separate from the RpcPacketType defined in the waveshell/pkg/packet package, as that one is intended for use communicating between wavesrv and waveshell. It is has a different set of required methods.
+type RpcPacket interface {
+	SetReqId(string)
+	SetTimeoutMs(int)
+	GetType() string
+}
+
+// An interface for rpc responses
+// This is separate from the RpcResponsePacketType defined in the waveshell/pkg/packet package, as that one is intended for use communicating between wavesrv and waveshell. It is has a different set of required methods.
+type RpcResponse interface {
+	SetError(string)
+	GetError() string
+	GetType() string
+}
+
+// A collection of channels that can receive rpc responses
+type RpcBus struct {
+	Bus[RpcResponse]
+}
+
+// Create a new RpcBus
+func MakeRpcBus() *RpcBus {
+	return &RpcBus{
+		Bus[RpcResponse]{
+			Lock:     &sync.Mutex{},
+			Channels: make(map[string]Channel[RpcResponse]),
+		},
+	}
+}
+
+// Get the user input channel for the given request id
+func (bus *RpcBus) GetRpcChannel(id string) (chan RpcResponse, bool) {
+	bus.Lock.Lock()
+	defer bus.Lock.Unlock()
+
+	if ch, ok := bus.Channels[id]; ok {
+		return ch.GetChannel(), ok
+	}
+	return nil, false
+}
+
+// Implements the Channel interface to allow receiving rpc responses
+type RpcChannel struct {
+	ch chan RpcResponse
+}
+
+func (ch *RpcChannel) GetChannel() chan RpcResponse {
+	return ch.ch
+}
+
+func (ch *RpcChannel) SetChannel(newCh chan RpcResponse) {
+	ch.ch = newCh
+}
+
+// This is a no-op, only used to satisfy the Channel interface
+func (ch *RpcChannel) Match(string) bool {
+	return true
+}
+
+// Enforce that RpcChannel is a Channel
+var _ Channel[RpcResponse] = (*RpcChannel)(nil)
+
+// Send a user input request to the frontend and wait for a response
+func (bus *RpcBus) DoRpc(ctx context.Context, pk RpcPacket) (RpcResponse, error) {
+	id := uuid.New().String()
+	ch := bus.RegisterChannel(id, &RpcChannel{})
+	pk.SetReqId(id)
+	defer bus.UnregisterChannel(id)
+
+	deadline, _ := ctx.Deadline()
+	pk.SetTimeoutMs(int(time.Until(deadline).Milliseconds()) - 500)
+
+	// Send the request to the frontend
+	mu := MakeUpdatePacket()
+	mu.AddUpdate(pk)
+	MainUpdateBus.DoUpdate(mu)
+
+	var response RpcResponse
+	var err error
+	// prepare to receive response
+	select {
+	case resp := <-ch:
+		response = resp
+	case <-ctx.Done():
+		return nil, fmt.Errorf("timed out waiting for rpc response")
+	}
+
+	if response.GetError() != "" {
+		err = fmt.Errorf(response.GetError())
+	}
+
+	return response, err
+}

wavesrv/pkg/scpacket/scpacket.go

@@ -83,7 +83,6 @@ const WatchScreenPacketStr = "watchscreen"
 const FeInputPacketStr = "feinput"
 const RemoteInputPacketStr = "remoteinput"
 const CmdInputTextPacketStr = "cmdinputtext"
-const UserInputResponsePacketStr = "userinputresp"
 
 type FeCommandPacketType struct {
 	Type        string            `json:"type"`
@@ -156,21 +155,16 @@ type CmdInputTextPacketType struct {
 	Text     utilfn.StrWithPos `json:"text"`
 }
 
-type UserInputResponsePacketType struct {
-	Type      string `json:"type"`
-	RequestId string `json:"requestid"`
-	Text      string `json:"text,omitempty"`
-	Confirm   bool   `json:"confirm,omitempty"`
-	ErrorMsg  string `json:"errormsg,omitempty"`
-}
-
 func init() {
 	packet.RegisterPacketType(FeCommandPacketStr, reflect.TypeOf(FeCommandPacketType{}))
 	packet.RegisterPacketType(WatchScreenPacketStr, reflect.TypeOf(WatchScreenPacketType{}))
 	packet.RegisterPacketType(FeInputPacketStr, reflect.TypeOf(FeInputPacketType{}))
 	packet.RegisterPacketType(RemoteInputPacketStr, reflect.TypeOf(RemoteInputPacketType{}))
 	packet.RegisterPacketType(CmdInputTextPacketStr, reflect.TypeOf(CmdInputTextPacketType{}))
-	packet.RegisterPacketType(UserInputResponsePacketStr, reflect.TypeOf(UserInputResponsePacketType{}))
+}
+
+type PacketType interface {
+	GetType() string
 }
 
 func (*CmdInputTextPacketType) GetType() string {
@@ -212,7 +206,3 @@ func MakeRemoteInputPacket() *RemoteInputPacketType {
 func (*RemoteInputPacketType) GetType() string {
 	return RemoteInputPacketStr
 }
-
-func (*UserInputResponsePacketType) GetType() string {
-	return UserInputResponsePacketStr
-}

wavesrv/pkg/scws/scws.go

@@ -15,8 +15,10 @@ import (
 	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/mapqueue"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/remote"
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/scbus"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/sstore"
+	"github.com/wavetermdev/waveterm/wavesrv/pkg/userinput"
 	"github.com/wavetermdev/waveterm/wavesrv/pkg/wsshell"
 )
 
@@ -35,8 +37,8 @@ type WSState struct {
 	ClientId      string
 	ConnectTime   time.Time
 	Shell         *wsshell.WSShell
-	UpdateCh      chan interface{}
-	UpdateQueue   []interface{}
+	UpdateCh      chan scbus.UpdatePacket
+	UpdateQueue   []any
 	Authenticated bool
 	AuthKey       string
 
@@ -71,7 +73,7 @@ func (ws *WSState) GetShell() *wsshell.WSShell {
 	return ws.Shell
 }
 
-func (ws *WSState) WriteUpdate(update interface{}) error {
+func (ws *WSState) WriteUpdate(update any) error {
 	shell := ws.GetShell()
 	if shell == nil {
 		return fmt.Errorf("cannot write update, empty shell")
@@ -103,26 +105,21 @@ func (ws *WSState) WatchScreen(sessionId string, screenId string) {
 	}
 	ws.SessionId = sessionId
 	ws.ScreenId = screenId
-	ws.UpdateCh = sstore.MainBus.RegisterChannel(ws.ClientId, ws.ScreenId)
+	ws.UpdateCh = scbus.MainUpdateBus.RegisterChannel(ws.ClientId, &scbus.UpdateChannel{ScreenId: ws.ScreenId})
+	log.Printf("[ws] watch screen clientid=%s sessionid=%s screenid=%s, updateCh=%v\n", ws.ClientId, sessionId, screenId, ws.UpdateCh)
 	go ws.RunUpdates(ws.UpdateCh)
 }
 
 func (ws *WSState) UnWatchScreen() {
 	ws.Lock.Lock()
 	defer ws.Lock.Unlock()
-	sstore.MainBus.UnregisterChannel(ws.ClientId)
+	scbus.MainUpdateBus.UnregisterChannel(ws.ClientId)
 	ws.SessionId = ""
 	ws.ScreenId = ""
 	log.Printf("[ws] unwatch screen clientid=%s\n", ws.ClientId)
 }
 
-func (ws *WSState) getUpdateCh() chan interface{} {
-	ws.Lock.Lock()
-	defer ws.Lock.Unlock()
-	return ws.UpdateCh
-}
-
-func (ws *WSState) RunUpdates(updateCh chan interface{}) {
+func (ws *WSState) RunUpdates(updateCh chan scbus.UpdatePacket) {
 	if updateCh == nil {
 		panic("invalid nil updateCh passed to RunUpdates")
 	}
@@ -141,7 +138,6 @@ func writeJsonProtected(shell *wsshell.WSShell, update any) {
 			return
 		}
 		log.Printf("[error] in scws RunUpdates WriteJson: %v\n", r)
-		return
 	}()
 	shell.WriteJson(update)
 }
@@ -155,7 +151,6 @@ func (ws *WSState) ReplaceShell(shell *wsshell.WSShell) {
 	}
 	ws.Shell.Conn.Close()
 	ws.Shell = shell
-	return
 }
 
 // returns all state required to display current UI
@@ -170,8 +165,8 @@ func (ws *WSState) handleConnection() error {
 	connectUpdate.Remotes = remotes
 	// restore status indicators
 	connectUpdate.ScreenStatusIndicators, connectUpdate.ScreenNumRunningCommands = sstore.GetCurrentIndicatorState()
-	mu := &sstore.ModelUpdate{}
-	sstore.AddUpdate(mu, *connectUpdate)
+	mu := scbus.MakeUpdatePacket()
+	mu.AddUpdate(*connectUpdate)
 	err = ws.Shell.WriteJson(mu)
 	if err != nil {
 		return err
@@ -282,11 +277,11 @@ func (ws *WSState) processMessage(msgBytes []byte) error {
 		sstore.ScreenMemSetCmdInputText(cmdInputPk.ScreenId, cmdInputPk.Text, cmdInputPk.SeqNum)
 		return nil
 	}
-	if pk.GetType() == scpacket.UserInputResponsePacketStr {
-		userInputRespPk := pk.(*scpacket.UserInputResponsePacketType)
-		uich, ok := sstore.MainBus.GetUserInputChannel(userInputRespPk.RequestId)
+	if pk.GetType() == userinput.UserInputResponsePacketStr {
+		userInputRespPk := pk.(*userinput.UserInputResponsePacketType)
+		uich, ok := scbus.MainRpcBus.GetRpcChannel(userInputRespPk.RequestId)
 		if !ok {
-			return fmt.Errorf("received User Input Response with invalid Id (%s): %v\n", userInputRespPk.RequestId, err)
+			return fmt.Errorf("received User Input Response with invalid Id (%s): %v", userInputRespPk.RequestId, err)
 		}
 		select {
 		case uich <- userInputRespPk:
@@ -302,7 +297,7 @@ func (ws *WSState) RunWSRead() {
 	if shell == nil {
 		return
 	}
-	shell.WriteJson(map[string]interface{}{"type": "hello"}) // let client know we accepted this connection, ignore error
+	shell.WriteJson(map[string]any{"type": "hello"}) // let client know we accepted this connection, ignore error
 	for msgBytes := range shell.ReadChan {
 		err := ws.processMessage(msgBytes)
 		if err != nil {