feat: first pass at sftp installer for waveshell

This does not completely work yet. In particular, you must attempt to
connect to the client before attempting an install. It also does not
work for auto installation yet.

go.work.sum

@@ -1,11 +1,45 @@
 github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8=
 github.com/kevinburke/ssh_config v1.2.0 h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4=
 github.com/kevinburke/ssh_config v1.2.0/go.mod h1:CT57kijsi8u/K/BOFA39wgDQJ9CxiF4nAY/ojJ6r6mM=
+github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
+github.com/pkg/sftp v1.13.6/go.mod h1:tz1ryNURKu77RL+GuCzmoJYxQczL3wLNNpPWagdg4Qk=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/wavetermdev/ssh_config v0.0.0-20240109090616-36c8da3d7376 h1:tFhJgTu7lgd+hldLfPSzDCoWUpXI8wHKR3rxq5jTLkQ=
 github.com/wavetermdev/ssh_config v0.0.0-20240109090616-36c8da3d7376/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.2.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.9.1/go.mod h1:owI94Op576fPu3cIGQeHs3joujW/2Oc6MtlxbF5dfNc=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

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/client.go

@@ -60,6 +60,18 @@ func (cw CmdWrap) Start() error {
 	return cw.Cmd.Start()
 }
 
+func (cw CmdWrap) StdinPipe() (io.WriteCloser, error) {
+	return cw.Cmd.StdinPipe()
+}
+
+func (cw CmdWrap) StdoutPipe() (io.ReadCloser, error) {
+	return cw.Cmd.StdoutPipe()
+}
+
+func (cw CmdWrap) StderrPipe() (io.ReadCloser, error) {
+	return cw.Cmd.StderrPipe()
+}
+
 type SessionWrap struct {
 	Session  *ssh.Session
 	StartCmd string
@@ -101,12 +113,35 @@ func (sw SessionWrap) Parser() (*packet.PacketParser, io.ReadCloser, io.ReadClos
 	return packetParser, io.NopCloser(stdoutReader), io.NopCloser(stderrReader), nil
 }
 
+func (sw SessionWrap) StdinPipe() (io.WriteCloser, error) {
+	return sw.Session.StdinPipe()
+}
+
+func (sw SessionWrap) StdoutPipe() (io.ReadCloser, error) {
+	outPipe, err := sw.Session.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	return io.NopCloser(outPipe), nil
+}
+
+func (sw SessionWrap) StderrPipe() (io.ReadCloser, error) {
+	errPipe, err := sw.Session.StderrPipe()
+	if err != nil {
+		return nil, err
+	}
+	return io.NopCloser(errPipe), nil
+}
+
 type ConnInterface interface {
 	Kill()
 	Wait() error
 	Sender() (*packet.PacketSender, io.WriteCloser, error)
 	Parser() (*packet.PacketParser, io.ReadCloser, io.ReadCloser, error)
 	Start() error
+	StdinPipe() (io.WriteCloser, error)
+	StdoutPipe() (io.ReadCloser, error)
+	StderrPipe() (io.ReadCloser, error)
 }
 
 type ClientProc struct {
@@ -175,6 +210,53 @@ func MakeClientProc(ctx context.Context, ecmd ConnInterface) (*ClientProc, *pack
 	return cproc, cproc.InitPk, nil
 }
 
+func MakeInstallProc(ctx context.Context, ecmd ConnInterface) (*ClientProc, *packet.InitPacketType, error) {
+	startTs := time.Now()
+	sender, inputWriter, err := ecmd.Sender()
+	if err != nil {
+		return nil, nil, err
+	}
+	packetParser, stdoutReader, stderrReader, err := ecmd.Parser()
+	if err != nil {
+		return nil, nil, err
+	}
+	err = ecmd.Start()
+	if err != nil {
+		return nil, nil, fmt.Errorf("running local client: %w", err)
+	}
+	cproc := &ClientProc{
+		Cmd:          ecmd,
+		StartTs:      startTs,
+		StdinWriter:  inputWriter,
+		StdoutReader: stdoutReader,
+		StderrReader: stderrReader,
+		Input:        sender,
+		Output:       packetParser,
+	}
+
+	var pk packet.PacketType
+	select {
+	case pk = <-packetParser.MainCh:
+	case <-ctx.Done():
+		cproc.Close()
+		return nil, nil, ctx.Err()
+	}
+
+	if pk != nil {
+		if pk.GetType() != packet.InitPacketStr {
+			cproc.Close()
+			return nil, nil, fmt.Errorf("invalid packet received from mshell client: %s", packet.AsString(pk))
+		}
+		initPk := pk.(*packet.InitPacketType)
+		cproc.InitPk = initPk
+	}
+	if cproc.InitPk == nil {
+		cproc.Close()
+		return nil, nil, fmt.Errorf("no init packet received from mshell client")
+	}
+	return cproc, cproc.InitPk, nil
+}
+
 func (cproc *ClientProc) Close() {
 	if cproc.Input != nil {
 		cproc.Input.Close()

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/go.mod

@@ -6,13 +6,14 @@ require (
 	github.com/alessio/shellescape v1.4.1
 	github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2
 	github.com/creack/pty v1.1.18
-	github.com/kevinburke/ssh_config v1.2.0
 	github.com/golang-migrate/migrate/v4 v4.16.2
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.5.0
 	github.com/jmoiron/sqlx v1.3.5
+	github.com/kevinburke/ssh_config v1.2.0
 	github.com/mattn/go-sqlite3 v1.14.16
+	github.com/pkg/sftp v1.13.6
 	github.com/sashabaranov/go-openai v1.9.0
 	github.com/sawka/txwrap v0.1.2
 	github.com/wavetermdev/waveterm/waveshell v0.0.0
@@ -26,8 +27,10 @@ require (
 	github.com/google/go-github/v57 v57.0.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/kr/fs v0.1.0 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 )
 
 replace github.com/wavetermdev/waveterm/waveshell => ../waveshell
+
 replace github.com/kevinburke/ssh_config => github.com/wavetermdev/ssh_config v0.0.0-20240109090616-36c8da3d7376

wavesrv/go.sum

@@ -29,6 +29,8 @@ github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+l
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g=
 github.com/jmoiron/sqlx v1.3.5/go.mod h1:nRVWtLre0KfCLJvgxzCsLVMogSvQ1zNJtpYr2Ccp0mQ=
+github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8=
+github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
@@ -36,6 +38,8 @@ github.com/lib/pq v1.10.2 h1:AqzbZs4ZoCBp+GtejcpCpcxM3zlSMx29dXbUSeVtJb8=
 github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/mattn/go-sqlite3 v1.14.16 h1:yOQRA0RpS5PFz/oikGwBEqvAWhWg5ufRz4ETLjwpU1Y=
 github.com/mattn/go-sqlite3 v1.14.16/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
+github.com/pkg/sftp v1.13.6 h1:JFZT4XbOU7l77xGSpOdW+pwIMqP044IyjXX6FGyEKFo=
+github.com/pkg/sftp v1.13.6/go.mod h1:tz1ryNURKu77RL+GuCzmoJYxQczL3wLNNpPWagdg4Qk=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rogpeppe/go-internal v1.10.1-0.20230524175051-ec119421bb97 h1:3RPlVWzZ/PDqmVuf/FKHARG5EMid/tl7cv54Sw/QRVY=
@@ -44,21 +48,56 @@ github.com/sashabaranov/go-openai v1.9.0/go.mod h1:lj5b/K+zjTSFxVLijLSTDZuP7adOg
 github.com/sawka/txwrap v0.1.2 h1:v8xS0Z1LE7/6vMZA81PYihI+0TSR6Zm1MalzzBIuXKc=
 github.com/sawka/txwrap v0.1.2/go.mod h1:T3nlw2gVpuolo6/XEetvBbk1oMXnY978YmBFy1UyHvw=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/wavetermdev/ssh_config v0.0.0-20240109090616-36c8da3d7376/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A=
 golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU=
 golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
 golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.10.0 h1:lFO9qtOdlre5W1jxS3r/4szv2/6iXxScdzjoBMXNhYk=
 golang.org/x/mod v0.10.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA=
 golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 mvdan.cc/sh/v3 v3.7.0 h1:lSTjdP/1xsddtaKfGg7Myu7DnlHItd3/M2tomOcNNBg=
 mvdan.cc/sh/v3 v3.7.0/go.mod h1:K2gwkaesF/D7av7Kxl0HbF5kGOd2ArupNTX3X44+8l8=

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

@@ -14,7 +14,9 @@ import (
 	"os"
 	"os/exec"
 	"path"
+	"path/filepath"
 	"regexp"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -25,6 +27,7 @@ import (
 	"github.com/armon/circbuf"
 	"github.com/creack/pty"
 	"github.com/google/uuid"
+	"github.com/pkg/sftp"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/base"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/packet"
 	"github.com/wavetermdev/waveterm/waveshell/pkg/server"
@@ -34,8 +37,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"
 )
@@ -86,6 +91,9 @@ const WaveshellServerRunOnlyFmt = `
   [%PINGPACKET%]
   mshell-[%VERSION%] --server
 `
+const InitCommand = `
+printf "\n##N{\"type\": \"init\", \"notfound\": true, \"uname\": \"%s|%s\"}\n" "$(uname -s)" "$(uname -m)";
+`
 
 func MakeLocalMShellCommandStr(isSudo bool) (string, error) {
 	mshellPath, err := scbase.LocalMShellBinaryPath()
@@ -142,6 +150,7 @@ type pendingStateKey struct {
 // remove once ssh library is stabilized
 type Launcher interface {
 	Launch(*MShellProc, bool)
+	RunInstall(*MShellProc)
 }
 
 type MShellProc struct {
@@ -173,6 +182,7 @@ type MShellProc struct {
 	RunningCmds      map[base.CommandKey]RunCmdType
 	PendingStateCmds map[pendingStateKey]base.CommandKey // key=[remoteinstance name]
 	launcher         Launcher                            // for conditional launch method based on ssh library in use. remove once ssh library is stabilized
+	Client           *ssh.Client
 }
 
 type RunCmdType struct {
@@ -199,6 +209,10 @@ func (msh *MShellProc) Launch(interactive bool) {
 	msh.launcher.Launch(msh, interactive)
 }
 
+func (msh *MShellProc) RunInstall() {
+	msh.launcher.RunInstall(msh)
+}
+
 func (msh *MShellProc) GetStatus() string {
 	msh.Lock.Lock()
 	defer msh.Lock.Unlock()
@@ -554,6 +568,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
 }
 
@@ -681,9 +698,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 {
@@ -943,13 +960,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 {
@@ -1052,7 +1069,197 @@ func (msh *MShellProc) WaitAndSendPassword(pw string) {
 	}
 }
 
-func (msh *MShellProc) RunInstall() {
+func (NewLauncher) RunInstall(msh *MShellProc) {
+	remoteCopy := msh.GetRemoteCopy()
+	if remoteCopy.Archived {
+		msh.WriteToPtyBuffer("*error: cannot install on archived remote\n")
+		return
+	}
+	baseStatus := msh.GetStatus()
+	if baseStatus == StatusConnecting || baseStatus == StatusConnected {
+		msh.WriteToPtyBuffer("*error: cannot install on remote that is connected/connecting, disconnect to install\n")
+		return
+	}
+	curStatus := msh.GetInstallStatus()
+	if curStatus == StatusConnecting {
+		msh.WriteToPtyBuffer("*error: cannot install on remote that is already trying to install, cancel current install to try again\n")
+		return
+	}
+
+	msh.WriteToPtyBuffer("installing mshell %s to %s...\n", scbase.MShellVersion, remoteCopy.RemoteCanonicalName)
+	clientCtx, clientCancelFn := context.WithCancel(context.Background())
+	defer clientCancelFn()
+	msh.WithLock(func() {
+		msh.InstallErr = nil
+		msh.InstallStatus = StatusConnecting
+		msh.InstallCancelFn = clientCancelFn
+		go msh.NotifyRemoteUpdate()
+	})
+
+	if msh.Remote.IsLocal() {
+		srcBinPath, err := scbase.MShellBinaryPath(base.MShellVersion, runtime.GOOS, runtime.GOARCH)
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		homeDir, err := os.UserHomeDir()
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		newBinName := fmt.Sprintf("mshell-%s", semver.MajorMinor(scbase.MShellVersion))
+		dstBinPath := filepath.Join(homeDir, ".mshell", newBinName)
+
+		srcBinFile, err := os.Open(srcBinPath)
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		defer srcBinFile.Close()
+		// destination file should be handled manually - not closed with defer
+		dstBinFile, err := os.OpenFile(dstBinPath, os.O_RDWR|os.O_CREATE, 0755)
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		_, err = io.Copy(dstBinFile, srcBinFile)
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			dstBinFile.Close()
+			return
+		}
+		err = dstBinFile.Close()
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+
+	} else {
+		session, err := msh.Client.NewSession()
+		defer session.Close()
+		if err != nil {
+			statusErr := fmt.Errorf("ssh cannot create session: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		cproc, initPk, err := shexec.MakeInstallProc(clientCtx, shexec.SessionWrap{Session: session, StartCmd: InitCommand}) //TODO
+		if err == context.Canceled {
+			msh.WriteToPtyBuffer("*install canceled\n")
+			msh.WithLock(func() {
+				msh.InstallStatus = StatusDisconnected
+				go msh.NotifyRemoteUpdate()
+			})
+			return
+		}
+		if err != nil {
+			statusErr := fmt.Errorf("cannot create init packet: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		defer cproc.Close()
+		goos, goarch, err := shexec.DetectGoArch(initPk.UName)
+		if err != nil {
+			statusErr := fmt.Errorf("cannot determine architecture: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		srcBinPath, err := scbase.MShellBinaryPath(base.MShellVersion, goos, goarch)
+		if err != nil {
+			statusErr := fmt.Errorf("cannot find source binary: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		fileClient, err := sftp.NewClient(msh.Client)
+		if err != nil {
+			statusErr := fmt.Errorf("cannot estabish sftp connection: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		homeDir, err := fileClient.Getwd()
+		if err != nil {
+			statusErr := fmt.Errorf("cannot determine remote home directory: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		newBinName := fmt.Sprintf("mshell-%s", semver.MajorMinor(scbase.MShellVersion))
+		dstBinPath := filepath.Join(homeDir, ".mshell", newBinName)
+
+		srcBinFile, err := os.Open(srcBinPath)
+		if err != nil {
+			statusErr := fmt.Errorf("cannot find source executable: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		defer srcBinFile.Close()
+		// destination file should be handled manually - not closed with defer
+		dstBinFile, err := fileClient.OpenFile(dstBinPath, os.O_RDWR|os.O_CREATE)
+		if err != nil {
+			statusErr := fmt.Errorf("cannot create destination executable: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		err = dstBinFile.Chmod(0755)
+		if err != nil {
+			statusErr := fmt.Errorf("cannot set executable permissions: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+		_, err = io.Copy(dstBinFile, srcBinFile)
+		if err != nil {
+			statusErr := fmt.Errorf("unable to copy to new waveshell executable: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			dstBinFile.Close()
+			return
+		}
+		err = dstBinFile.Close()
+		if err != nil {
+			statusErr := fmt.Errorf("unable to save waveshell executable: %w", err)
+			msh.WriteToPtyBuffer("*error, %s\n", statusErr.Error())
+			msh.setInstallErrorStatus(statusErr)
+			return
+		}
+	}
+	var connectMode string
+	msh.WithLock(func() {
+		msh.InstallStatus = StatusDisconnected
+		msh.InstallCancelFn = nil
+		msh.NeedsMShellUpgrade = false
+		msh.Status = StatusDisconnected
+		msh.Err = nil
+		connectMode = msh.Remote.ConnectMode
+	})
+	msh.WriteToPtyBuffer("successfully installed mshell %s to ~/.mshell\n", scbase.MShellVersion)
+	go msh.NotifyRemoteUpdate()
+	if connectMode == sstore.ConnectModeStartup || connectMode == sstore.ConnectModeAuto {
+		// the install was successful, and we don't have a manual connect mode, try to connect
+		go msh.Launch(true)
+	}
+	return
+}
+
+func (LegacyLauncher) RunInstall(msh *MShellProc) {
 	remoteCopy := msh.GetRemoteCopy()
 	if remoteCopy.Archived {
 		msh.WriteToPtyBuffer("*error: cannot install on archived remote\n")
@@ -1376,6 +1583,7 @@ func (NewLauncher) Launch(msh *MShellProc, interactive bool) {
 			msh.setErrorStatus(statusErr)
 			return
 		}
+		msh.Client = client
 		var session *ssh.Session
 		session, err = client.NewSession()
 		if err != nil {
@@ -2016,9 +2224,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)
@@ -2047,7 +2255,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
@@ -2059,7 +2267,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 {
@@ -2079,7 +2287,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))
@@ -2093,7 +2301,7 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
 			// fall-through (nothing to do)
 		}
 	}
-	sstore.MainBus.SendUpdate(update)
+	scbus.MainUpdateBus.DoUpdate(update)
 	return
 }
 
@@ -2122,13 +2330,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
@@ -2164,7 +2372,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 {
@@ -2193,9 +2401,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

@@ -22,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"
 )
@@ -104,14 +105,13 @@ func createPublicKeyCallback(sshKeywords *SshKeywords, passphrase string) func()
 			return createDummySigner()
 		}
 
-		request := &sstore.UserInputRequestType{
+		request := &userinput.UserInputRequestType{
 			ResponseType: "text",
 			QueryText:    fmt.Sprintf("Enter passphrase for the SSH key: %s", identityFile),
 			Title:        "Publickey Auth + Passphrase",
 		}
-		ctx, cancelFn := context.WithTimeout(context.Background(), 60*time.Second)
-		defer cancelFn()
-		response, err := sstore.MainBus.GetUserInput(ctx, request)
+		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
@@ -141,12 +141,12 @@ func createInteractivePasswordCallbackPrompt() func() (secret string, err error)
 		// in the future
 		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
 		}
@@ -201,12 +201,12 @@ func promptChallengeQuestion(question string, echo bool) (answer string, err err
 	// in the future
 	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
 	}
@@ -234,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
@@ -270,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 "+
@@ -280,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 "+
@@ -304,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)
 	}
 }
 
@@ -444,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")
 		}
 

wavesrv/pkg/scbus/modelupdate.go

@@ -0,0 +1,126 @@
+// 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 (mu *ModelUpdatePacketType) IsEmpty() bool {
+	if mu == nil || mu.Data == nil {
+		return true
+	}
+	return mu.Data.IsEmpty()
+}
+
+// Clean the ClientData in an update, if present
+func (upk *ModelUpdatePacketType) Clean() {
+	if upk == nil || upk.Data == nil {
+		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 {
+	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{}))
+}

wavesrv/pkg/scbus/ptydataupdate.go

@@ -0,0 +1,50 @@
+// 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{}))
+}