enhanced ssh session handling a bit

backend/internal/handlers/ssh-broker-runtime.go

@@ -211,10 +211,35 @@ func (r *SSHSessionRegistry) RequestClose(id string, reason string) bool {
 }
 
 func (s *sshActiveSession) SetResources(ws *websocket.Conn, client *ssh.Client, session *ssh.Session, stdin io.WriteCloser) {
+	var closeWS *websocket.Conn
+	var closeClient *ssh.Client
+	var closeSession *ssh.Session
+	var closeStdin io.WriteCloser
+
 	if s == nil {
 		return
 	}
 	s.mu.Lock()
+	if s.closed {
+		closeWS = ws
+		closeClient = client
+		closeSession = session
+		closeStdin = stdin
+		s.mu.Unlock()
+		if closeStdin != nil {
+			_ = closeStdin.Close()
+		}
+		if closeSession != nil {
+			_ = closeSession.Close()
+		}
+		if closeClient != nil {
+			_ = closeClient.Close()
+		}
+		if closeWS != nil {
+			_ = closeWS.Close()
+		}
+		return
+	}
 	if ws != nil {
 		s.ws = ws
 	}

backend/internal/handlers/ssh-broker.go

@@ -22,6 +22,9 @@ import "codit/internal/middleware"
 import "codit/internal/models"
 
 const logIDSSHBroker string = "ssh-broker"
+const sshWebsocketWriteTimeout time.Duration = 5 * time.Second
+const sshSessionInputBufferSize int = 64
+
 type sshServerRequest struct {
 	Name          string   `json:"name"`
 	Host          string   `json:"host"`
@@ -137,6 +140,11 @@ type sshWorkspaceAttachment struct {
 	InputErrCh chan error
 }
 
+type sshWorkspaceAttachmentDone struct {
+	SessionID   string
+	Attachment *sshWorkspaceAttachment
+}
+
 const sshAuthMethodManagedSSHCert string = "managed_ssh_cert"
 const sshAuthMethodPromptedPassword string = "prompted_password"
 const sshAuthMethodStoredPassword string = "stored_password"
@@ -2392,14 +2400,27 @@ func (api *API) serveSSHSessionStream(ws *websocket.Conn, store *db.Store, user
 	var sendMu sync.Mutex
 	var inputCh chan sshSessionStreamMessage
 	var inputErrCh chan error
+	var readerErrCh chan error
 	var prepared sshSessionPrepared
 	var preparedOK bool
 	var preparedPtr *sshSessionPrepared
 
 	defer ws.Close()
-	inputCh = make(chan sshSessionStreamMessage)
+	inputCh = make(chan sshSessionStreamMessage, sshSessionInputBufferSize)
 	inputErrCh = make(chan error, 1)
-	go api.readSSHSessionWebsocket(ws, inputCh, inputErrCh) // read websocket and write to inputCh
+	readerErrCh = make(chan error, 1)
+	go api.readSSHSessionWebsocket(ws, inputCh, readerErrCh) // read websocket and write to inputCh
+	go func() {
+		var err error
+		var reason string
+
+		err = <-readerErrCh
+		reason = sshSessionWebsocketCloseReason(err)
+		if api.SSHSessionRegistry != nil {
+			api.SSHSessionRegistry.RequestClose(sessionItem.ID, reason)
+		}
+		inputErrCh <- err
+	}()
 	prepared, preparedOK = api.SSHPreparedSessionStore.Take(sessionItem.ID)
 	if !preparedOK {
 		prepared = sshSessionPrepared{}
@@ -2410,8 +2431,22 @@ func (api *API) serveSSHSessionStream(ws *websocket.Conn, store *db.Store, user
 	// this reads the inputCh
 	api.runSSHSessionStream(store, user, sessionItem, ws, inputCh, inputErrCh,
 		preparedPtr,
-		func(msg sshSessionStreamMessage) { api.sendSSHSessionWS(&sendMu, ws, msg) },
-		func(data []byte) { api.sendSSHSessionBinary(&sendMu, ws, data) })
+		func(msg sshSessionStreamMessage) {
+			var err error
+
+			err = api.sendSSHSessionWS(&sendMu, ws, msg)
+			if err != nil && api.SSHSessionRegistry != nil {
+				api.SSHSessionRegistry.RequestClose(sessionItem.ID, "websocket send failed")
+			}
+		},
+		func(data []byte) {
+			var err error
+
+			err = api.sendSSHSessionBinary(&sendMu, ws, data)
+			if err != nil && api.SSHSessionRegistry != nil {
+				api.SSHSessionRegistry.RequestClose(sessionItem.ID, "websocket send failed")
+			}
+		})
 }
 
 func (api *API) serveSSHWorkspaceStream(ws *websocket.Conn, store *db.Store, user models.User, remoteAddr string) {
@@ -2426,10 +2461,15 @@ func (api *API) serveSSHWorkspaceStream(ws *websocket.Conn, store *db.Store, use
 	var preparedOK bool
 	var err error
 	var ok bool
+	var attachedID string
+	var attachmentDoneCh chan sshWorkspaceAttachmentDone
+	var attachmentDone sshWorkspaceAttachmentDone
+	var websocketCloseReason string
 
 	defer ws.Close()
-	controlCh = make(chan sshSessionStreamMessage)
+	controlCh = make(chan sshSessionStreamMessage, sshSessionInputBufferSize)
 	controlErrCh = make(chan error, 1)
+	attachmentDoneCh = make(chan sshWorkspaceAttachmentDone, sshSessionInputBufferSize)
 	attachments = map[string]*sshWorkspaceAttachment{}
 	go api.readSSHSessionWebsocket(ws, controlCh, controlErrCh)
 
@@ -2437,20 +2477,27 @@ event_loop:
 	for {
 		select {
 		case err = <-controlErrCh:
+			websocketCloseReason = sshSessionWebsocketCloseReason(err)
 			api.Logger.Write(logIDSSHBroker, codit_logger.LOG_WARN,
 				"workspace stream closed requester=%s user=%s err=%v",
 				remoteAddr,
 				user.Username,
 				err)
 
-			// write the error message to the error notification channel of all existing attachments
-			for _, attachment = range attachments {
+			for attachedID, attachment = range attachments {
+				if api.SSHSessionRegistry != nil {
+					api.SSHSessionRegistry.RequestClose(attachedID, websocketCloseReason)
+				}
 				select {
 				case attachment.InputErrCh <- err:
 				default: // make it non-blocking
 				}
 			}
 			break event_loop
+		case attachmentDone = <-attachmentDoneCh:
+			if attachments[attachmentDone.SessionID] == attachmentDone.Attachment {
+				delete(attachments, attachmentDone.SessionID)
+			}
 		case msg = <-controlCh:
 			if msg.Type == "attach" {
 				var attachedItem models.SSHSession
@@ -2493,7 +2540,7 @@ event_loop:
 					continue
 				}
 				attachment = &sshWorkspaceAttachment{
-					InputCh:    make(chan sshSessionStreamMessage, 16),
+					InputCh:    make(chan sshSessionStreamMessage, sshSessionInputBufferSize),
 					InputErrCh: make(chan error, 1),
 				}
 				attachments[msg.SessionID] = attachment
@@ -2510,19 +2557,40 @@ event_loop:
 						attachedPrepared = &prepared
 					}
 				}
-				go api.runSSHSessionStream(store, user, attachedItem, nil, attachment.InputCh, attachment.InputErrCh,
-					attachedPrepared,
+				go func(runItem models.SSHSession, runPrepared *sshSessionPrepared, runSessionID string, runAttachment *sshWorkspaceAttachment) {
+					defer func() {
+						select {
+						case attachmentDoneCh <- sshWorkspaceAttachmentDone{SessionID: runSessionID, Attachment: runAttachment}:
+						default:
+						}
+					}()
+					api.runSSHSessionStream(store, user, runItem, nil, runAttachment.InputCh, runAttachment.InputErrCh,
+						runPrepared,
 						func(status sshSessionStreamMessage) {
-						status.SessionID = attachedSessionID
-						api.sendSSHSessionWS(&sendMu, ws, status)
+							var sendErr error
+
+							status.SessionID = runSessionID
+							sendErr = api.sendSSHSessionWS(&sendMu, ws, status)
+							if sendErr != nil && api.SSHSessionRegistry != nil {
+								api.SSHSessionRegistry.RequestClose(runSessionID, "websocket send failed")
+							}
 						},
 						func(data []byte) {
-						api.sendSSHWorkspaceOutput(&sendMu, ws, attachedSessionID, data)
+							var sendErr error
+
+							sendErr = api.sendSSHWorkspaceOutput(&sendMu, ws, runSessionID, data)
+							if sendErr != nil && api.SSHSessionRegistry != nil {
+								api.SSHSessionRegistry.RequestClose(runSessionID, "websocket send failed")
+							}
 						})
+				}(attachedItem, attachedPrepared, attachedSessionID, attachment)
 				continue
 			}
 			attachment, ok = attachments[msg.SessionID]
 			if !ok {
+				if msg.Type == "input" || msg.Type == "resize" || msg.Type == "detach" {
+					continue
+				}
 				api.sendSSHSessionWS(&sendMu, ws, sshSessionStreamMessage{
 					SessionID: msg.SessionID,
 					Type:      "status",
@@ -2532,6 +2600,9 @@ event_loop:
 				continue
 			}
 			if msg.Type == "detach" {
+				if api.SSHSessionRegistry != nil {
+					api.SSHSessionRegistry.RequestClose(msg.SessionID, "detached")
+				}
 				select {
 				case attachment.InputErrCh <- io.EOF:
 				default:
@@ -2541,7 +2612,23 @@ event_loop:
 			}
 
 			/* other message types. write to the channel */
-			attachment.InputCh <- msg
+			select {
+			case attachment.InputCh <- msg:
+			default:
+				api.Logger.Write(logIDSSHBroker, codit_logger.LOG_WARN,
+					"workspace input queue full session=%s requester=%s user=%s type=%s",
+					msg.SessionID,
+					remoteAddr,
+					user.Username,
+					msg.Type)
+				if api.SSHSessionRegistry != nil {
+					api.SSHSessionRegistry.RequestClose(msg.SessionID, "websocket input queue full")
+				}
+				select {
+				case attachment.InputErrCh <- errors.New("websocket input queue full"):
+				default:
+				}
+			}
 		}
 	}
 }
@@ -2716,8 +2803,28 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 			pinnedHostKey.Algorithm)
 	}
 	runtimeSession.SetResources(transportWS, sshClient, nil, nil)
+	closeReason = runtimeSession.CloseReason()
+	if closeReason != "" {
+		now = time.Now().UTC().Unix()
+		_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+		api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+		if sendStatus != nil {
+			sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+		}
+		return
+	}
 	sshSession, err = sshClient.NewSession()
 	if err != nil {
+		closeReason = runtimeSession.CloseReason()
+		if closeReason != "" {
+			now = time.Now().UTC().Unix()
+			_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+			api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+			if sendStatus != nil {
+				sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+			}
+			return
+		}
 		api.logSSHSessionConnectFailure(sessionItem, profile, user, "new_session", err)
 		if sendStatus != nil {
 			sendStatus(sshSessionStreamMessage{Type: "status", Status: "error", Message: err.Error()})
@@ -2728,8 +2835,28 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 	}
 	defer sshSession.Close()
 	runtimeSession.SetResources(transportWS, sshClient, sshSession, nil)
+	closeReason = runtimeSession.CloseReason()
+	if closeReason != "" {
+		now = time.Now().UTC().Unix()
+		_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+		api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+		if sendStatus != nil {
+			sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+		}
+		return
+	}
 	stdin, err = sshSession.StdinPipe()
 	if err != nil {
+		closeReason = runtimeSession.CloseReason()
+		if closeReason != "" {
+			now = time.Now().UTC().Unix()
+			_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+			api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+			if sendStatus != nil {
+				sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+			}
+			return
+		}
 		api.logSSHSessionConnectFailure(sessionItem, profile, user, "stdin_pipe", err)
 		if sendStatus != nil {
 			sendStatus(sshSessionStreamMessage{Type: "status", Status: "error", Message: err.Error()})
@@ -2740,6 +2867,16 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 	}
 	stdout, err = sshSession.StdoutPipe()
 	if err != nil {
+		closeReason = runtimeSession.CloseReason()
+		if closeReason != "" {
+			now = time.Now().UTC().Unix()
+			_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+			api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+			if sendStatus != nil {
+				sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+			}
+			return
+		}
 		api.logSSHSessionConnectFailure(sessionItem, profile, user, "stdout_pipe", err)
 		if sendStatus != nil {
 			sendStatus(sshSessionStreamMessage{Type: "status", Status: "error", Message: err.Error()})
@@ -2750,6 +2887,16 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 	}
 	stderr, err = sshSession.StderrPipe()
 	if err != nil {
+		closeReason = runtimeSession.CloseReason()
+		if closeReason != "" {
+			now = time.Now().UTC().Unix()
+			_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+			api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+			if sendStatus != nil {
+				sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+			}
+			return
+		}
 		api.logSSHSessionConnectFailure(sessionItem, profile, user, "stderr_pipe", err)
 		if sendStatus != nil {
 			sendStatus(sshSessionStreamMessage{Type: "status", Status: "error", Message: err.Error()})
@@ -2774,6 +2921,16 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 		ssh.TTY_OP_OSPEED: 14400,
 	})
 	if err != nil {
+		closeReason = runtimeSession.CloseReason()
+		if closeReason != "" {
+			now = time.Now().UTC().Unix()
+			_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+			api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+			if sendStatus != nil {
+				sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+			}
+			return
+		}
 		api.logSSHSessionConnectFailure(sessionItem, profile, user, "request_pty", err)
 		if sendStatus != nil {
 			sendStatus(sshSessionStreamMessage{Type: "status", Status: "error", Message: err.Error()})
@@ -2782,8 +2939,28 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 		_ = store.UpdateSSHSessionStatus(sessionItem.ID, "error", connectedFingerprint, 0, now, err.Error())
 		return
 	}
+	closeReason = runtimeSession.CloseReason()
+	if closeReason != "" {
+		now = time.Now().UTC().Unix()
+		_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+		api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+		if sendStatus != nil {
+			sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+		}
+		return
+	}
 	err = sshSession.Shell()
 	if err != nil {
+		closeReason = runtimeSession.CloseReason()
+		if closeReason != "" {
+			now = time.Now().UTC().Unix()
+			_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+			api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+			if sendStatus != nil {
+				sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+			}
+			return
+		}
 		api.logSSHSessionConnectFailure(sessionItem, profile, user, "shell", err)
 		if sendStatus != nil {
 			sendStatus(sshSessionStreamMessage{Type: "status", Status: "error", Message: err.Error()})
@@ -2792,6 +2969,16 @@ func (api *API) runSSHSessionStream(store *db.Store, user models.User, sessionIt
 		_ = store.UpdateSSHSessionStatus(sessionItem.ID, "error", connectedFingerprint, 0, now, err.Error())
 		return
 	}
+	closeReason = runtimeSession.CloseReason()
+	if closeReason != "" {
+		now = time.Now().UTC().Unix()
+		_ = store.UpdateSSHSessionStatus(sessionItem.ID, "closed", connectedFingerprint, 0, now, closeReason)
+		api.logSSHSessionClosed(sessionItem, profile, user, "closed", closeReason, 0, now)
+		if sendStatus != nil {
+			sendStatus(sshSessionStreamMessage{Type: "status", Status: "closed", Message: closeReason})
+		}
+		return
+	}
 	connectedAt = time.Now().UTC().Unix()
 	_ = store.UpdateSSHSessionStatus(sessionItem.ID, "connected", connectedFingerprint, connectedAt, 0, "")
 	api.logSSHSessionConnectSuccess(sessionItem, profile, user, connectedFingerprint)
@@ -3137,28 +3324,66 @@ func (api *API) readSSHSessionWebsocket(ws *websocket.Conn, inputCh chan<- sshSe
 			errCh <- err
 			return
 		}
-		inputCh <- msg
+		select {
+		case inputCh <- msg:
+		default:
+			errCh <- errors.New("websocket input queue full")
+			return
+		}
 	}
 }
 
-func (api *API) sendSSHSessionWS(mu *sync.Mutex, ws *websocket.Conn, msg sshSessionStreamMessage) {
-	if mu == nil { return }
-	mu.Lock()
-	defer mu.Unlock()
-	_ = websocket.JSON.Send(ws, msg)
+func sshSessionWebsocketCloseReason(err error) string {
+	if err != nil && strings.Contains(err.Error(), "input queue full") {
+		return err.Error()
+	}
+	return "websocket closed"
 }
 
-func (api *API) sendSSHSessionBinary(mu *sync.Mutex, ws *websocket.Conn, data []byte) {
-	if mu == nil { return }
+func (api *API) sendSSHSessionWS(mu *sync.Mutex, ws *websocket.Conn, msg sshSessionStreamMessage) error {
+	var err error
+
+	if mu == nil || ws == nil { return nil }
 	mu.Lock()
 	defer mu.Unlock()
-	_ = websocket.Message.Send(ws, data)
+	_ = ws.SetWriteDeadline(time.Now().Add(sshWebsocketWriteTimeout))
+	err = websocket.JSON.Send(ws, msg)
+	_ = ws.SetWriteDeadline(time.Time{})
+	if err != nil {
+		api.Logger.Write(logIDSSHBroker, codit_logger.LOG_WARN,
+			"websocket json send failed session=%s type=%s err=%v",
+			msg.SessionID,
+			msg.Type,
+			err)
+	}
+	return err
 }
 
-func (api *API) sendSSHWorkspaceOutput(mu *sync.Mutex, ws *websocket.Conn, sessionID string, data []byte) {
-	api.sendSSHSessionWS(mu, ws, sshSessionStreamMessage{
+func (api *API) sendSSHSessionBinary(mu *sync.Mutex, ws *websocket.Conn, data []byte) error {
+	var err error
+
+	if mu == nil || ws == nil { return nil }
+	mu.Lock()
+	defer mu.Unlock()
+	_ = ws.SetWriteDeadline(time.Now().Add(sshWebsocketWriteTimeout))
+	err = websocket.Message.Send(ws, data)
+	_ = ws.SetWriteDeadline(time.Time{})
+	if err != nil {
+		api.Logger.Write(logIDSSHBroker, codit_logger.LOG_WARN,
+			"websocket binary send failed bytes=%d err=%v",
+			len(data),
+			err)
+	}
+	return err
+}
+
+func (api *API) sendSSHWorkspaceOutput(mu *sync.Mutex, ws *websocket.Conn, sessionID string, data []byte) error {
+	var err error
+
+	err = api.sendSSHSessionWS(mu, ws, sshSessionStreamMessage{
 		SessionID: sessionID,
 		Type:      "output",
 		Data:      base64.StdEncoding.EncodeToString(data),
 	})
+	return err
 }

frontend/src/pages/SSHSessionPage.tsx

@@ -13,7 +13,7 @@ import Typography from '@mui/material/Typography'
 import useMediaQuery from '@mui/material/useMediaQuery'
 import { useTheme } from '@mui/material/styles'
 import MoreVertIcon from '@mui/icons-material/MoreVert'
-import { PointerEvent as ReactPointerEvent, useCallback, useEffect, useMemo, useRef, useState } from 'react'
+import { MutableRefObject, PointerEvent as ReactPointerEvent, useCallback, useEffect, useMemo, useRef, useState } from 'react'
 import { useLocation, useNavigate, useParams } from 'react-router-dom'
 import { FitAddon, ITerminalDimensions } from '@xterm/addon-fit'
 import { Unicode11Addon } from '@xterm/addon-unicode11'
@@ -144,6 +144,10 @@ function downloadTextFile(filename: string, text: string): void {
 	window.URL.revokeObjectURL(url)
 }
 
+function isSSHSessionStreamActive(status: string): boolean {
+	return status !== 'closed' && status !== 'error'
+}
+
 function SessionTerminalPanel(props: SessionPanelProps) {
 	const { sessionID, layoutToken, registerStream, unregisterStream, sendStreamInput, sendStreamResize, onDuplicate, onClose, onReplace, onSummaryChange } = props
 	const [session, setSession] = useState<SSHSession | null>(null)
@@ -163,6 +167,7 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 	const terminalRef = useRef<Terminal | null>(null)
 	const fitAddonRef = useRef<FitAddon | null>(null)
 	const unicodeAddonRef = useRef<Unicode11Addon | null>(null)
+	const streamActiveRef: MutableRefObject<boolean> = useRef<boolean>(true)
 	const sessionReady = Boolean(session)
 	const sessionTitle = useMemo(() => buildSessionTitle(session), [session])
 	const canReconnect = status === 'closed' || status === 'error'
@@ -179,6 +184,7 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 		const fitAddon = fitAddonRef.current
 		let dimensions: ITerminalDimensions | undefined
 
+		if (!streamActiveRef.current) { return }
 		if (!fitAddon) { return }
 		fitAddon.fit()
 		dimensions = fitAddon.proposeDimensions()
@@ -195,6 +201,7 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 		setSession(null)
 		setStatus('pending')
 		setError(null)
+		streamActiveRef.current = true
 
 		const load = async () => {
 			let item: SSHSession
@@ -209,6 +216,7 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 				if (cancelled) { return }
 				setSession(item)
 				setStatus(item.status || 'pending')
+				streamActiveRef.current = isSSHSessionStreamActive(item.status || 'pending')
 			} catch (err) {
 				if (!cancelled) setError(err instanceof Error ? err.message : 'Failed to load SSH session')
 			} finally {
@@ -309,15 +317,21 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 		fitAddonRef.current = fitAddon
 		unicodeAddonRef.current = unicodeAddon
 
-		term.onData((data: string) => { sendStreamInput(sessionID, data) })
+		term.onData((data: string) => {
+			if (!streamActiveRef.current) { return }
+			sendStreamInput(sessionID, data)
+		})
 
 		registerStream(sessionID, {
 			onOutput: (data: Uint8Array<ArrayBufferLike>) => {
 				if (!disposed && term) term.write(data)
 			},
 			onStatus: (message: SSHSessionStreamMessage) => {
+				const nextStatus: string = message.status || 'unknown'
+
 				if (disposed || !term) return
-				setStatus(message.status || 'unknown')
+				streamActiveRef.current = isSSHSessionStreamActive(nextStatus)
+				setStatus(nextStatus)
 				if (message.status === 'connected') {
 					setSession((prev) => (prev ? { ...prev, host_key_fingerprint: message.message || prev.host_key_fingerprint, status: 'connected' } : prev))
 					scheduleResize()
@@ -335,6 +349,7 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 			},
 			onTransportClosed: (message: string) => {
 				if (disposed) return
+				streamActiveRef.current = false
 				setStatus((prev) => (prev === 'closed' || prev === 'error' ? prev : 'closed'))
 				if (message) setError((prev) => prev || message)
 			}
@@ -383,6 +398,7 @@ function SessionTerminalPanel(props: SessionPanelProps) {
 		setActionPending(true)
 		try {
 			await api.disconnectSSHSessionForSelf(sessionID)
+			streamActiveRef.current = false
 			setStatus('closed')
 		} catch (err) {
 			setError(err instanceof Error ? err.message : 'Failed to disconnect SSH session')