FEATURE: Add Session API for pipe-based bash sessions in execd

components/execd/pkg/runtime/bash_session_windows.go

@@ -0,0 +1,40 @@
+// Copyright 2026 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build windows
+// +build windows
+
+package runtime
+
+import (
+	"context"
+	"errors"
+)
+
+var errBashSessionNotSupported = errors.New("bash session is not supported on windows")
+
+// CreateBashSession is not supported on Windows.
+func (c *Controller) CreateBashSession(_ *CreateContextRequest) (string, error) { //nolint:revive
+	return "", errBashSessionNotSupported
+}
+
+// RunInBashSession is not supported on Windows.
+func (c *Controller) RunInBashSession(_ context.Context, _ *ExecuteCodeRequest) error { //nolint:revive
+	return errBashSessionNotSupported
+}
+
+// DeleteBashSession is not supported on Windows.
+func (c *Controller) DeleteBashSession(_ string) error { //nolint:revive
+	return errBashSessionNotSupported
+}

components/execd/pkg/runtime/command_common.go

@@ -46,18 +46,17 @@ func (c *Controller) tailStdPipe(file string, onExecute func(text string), done
 
 // getCommandKernel retrieves a command execution context.
 func (c *Controller) getCommandKernel(sessionID string) *commandKernel {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	return c.commandClientMap[sessionID]
+	if v, ok := c.commandClientMap.Load(sessionID); ok {
+		if kernel, ok := v.(*commandKernel); ok {
+			return kernel
+		}
+	}
+	return nil
 }
 
 // storeCommandKernel registers a command execution context.
 func (c *Controller) storeCommandKernel(sessionID string, kernel *commandKernel) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.commandClientMap[sessionID] = kernel
+	c.commandClientMap.Store(sessionID, kernel)
 }
 
 // stdLogDescriptor creates temporary files for capturing command output.

components/execd/pkg/runtime/command_status.go

