From 0ec3efa812ede11a68338f0e06b6fa03d519c17d Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 3 Jul 2026 16:36:35 +0800 Subject: [PATCH] feat(shell): add parallel bash shell subsystem auto-started per workspace Co-Authored-By: Claude --- README.md | 18 ++ cmd/server/main.go | 7 +- configs/config.yaml | 8 +- internal/api/process_ws_test.go | 13 +- internal/api/router.go | 9 +- internal/api/router_test.go | 7 +- internal/api/shell_handler.go | 157 +++++++++++++++ internal/model/shell.go | 8 + internal/service/shell_service.go | 98 +++++++++ internal/service/workspace_service.go | 15 +- internal/shell/manager.go | 279 ++++++++++++++++++++++++++ internal/shell/session.go | 39 ++++ internal/shell/subscription.go | 66 ++++++ pkg/config/config.go | 18 ++ 14 files changed, 730 insertions(+), 12 deletions(-) create mode 100644 internal/api/shell_handler.go create mode 100644 internal/model/shell.go create mode 100644 internal/service/shell_service.go create mode 100644 internal/shell/manager.go create mode 100644 internal/shell/session.go create mode 100644 internal/shell/subscription.go diff --git a/README.md b/README.md index b7700da..1e6b2cf 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,12 @@ workspace: root: "./workspaces" process: opencodeCommand: "opencode" + +shell: + command: "bash" + args: + - "-i" + file: maxWriteBytes: 1048576 gin: @@ -60,6 +66,8 @@ log: | `CODESPACE_ADDR` | Listen address | `:8080` | | `CODESPACE_WORKSPACE_ROOT` | Workspace storage root | `./workspaces` | | `CODESPACE_OPENCODE_COMMAND` | OpenCode binary path | `opencode` | +| `CODESPACE_SHELL_COMMAND` | Shell binary path | `bash` | +| `CODESPACE_SHELL_ARGS` | Comma-separated shell arguments | `-i` | | `CODESPACE_FILE_MAX_WRITE_BYTES` | Max bytes per file write | `1048576` | | `CODESPACE_GIN_MODE` | Gin mode (`debug`, `release`, `test`) | `release` | | `CODESPACE_READ_TIMEOUT` | HTTP read timeout | `15s` | @@ -107,6 +115,16 @@ All runtime logging uses slog; `fmt`, `log.Printf`, and the Gin default logger a | `GET` | `/api/workspaces/:id/process/status` | Get process running status and PID | | `GET` | `/api/workspaces/:id/process/ws` | WebSocket: stream process stdout/stderr, send stdin (text frames, raw bytes) | +### Shell + +| Method | Path | Description | +|---|---|---| +| `POST` | `/api/workspaces/:id/shell/start` | Start interactive shell for workspace | +| `POST` | `/api/workspaces/:id/shell/stop` | Stop interactive shell | +| `POST` | `/api/workspaces/:id/shell/restart` | Restart interactive shell | +| `GET` | `/api/workspaces/:id/shell/status` | Get shell running status and PID | +| `GET` | `/api/workspaces/:id/shell/ws` | WebSocket: stream shell stdout/stderr, send stdin (text frames, raw bytes) | + ### Health | Method | Path | Description | diff --git a/cmd/server/main.go b/cmd/server/main.go index 3d21b2b..bcbb2c8 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -12,6 +12,7 @@ import ( "codespace/internal/api" "codespace/internal/process" "codespace/internal/service" + "codespace/internal/shell" "codespace/internal/workspace" "codespace/pkg/config" "codespace/pkg/logger" @@ -34,12 +35,14 @@ func main() { workspaces := workspace.NewLocalManager(cfg.Workspace.Root) processes := process.NewManager(cfg.Process.OpenCodeCommand) + shellMgr := shell.NewManager(cfg.Shell.Command, cfg.Shell.Args) - workspaceSvc := service.NewWorkspaceService(workspaces, processes, lg) + workspaceSvc := service.NewWorkspaceService(workspaces, processes, shellMgr, lg) fileSvc := service.NewFileService(workspaces, cfg.File.MaxWriteBytes) processSvc := service.NewProcessService(workspaces, processes, lg) + shellSvc := service.NewShellService(workspaces, shellMgr, lg) - router := api.NewRouter(workspaceSvc, fileSvc, processSvc, lg, cfg.Gin.Mode) + router := api.NewRouter(workspaceSvc, fileSvc, processSvc, shellSvc, lg, cfg.Gin.Mode) server := &http.Server{ Addr: cfg.Server.Addr, diff --git a/configs/config.yaml b/configs/config.yaml index d2ed34c..43b7d7e 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -8,10 +8,16 @@ workspace: root: "./workspaces" process: opencodeCommand: "opencode" + +shell: + command: "bash" + args: + - "-i" + file: maxWriteBytes: 1048576 gin: - mode: "release" + mode: "debug" log: level: "info" format: "json" diff --git a/internal/api/process_ws_test.go b/internal/api/process_ws_test.go index a0d317f..02cd469 100644 --- a/internal/api/process_ws_test.go +++ b/internal/api/process_ws_test.go @@ -11,6 +11,7 @@ import ( "codespace/internal/process" "codespace/internal/service" + "codespace/internal/shell" "codespace/internal/workspace" "github.com/gin-gonic/gin" @@ -29,11 +30,13 @@ func TestProcessWS(t *testing.T) { wsRoot := filepath.Join(tmpDir, "workspaces") wsMgr := workspace.NewLocalManager(wsRoot) procMgr := process.NewManager(opencodePath) - wsSvc := service.NewWorkspaceService(wsMgr, procMgr, nil) + shellMgr := shell.NewManager("bash", []string{"-i"}) + wsSvc := service.NewWorkspaceService(wsMgr, procMgr, shellMgr, nil) fileSvc := service.NewFileService(wsMgr, 1<<20) procSvc := service.NewProcessService(wsMgr, procMgr, nil) + shellSvc := service.NewShellService(wsMgr, shellMgr, nil) - r := NewRouter(wsSvc, fileSvc, procSvc, nil, gin.TestMode) + r := NewRouter(wsSvc, fileSvc, procSvc, shellSvc, nil, gin.TestMode) srv := httptest.NewServer(r) defer srv.Close() @@ -102,11 +105,13 @@ func TestProcessWSMultiSubscriber(t *testing.T) { wsRoot := filepath.Join(tmpDir, "workspaces") wsMgr := workspace.NewLocalManager(wsRoot) procMgr := process.NewManager(opencodePath) - wsSvc := service.NewWorkspaceService(wsMgr, procMgr, nil) + shellMgr := shell.NewManager("bash", []string{"-i"}) + wsSvc := service.NewWorkspaceService(wsMgr, procMgr, shellMgr, nil) fileSvc := service.NewFileService(wsMgr, 1<<20) procSvc := service.NewProcessService(wsMgr, procMgr, nil) + shellSvc := service.NewShellService(wsMgr, shellMgr, nil) - r := NewRouter(wsSvc, fileSvc, procSvc, nil, gin.TestMode) + r := NewRouter(wsSvc, fileSvc, procSvc, shellSvc, nil, gin.TestMode) srv := httptest.NewServer(r) defer srv.Close() diff --git a/internal/api/router.go b/internal/api/router.go index 6ba9183..36a275e 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -10,7 +10,7 @@ import ( ) // NewRouter builds a Gin engine with all API routes registered. -func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, processes *service.ProcessService, lg *slog.Logger, ginMode string) *gin.Engine { +func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, processes *service.ProcessService, shells *service.ShellService, lg *slog.Logger, ginMode string) *gin.Engine { gin.SetMode(ginMode) r := gin.New() r.Use(gin.Recovery()) @@ -19,6 +19,7 @@ func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, wsHandler := &workspaceHandler{svc: workspaces} fileHandler := &fileHandler{svc: files} procHandler := &processHandler{svc: processes} + shellHandler := &shellHandler{svc: shells} r.GET("/healthz", healthHandler) @@ -42,6 +43,12 @@ func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, api.GET("/workspaces/:id/process/status", procHandler.status) api.GET("/workspaces/:id/process/ws", procHandler.ws) + api.POST("/workspaces/:id/shell/start", shellHandler.start) + api.POST("/workspaces/:id/shell/stop", shellHandler.stop) + api.POST("/workspaces/:id/shell/restart", shellHandler.restart) + api.GET("/workspaces/:id/shell/status", shellHandler.status) + api.GET("/workspaces/:id/shell/ws", shellHandler.ws) + return r } diff --git a/internal/api/router_test.go b/internal/api/router_test.go index a5fce53..75fccf9 100644 --- a/internal/api/router_test.go +++ b/internal/api/router_test.go @@ -9,6 +9,7 @@ import ( "codespace/internal/process" "codespace/internal/service" + "codespace/internal/shell" "codespace/internal/workspace" "github.com/gin-gonic/gin" @@ -24,10 +25,12 @@ func setupTestRouterWithMaxWriteBytes(t *testing.T, maxWriteBytes int64) http.Ha root := t.TempDir() wsMgr := workspace.NewLocalManager(root) procMgr := process.NewManager("") - wsSvc := service.NewWorkspaceService(wsMgr, procMgr, nil) + shellMgr := shell.NewManager("bash", []string{"-i"}) + wsSvc := service.NewWorkspaceService(wsMgr, procMgr, shellMgr, nil) fileSvc := service.NewFileService(wsMgr, maxWriteBytes) procSvc := service.NewProcessService(wsMgr, procMgr, nil) - return NewRouter(wsSvc, fileSvc, procSvc, nil, gin.TestMode) + shellSvc := service.NewShellService(wsMgr, shellMgr, nil) + return NewRouter(wsSvc, fileSvc, procSvc, shellSvc, nil, gin.TestMode) } func createWorkspaceForTest(t *testing.T, router http.Handler, id string) { diff --git a/internal/api/shell_handler.go b/internal/api/shell_handler.go new file mode 100644 index 0000000..760181b --- /dev/null +++ b/internal/api/shell_handler.go @@ -0,0 +1,157 @@ +package api + +import ( + "fmt" + "net/http" + "sync" + "time" + + "codespace/internal/model" + "codespace/internal/service" + "codespace/internal/shell" + "codespace/internal/util" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" +) + +type shellHandler struct { + svc *service.ShellService +} + +func shellExitBanner(exit shell.ExitInfo) string { + if exit.Signal != "" { + return fmt.Sprintf("\r\n[process exited: signal %s]\r\n", exit.Signal) + } + return fmt.Sprintf("\r\n[process exited with code %d]\r\n", exit.Code) +} + +func (h *shellHandler) start(c *gin.Context) { + id := c.Param("id") + if err := h.svc.Start(id); err != nil { + writeError(c, err) + return + } + c.Status(http.StatusNoContent) +} + +func (h *shellHandler) stop(c *gin.Context) { + id := c.Param("id") + if err := h.svc.Stop(id); err != nil { + writeError(c, err) + return + } + c.Status(http.StatusNoContent) +} + +func (h *shellHandler) restart(c *gin.Context) { + id := c.Param("id") + if err := h.svc.Restart(id); err != nil { + writeError(c, err) + return + } + c.Status(http.StatusNoContent) +} + +func (h *shellHandler) status(c *gin.Context) { + id := c.Param("id") + status, err := h.svc.Status(id) + if err != nil { + writeError(c, err) + return + } + c.JSON(http.StatusOK, model.ShellStatusResponse{ + WorkspaceID: status.WorkspaceID, + Running: status.Running, + PID: status.PID, + }) +} + +func (h *shellHandler) ws(c *gin.Context) { + id := c.Param("id") + + stdin, err := h.svc.Input(id) + if err != nil { + if util.CodeOf(err) == util.CodeNotFound { + c.JSON(409, gin.H{"error": "shell not running"}) + return + } + writeError(c, err) + return + } + + sub, err := h.svc.Subscribe(id) + if err != nil { + writeError(c, err) + return + } + + exit, _ := h.svc.ExitStatus(id) + + upgrader := websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { return true }, + ReadBufferSize: 4096, + WriteBufferSize: 4096, + } + + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + return + } + + done := make(chan struct{}) + var closeOnce sync.Once + closeAll := func() { + closeOnce.Do(func() { + sub.Close() + conn.Close() + stdin.Close() + close(done) + }) + } + + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil { + closeAll() + return + } + case chunk, ok := <-sub.Output(): + if !ok { + banner := shellExitBanner(exit) + conn.WriteMessage(websocket.TextMessage, []byte(banner)) + conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) + closeAll() + return + } + if err := conn.WriteMessage(websocket.TextMessage, chunk); err != nil { + closeAll() + return + } + } + } + }() + + conn.SetReadLimit(1 << 20) + conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + conn.SetPongHandler(func(string) error { + conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + return nil + }) + for { + mt, data, err := conn.ReadMessage() + if err != nil { + break + } + if mt == websocket.TextMessage { + stdin.Write(data) + } + } + closeAll() +} diff --git a/internal/model/shell.go b/internal/model/shell.go new file mode 100644 index 0000000..7ac0eaf --- /dev/null +++ b/internal/model/shell.go @@ -0,0 +1,8 @@ +package model + +// ShellStatusResponse is the API response for shell status. +type ShellStatusResponse struct { + WorkspaceID string `json:"workspaceId"` + Running bool `json:"running"` + PID int `json:"pid,omitempty"` +} diff --git a/internal/service/shell_service.go b/internal/service/shell_service.go new file mode 100644 index 0000000..cd9cc17 --- /dev/null +++ b/internal/service/shell_service.go @@ -0,0 +1,98 @@ +package service + +import ( + "log/slog" + + "io" + + "codespace/internal/shell" + "codespace/internal/workspace" +) + +// ShellService manages interactive shells for workspaces. +type ShellService struct { + workspaces workspace.Manager + shells shell.Manager + logger *slog.Logger +} + +// NewShellService creates a ShellService. +// If lg is nil, slog.Default() is used. +func NewShellService(workspaces workspace.Manager, shells shell.Manager, lg *slog.Logger) *ShellService { + if lg == nil { + lg = slog.Default() + } + return &ShellService{workspaces: workspaces, shells: shells, logger: lg} +} + +// Start starts an interactive shell for the given workspace. +func (s *ShellService) Start(workspaceID string) error { + ws, err := s.workspaces.Get(workspaceID) + if err != nil { + return err + } + if err := s.shells.Start(workspaceID, ws.Root); err != nil { + s.logger.Error("shell start failed", "workspace_id", workspaceID, "error", err) + return err + } + status := s.shells.Status(workspaceID) + s.logger.Info("shell started", "workspace_id", workspaceID, "pid", status.PID) + return nil +} + +// Stop stops the interactive shell for the given workspace. +func (s *ShellService) Stop(workspaceID string) error { + if err := s.shells.Stop(workspaceID); err != nil { + s.logger.Error("shell stop failed", "workspace_id", workspaceID, "error", err) + return err + } + s.logger.Info("shell stopped", "workspace_id", workspaceID) + return nil +} + +// Restart restarts the interactive shell for the given workspace. +func (s *ShellService) Restart(workspaceID string) error { + ws, err := s.workspaces.Get(workspaceID) + if err != nil { + return err + } + if err := s.shells.Restart(workspaceID, ws.Root); err != nil { + s.logger.Error("shell restart failed", "workspace_id", workspaceID, "error", err) + return err + } + status := s.shells.Status(workspaceID) + s.logger.Info("shell restarted", "workspace_id", workspaceID, "pid", status.PID) + return nil +} + +// Status returns the shell status for the given workspace. +func (s *ShellService) Status(workspaceID string) (shell.Status, error) { + if _, err := s.workspaces.Get(workspaceID); err != nil { + return shell.Status{}, err + } + return s.shells.Status(workspaceID), nil +} + +// Subscribe subscribes to output events for the workspace shell. +func (s *ShellService) Subscribe(workspaceID string) (shell.Subscription, error) { + if _, err := s.workspaces.Get(workspaceID); err != nil { + return nil, err + } + return s.shells.Subscribe(workspaceID) +} + +// Input returns the stdin writer for the workspace shell. +func (s *ShellService) Input(workspaceID string) (io.WriteCloser, error) { + if _, err := s.workspaces.Get(workspaceID); err != nil { + return nil, err + } + return s.shells.Stdin(workspaceID) +} + +// ExitStatus returns the exit status for the workspace shell. +func (s *ShellService) ExitStatus(workspaceID string) (shell.ExitInfo, error) { + if _, err := s.workspaces.Get(workspaceID); err != nil { + return shell.ExitInfo{}, err + } + return s.shells.ExitStatus(workspaceID) +} diff --git a/internal/service/workspace_service.go b/internal/service/workspace_service.go index 810b5ad..a369ba9 100644 --- a/internal/service/workspace_service.go +++ b/internal/service/workspace_service.go @@ -4,6 +4,7 @@ import ( "log/slog" "codespace/internal/process" + "codespace/internal/shell" "codespace/internal/workspace" ) @@ -11,16 +12,17 @@ import ( type WorkspaceService struct { workspaces workspace.Manager processes process.Manager + shells shell.Manager logger *slog.Logger } // NewWorkspaceService creates a WorkspaceService. // If lg is nil, slog.Default() is used. -func NewWorkspaceService(workspaces workspace.Manager, processes process.Manager, lg *slog.Logger) *WorkspaceService { +func NewWorkspaceService(workspaces workspace.Manager, processes process.Manager, shells shell.Manager, lg *slog.Logger) *WorkspaceService { if lg == nil { lg = slog.Default() } - return &WorkspaceService{workspaces: workspaces, processes: processes, logger: lg} + return &WorkspaceService{workspaces: workspaces, processes: processes, shells: shells, logger: lg} } // Create creates a new workspace. @@ -30,6 +32,11 @@ func (s *WorkspaceService) Create(id string) (*workspace.Workspace, error) { return nil, err } s.logger.Info("workspace created", "workspace_id", id) + + if err := s.shells.Start(ws.ID, ws.Root); err != nil { + s.logger.Warn("failed to auto-start shell for workspace", "workspace_id", id, "error", err) + } + return ws, nil } @@ -45,6 +52,10 @@ func (s *WorkspaceService) List() ([]workspace.Workspace, error) { // Delete stops any running process for the workspace, then removes it. func (s *WorkspaceService) Delete(id string) error { + if err := s.shells.Stop(id); err != nil { + s.logger.Warn("failed to stop workspace shell before delete", "workspace_id", id, "error", err) + } + status := s.processes.Status(id) if status.Running { if err := s.processes.Stop(id); err != nil { diff --git a/internal/shell/manager.go b/internal/shell/manager.go new file mode 100644 index 0000000..2cf6da2 --- /dev/null +++ b/internal/shell/manager.go @@ -0,0 +1,279 @@ +package shell + +import ( + "fmt" + "io" + "os" + "os/exec" + "sync" + "syscall" + + "codespace/internal/util" +) + +const ( + outputBufferSize = 4 * 1024 +) + +// DefaultShellCommand is the default command used to launch a shell. +const DefaultShellCommand = "bash" + +// Manager manages interactive shells per workspace. +type Manager interface { + Start(workspaceID string, workspaceRoot string) error + Stop(workspaceID string) error + Restart(workspaceID string, workspaceRoot string) error + Status(workspaceID string) Status + Subscribe(workspaceID string) (Subscription, error) + Stdin(workspaceID string) (io.WriteCloser, error) + ExitStatus(workspaceID string) (ExitInfo, error) +} + +// LocalManager implements Manager using local OS processes. +type LocalManager struct { + command string + args []string + mu sync.Mutex + sessions map[string]*Session +} + +// NewManager creates a LocalManager with the given command and arguments. +// If command is empty, it defaults to "bash". +func NewManager(command string, args []string) *LocalManager { + return &LocalManager{ + command: normalizeCommand(command), + args: args, + sessions: make(map[string]*Session), + } +} + +func normalizeCommand(command string) string { + if command == "" { + return DefaultShellCommand + } + return command +} + +// Start launches a shell for the given workspace. +// Returns CodeConflict if a session is already running. +func (m *LocalManager) Start(workspaceID string, workspaceRoot string) error { + m.mu.Lock() + defer m.mu.Unlock() + + if sess, ok := m.sessions[workspaceID]; ok && sess.Cmd.Process != nil { + if sess.Cmd.ProcessState == nil { + return util.New(util.CodeConflict, "shell already running") + } + } + + cmd := exec.Command(m.command, m.args...) + cmd.Dir = workspaceRoot + cmd.Env = append(os.Environ(), fmt.Sprintf("HOME=%s", workspaceRoot)) + + stdinR, stdinW, err := os.Pipe() + if err != nil { + return util.Wrap(util.CodeInternal, "failed to create stdin pipe", err) + } + + stdoutR, stdoutW, err := os.Pipe() + if err != nil { + stdinR.Close() + stdinW.Close() + return util.Wrap(util.CodeInternal, "failed to create stdout pipe", err) + } + + cmd.Stdin = stdinR + cmd.Stdout = stdoutW + cmd.Stderr = stdoutW + + if err := cmd.Start(); err != nil { + stdinR.Close() + stdinW.Close() + stdoutR.Close() + stdoutW.Close() + return util.Wrap(util.CodeInternal, "failed to start shell", err) + } + + stdinR.Close() + stdoutW.Close() + + sess := &Session{ + WorkspaceID: workspaceID, + Root: workspaceRoot, + Cmd: cmd, + Stdin: stdinW, + Subscribers: make(map[Subscription]struct{}), + } + m.sessions[workspaceID] = sess + + outputDone := make(chan struct{}) + go m.captureOutput(sess, stdoutR, outputDone) + go m.waitExit(sess, outputDone) + + return nil +} + +// captureOutput reads from the merged stdout/stderr pipe and fans out each +// chunk to all subscribers using non-blocking sends. +func (m *LocalManager) captureOutput(sess *Session, stdoutR *os.File, done chan<- struct{}) { + defer close(done) + defer stdoutR.Close() + + buf := make([]byte, outputBufferSize) + + for { + n, err := stdoutR.Read(buf) + if n > 0 { + chunk := make([]byte, n) + copy(chunk, buf[:n]) + + sess.mu.Lock() + for sub := range sess.Subscribers { + sub.(*subscription).send(chunk) + } + sess.mu.Unlock() + } + if err != nil { + if err != io.EOF { + // Ignore read errors; the pipe is closing. + } + break + } + } +} + +// waitExit waits for the shell to exit, then records the exit status, closes +// the stdin writer, and closes every subscriber channel exactly once. +func (m *LocalManager) waitExit(sess *Session, outputDone <-chan struct{}) { + _ = sess.Cmd.Wait() + <-outputDone + + sess.mu.Lock() + defer sess.mu.Unlock() + + if sess.Cmd.ProcessState != nil { + sess.Exit.Code = sess.Cmd.ProcessState.ExitCode() + if ws, ok := sess.Cmd.ProcessState.Sys().(syscall.WaitStatus); ok && ws.Signaled() { + sess.Exit.Signal = ws.Signal().String() + } + } + + if sess.Stdin != nil { + _ = sess.Stdin.Close() + } + + for sub := range sess.Subscribers { + sub.(*subscription).closeChan() + } +} + +// Stop kills the shell for the given workspace. +// Returns CodeNotFound if no session exists. +func (m *LocalManager) Stop(workspaceID string) error { + m.mu.Lock() + defer m.mu.Unlock() + + sess, ok := m.sessions[workspaceID] + if !ok || sess.Cmd.Process == nil { + return util.New(util.CodeNotFound, "no running shell for workspace") + } + + if err := sess.Cmd.Process.Kill(); err != nil { + return util.Wrap(util.CodeInternal, "failed to stop shell", err) + } + + delete(m.sessions, workspaceID) + return nil +} + +// Restart stops (if running) then starts the shell. +func (m *LocalManager) Restart(workspaceID string, workspaceRoot string) error { + m.mu.Lock() + if sess, ok := m.sessions[workspaceID]; ok && sess.Cmd.Process != nil { + if sess.Cmd.ProcessState == nil { + _ = sess.Cmd.Process.Kill() + delete(m.sessions, workspaceID) + } + } + m.mu.Unlock() + + return m.Start(workspaceID, workspaceRoot) +} + +// Status returns the current shell status for the workspace. +func (m *LocalManager) Status(workspaceID string) Status { + m.mu.Lock() + defer m.mu.Unlock() + + sess, ok := m.sessions[workspaceID] + if !ok || sess.Cmd.Process == nil { + return Status{WorkspaceID: workspaceID, Running: false} + } + + if sess.Cmd.ProcessState != nil { + return Status{WorkspaceID: workspaceID, Running: false} + } + + return Status{ + WorkspaceID: workspaceID, + Running: true, + PID: sess.Cmd.Process.Pid, + } +} + +// Subscribe creates a new output subscription for the workspace. +// Returns CodeNotFound if the workspace has no session. +func (m *LocalManager) Subscribe(workspaceID string) (Subscription, error) { + m.mu.Lock() + sess, ok := m.sessions[workspaceID] + m.mu.Unlock() + + if !ok { + return nil, util.New(util.CodeNotFound, "no running shell") + } + + sess.mu.Lock() + sub := newSubscription(sess) + sess.Subscribers[sub] = struct{}{} + if sess.Cmd.ProcessState != nil { + sub.closeChan() + } + sess.mu.Unlock() + + return sub, nil +} + +// Stdin returns the stdin writer for the workspace shell. +func (m *LocalManager) Stdin(workspaceID string) (io.WriteCloser, error) { + return m.stdinOf(workspaceID) +} + +// ExitStatus returns the exit status for the workspace shell. +func (m *LocalManager) ExitStatus(workspaceID string) (ExitInfo, error) { + return m.exitOf(workspaceID) +} + +// stdinOf returns the session's stdin writer or CodeNotFound. +func (m *LocalManager) stdinOf(workspaceID string) (io.WriteCloser, error) { + m.mu.Lock() + defer m.mu.Unlock() + sess, ok := m.sessions[workspaceID] + if !ok { + return nil, util.New(util.CodeNotFound, "no running shell") + } + return sess.Stdin, nil +} + +// exitOf returns the session's recorded exit info or CodeNotFound. +func (m *LocalManager) exitOf(workspaceID string) (ExitInfo, error) { + m.mu.Lock() + sess, ok := m.sessions[workspaceID] + m.mu.Unlock() + if !ok { + return ExitInfo{}, util.New(util.CodeNotFound, "no running shell") + } + + sess.mu.Lock() + defer sess.mu.Unlock() + return sess.Exit, nil +} diff --git a/internal/shell/session.go b/internal/shell/session.go new file mode 100644 index 0000000..fc8844d --- /dev/null +++ b/internal/shell/session.go @@ -0,0 +1,39 @@ +package shell + +import ( + "io" + "os/exec" + "sync" +) + +// Status represents the status of a shell for a workspace. +type Status struct { + WorkspaceID string `json:"workspaceId"` + Running bool `json:"running"` + PID int `json:"pid,omitempty"` +} + +// ExitInfo holds the exit reason of a shell. +type ExitInfo struct { + Code int + Signal string +} + +// Session holds the runtime state of a started shell. +type Session struct { + WorkspaceID string + Root string + Cmd *exec.Cmd + Stdin io.WriteCloser + Subscribers map[Subscription]struct{} + Exit ExitInfo + mu sync.Mutex +} + +// removeSubscription removes sub from s.Subscribers. It is safe to call when +// the subscription is no longer present. +func (s *Session) removeSubscription(sub *subscription) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.Subscribers, sub) +} diff --git a/internal/shell/subscription.go b/internal/shell/subscription.go new file mode 100644 index 0000000..0ebfee9 --- /dev/null +++ b/internal/shell/subscription.go @@ -0,0 +1,66 @@ +package shell + +import "sync" + +// Subscription is a consumer handle for a shell output stream. +type Subscription interface { + Output() <-chan []byte + Close() error +} + +// subscription is a buffered, closable fan-out channel for a Session. +type subscription struct { + session *Session + ch chan []byte + once sync.Once + mu sync.Mutex + closed bool +} + +func newSubscription(sess *Session) *subscription { + return &subscription{ + session: sess, + ch: make(chan []byte, 32), + } +} + +func (s *subscription) Output() <-chan []byte { + return s.ch +} + +func (s *subscription) Close() error { + s.once.Do(func() { + s.mu.Lock() + s.closed = true + close(s.ch) + s.mu.Unlock() + s.session.removeSubscription(s) + }) + return nil +} + +// closeChan closes the underlying channel exactly once without removing the +// subscription from the session's subscriber list. It is used by the manager +// when the shell exits. +func (s *subscription) closeChan() { + s.once.Do(func() { + s.mu.Lock() + s.closed = true + close(s.ch) + s.mu.Unlock() + }) +} + +// send performs a non-blocking send to the subscription channel. If the +// channel is full or already closed, the chunk is dropped. +func (s *subscription) send(chunk []byte) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + select { + case s.ch <- chunk: + default: + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 6465767..3b2c4e0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -4,6 +4,7 @@ import ( "fmt" "os" "strconv" + "strings" "time" "gopkg.in/yaml.v3" @@ -14,6 +15,7 @@ type Config struct { Server ServerConfig `yaml:"server"` Workspace WorkspaceConfig `yaml:"workspace"` Process ProcessConfig `yaml:"process"` + Shell ShellConfig `yaml:"shell"` File FileConfig `yaml:"file"` Gin GinConfig `yaml:"gin"` Log LogConfig `yaml:"log"` @@ -38,6 +40,12 @@ type ProcessConfig struct { OpenCodeCommand string `yaml:"opencodeCommand"` } +// ShellConfig holds interactive shell settings. +type ShellConfig struct { + Command string `yaml:"command"` + Args []string `yaml:"args"` +} + // FileConfig holds file operation limits. type FileConfig struct { MaxWriteBytes int64 `yaml:"maxWriteBytes"` @@ -60,6 +68,7 @@ type rawConfig struct { Server rawServerConfig `yaml:"server"` Workspace WorkspaceConfig `yaml:"workspace"` Process ProcessConfig `yaml:"process"` + Shell ShellConfig `yaml:"shell"` File FileConfig `yaml:"file"` Gin GinConfig `yaml:"gin"` Log LogConfig `yaml:"log"` @@ -85,6 +94,7 @@ func Default() Config { }, Workspace: WorkspaceConfig{Root: "./workspaces"}, Process: ProcessConfig{OpenCodeCommand: "opencode"}, + Shell: ShellConfig{Command: "bash", Args: []string{"-i"}}, File: FileConfig{MaxWriteBytes: 1 << 20}, Gin: GinConfig{Mode: "release"}, Log: LogConfig{Level: "info", Format: "json"}, @@ -140,6 +150,8 @@ func applyRaw(cfg *Config, raw rawConfig) error { if raw.Process.OpenCodeCommand != "" { cfg.Process.OpenCodeCommand = raw.Process.OpenCodeCommand } + cfg.Shell.Command = raw.Shell.Command + cfg.Shell.Args = raw.Shell.Args if raw.File.MaxWriteBytes != 0 { cfg.File.MaxWriteBytes = raw.File.MaxWriteBytes } @@ -199,6 +211,12 @@ func applyEnv(cfg *Config) error { if v := os.Getenv("CODESPACE_OPENCODE_COMMAND"); v != "" { cfg.Process.OpenCodeCommand = v } + if v := os.Getenv("CODESPACE_SHELL_COMMAND"); v != "" { + cfg.Shell.Command = v + } + if v := os.Getenv("CODESPACE_SHELL_ARGS"); v != "" { + cfg.Shell.Args = strings.Split(v, ",") + } if v := os.Getenv("CODESPACE_FILE_MAX_WRITE_BYTES"); v != "" { n, err := strconv.ParseInt(v, 10, 64) if err != nil {