@@ -40,11 +40,11 @@ type CommandOutput struct {
 }
 
 func (c *Controller) commandSnapshot(session string) *commandKernel {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	kernel, ok := c.commandClientMap[session]
-	if !ok || kernel == nil {
+	var kernel *commandKernel
+	if v, ok := c.commandClientMap.Load(session); ok {
+		kernel, _ = v.(*commandKernel)
+	}
+	if kernel == nil {
 		return nil
 	}
 
@@ -116,8 +116,11 @@ func (c *Controller) markCommandFinished(session string, exitCode int, errMsg st
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	kernel, ok := c.commandClientMap[session]
-	if !ok || kernel == nil {
+	var kernel *commandKernel
+	if v, ok := c.commandClientMap.Load(session); ok {
+		kernel, _ = v.(*commandKernel)
+	}
+	if kernel == nil {
 		return
 	}
 

components/execd/pkg/runtime/context.go

@@ -31,7 +31,9 @@ import (
 )
 
 // CreateContext provisions a kernel-backed session and returns its ID.
+// Bash language uses Jupyter kernel like other languages; for pipe-based bash sessions use CreateBashSession (session API).
 func (c *Controller) CreateContext(req *CreateContextRequest) (string, error) {
+	// Create a new Jupyter session.
 	var (
 		client  *jupyter.Client
 		session *jupytersession.Session
@@ -42,7 +44,7 @@ func (c *Controller) CreateContext(req *CreateContextRequest) (string, error) {
 		log.Error("failed to create session, retrying: %v", err)
 		return err != nil
 	}, func() error {
-		client, session, err = c.createContext(*req)
+		client, session, err = c.createJupyterContext(*req)
 		return err
 	})
 	if err != nil {
@@ -114,20 +116,11 @@ func (c *Controller) deleteSessionAndCleanup(session string) error {
 	if c.getJupyterKernel(session) == nil {
 		return ErrContextNotFound
 	}
-
 	if err := c.jupyterClient().DeleteSession(session); err != nil {
 		return err
 	}
-
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	delete(c.jupyterClientMap, session)
-	for lang, id := range c.defaultLanguageJupyterSessions {
-		if id == session {
-			delete(c.defaultLanguageJupyterSessions, lang)
-		}
-	}
+	c.jupyterClientMap.Delete(session)
+	c.deleteDefaultSessionByID(session)
 	return nil
 }
 
@@ -146,8 +139,12 @@ func (c *Controller) newIpynbPath(sessionID, cwd string) (string, error) {
 	return filepath.Join(cwd, fmt.Sprintf("%s.ipynb", sessionID)), nil
 }
 
-// createDefaultLanguageContext prewarms a session for stateless execution.
-func (c *Controller) createDefaultLanguageContext(language Language) error {
+// createDefaultLanguageJupyterContext prewarms a session for stateless execution.
+func (c *Controller) createDefaultLanguageJupyterContext(language Language) error {
+	if c.getDefaultLanguageSession(language) != "" {
+		return nil
+	}
+
 	var (
 		client  *jupyter.Client
 		session *jupytersession.Session
@@ -157,7 +154,7 @@ func (c *Controller) createDefaultLanguageContext(language Language) error {
 		log.Error("failed to create context, retrying: %v", err)
 		return err != nil
 	}, func() error {
-		client, session, err = c.createContext(CreateContextRequest{
+		client, session, err = c.createJupyterContext(CreateContextRequest{
 			Language: language,
 			Cwd:      "",
 		})
@@ -167,20 +164,17 @@ func (c *Controller) createDefaultLanguageContext(language Language) error {
 		return err
 	}
 
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.defaultLanguageJupyterSessions[language] = session.ID
-	c.jupyterClientMap[session.ID] = &jupyterKernel{
+	c.setDefaultLanguageSession(language, session.ID)
+	c.jupyterClientMap.Store(session.ID, &jupyterKernel{
 		kernelID: session.Kernel.ID,
 		client:   client,
 		language: language,
-	}
+	})
 	return nil
 }
 
-// createContext performs the actual context creation workflow.
-func (c *Controller) createContext(request CreateContextRequest) (*jupyter.Client, *jupytersession.Session, error) {
+// createJupyterContext performs the actual context creation workflow.
+func (c *Controller) createJupyterContext(request CreateContextRequest) (*jupyter.Client, *jupytersession.Session, error) {
 	client := c.jupyterClient()
 
 	kernel, err := c.searchKernel(client, request.Language)
@@ -220,10 +214,7 @@ func (c *Controller) createContext(request CreateContextRequest) (*jupyter.Clien
 
 // storeJupyterKernel caches a session -> kernel mapping.
 func (c *Controller) storeJupyterKernel(sessionID string, kernel *jupyterKernel) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.jupyterClientMap[sessionID] = kernel
+	c.jupyterClientMap.Store(sessionID, kernel)
 }
 
 func (c *Controller) jupyterClient() *jupyter.Client {
@@ -239,49 +230,63 @@ func (c *Controller) jupyterClient() *jupyter.Client {
 		jupyter.WithHTTPClient(httpClient))
 }
 
-func (c *Controller) listAllContexts() ([]CodeContext, error) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
+func (c *Controller) getDefaultLanguageSession(language Language) string {
+	if v, ok := c.defaultLanguageSessions.Load(language); ok {
+		if session, ok := v.(string); ok {
+			return session
+		}
+	}
+	return ""
+}
+
+func (c *Controller) setDefaultLanguageSession(language Language, sessionID string) {
+	c.defaultLanguageSessions.Store(language, sessionID)
+}
 
+func (c *Controller) deleteDefaultSessionByID(sessionID string) {
+	c.defaultLanguageSessions.Range(func(key, value any) bool {
+		if s, ok := value.(string); ok && s == sessionID {
+			c.defaultLanguageSessions.Delete(key)
+		}
+		return true
+	})
+}
+
+func (c *Controller) listAllContexts() ([]CodeContext, error) {
 	contexts := make([]CodeContext, 0)
-	for session, kernel := range c.jupyterClientMap {
-		if kernel != nil {
-			contexts = append(contexts, CodeContext{
-				ID:       session,
-				Language: kernel.language,
-			})
+	c.jupyterClientMap.Range(func(key, value any) bool {
+		session, _ := key.(string)
+		if kernel, ok := value.(*jupyterKernel); ok && kernel != nil {
+			contexts = append(contexts, CodeContext{ID: session, Language: kernel.language})
 		}
-	}
+		return true
+	})
 
-	for language, defaultContext := range c.defaultLanguageJupyterSessions {
-		contexts = append(contexts, CodeContext{
-			ID:       defaultContext,
-			Language: language,
-		})
-	}
+	c.defaultLanguageSessions.Range(func(key, value any) bool {
+		lang, _ := key.(Language)
+		session, _ := value.(string)
+		if session == "" {
+			return true
+		}
+		contexts = append(contexts, CodeContext{ID: session, Language: lang})
+		return true
+	})
 
 	return contexts, nil
 }
 
 func (c *Controller) listLanguageContexts(language Language) ([]CodeContext, error) {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
 	contexts := make([]CodeContext, 0)
-	for session, kernel := range c.jupyterClientMap {
-		if kernel != nil && kernel.language == language {
-			contexts = append(contexts, CodeContext{
-				ID:       session,
-				Language: language,
-			})
+	c.jupyterClientMap.Range(func(key, value any) bool {
+		session, _ := key.(string)
+		if kernel, ok := value.(*jupyterKernel); ok && kernel != nil && kernel.language == language {
+			contexts = append(contexts, CodeContext{ID: session, Language: language})
 		}
-	}
+		return true
+	})
 
-	if defaultContext := c.defaultLanguageJupyterSessions[language]; defaultContext != "" {
-		contexts = append(contexts, CodeContext{
-			ID:       defaultContext,
-			Language: language,
-		})
+	if defaultContext := c.getDefaultLanguageSession(language); defaultContext != "" {
+		contexts = append(contexts, CodeContext{ID: defaultContext, Language: language})
 	}
 
 	return contexts, nil

components/execd/pkg/runtime/context_test.go

@@ -27,8 +27,8 @@ import (
 
 func TestListContextsAndNewIpynbPath(t *testing.T) {
 	c := NewController("http://example", "token")
-	c.jupyterClientMap["session-python"] = &jupyterKernel{language: Python}
-	c.defaultLanguageJupyterSessions[Go] = "session-go-default"
+	c.jupyterClientMap.Store("session-python", &jupyterKernel{language: Python})
+	c.defaultLanguageSessions.Store(Go, "session-go-default")
 
 	pyContexts, err := c.listLanguageContexts(Python)
 	require.NoError(t, err)
@@ -107,13 +107,13 @@ func TestDeleteContext_RemovesCacheOnSuccess(t *testing.T) {
 	defer server.Close()
 
 	c := NewController(server.URL, "token")
-	c.jupyterClientMap[sessionID] = &jupyterKernel{language: Python}
-	c.defaultLanguageJupyterSessions[Python] = sessionID
+	c.jupyterClientMap.Store(sessionID, &jupyterKernel{language: Python})
+	c.defaultLanguageSessions.Store(Python, sessionID)
 
 	require.NoError(t, c.DeleteContext(sessionID))
 
 	require.Nil(t, c.getJupyterKernel(sessionID), "expected cache to be cleared")
-	_, ok := c.defaultLanguageJupyterSessions[Python]
+	_, ok := c.defaultLanguageSessions.Load(Python)
 	require.False(t, ok, "expected default session entry to be removed")
 }
 
@@ -138,17 +138,17 @@ func TestDeleteLanguageContext_RemovesCacheOnSuccess(t *testing.T) {
 	defer server.Close()
 
 	c := NewController(server.URL, "token")
-	c.jupyterClientMap[session1] = &jupyterKernel{language: lang}
-	c.jupyterClientMap[session2] = &jupyterKernel{language: lang}
-	c.defaultLanguageJupyterSessions[lang] = session2
+	c.jupyterClientMap.Store(session1, &jupyterKernel{language: lang})
+	c.jupyterClientMap.Store(session2, &jupyterKernel{language: lang})
+	c.defaultLanguageSessions.Store(lang, session2)
 
 	require.NoError(t, c.DeleteLanguageContext(lang))
 
-	_, ok := c.jupyterClientMap[session1]
+	_, ok := c.jupyterClientMap.Load(session1)
 	require.False(t, ok, "expected session1 removed from cache")
-	_, ok = c.jupyterClientMap[session2]
+	_, ok = c.jupyterClientMap.Load(session2)
 	require.False(t, ok, "expected session2 removed from cache")
-	_, ok = c.defaultLanguageJupyterSessions[lang]
+	_, ok = c.defaultLanguageSessions.Load(lang)
 	require.False(t, ok, "expected default entry removed")
 	require.Equal(t, 1, deleteCalls[session1])
 	require.Equal(t, 1, deleteCalls[session2])

components/execd/pkg/runtime/ctrl.go

@@ -35,14 +35,15 @@ var kernelWaitingBackoff = wait.Backoff{
 
 // Controller manages code execution across runtimes.
 type Controller struct {
-	baseURL                        string
-	token                          string
-	mu                             sync.RWMutex
-	jupyterClientMap               map[string]*jupyterKernel
-	defaultLanguageJupyterSessions map[Language]string
-	commandClientMap               map[string]*commandKernel
-	db                             *sql.DB
-	dbOnce                         sync.Once
+	baseURL                 string
+	token                   string
+	mu                      sync.RWMutex
+	jupyterClientMap        sync.Map // map[sessionID]*jupyterKernel
+	defaultLanguageSessions sync.Map // map[Language]string
+	commandClientMap        sync.Map // map[sessionID]*commandKernel
+	bashSessionClientMap    sync.Map // map[sessionID]*bashSession
+	db                      *sql.DB
+	dbOnce                  sync.Once
 }
 
 type jupyterKernel struct {
@@ -70,10 +71,6 @@ func NewController(baseURL, token string) *Controller {
 	return &Controller{
 		baseURL: baseURL,
 		token:   token,
-
-		jupyterClientMap:               make(map[string]*jupyterKernel),
-		defaultLanguageJupyterSessions: make(map[Language]string),
-		commandClientMap:               make(map[string]*commandKernel),
 	}
 }
 

components/execd/pkg/runtime/interrupt.go

@@ -38,6 +38,8 @@ func (c *Controller) Interrupt(sessionID string) error {
 	case c.getCommandKernel(sessionID) != nil:
 		kernel := c.getCommandKernel(sessionID)
 		return c.killPid(kernel.pid)
+	case c.getBashSession(sessionID) != nil:
+		return c.closeBashSession(sessionID)
 	default:
 		return errors.New("no such session")
 	}