diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index b18f376b99..74663b9a19 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -413,6 +413,13 @@ func main() { workspaceFileHandlers := agenthandlers.NewWorkspaceFileHandlers(lifecycleMgr, log) workspaceFileHandlers.RegisterHandlers(gateway.Dispatcher) log.Info("Registered Workspace File WebSocket handlers") + + // Register shell handlers + shellHandlers := agenthandlers.NewShellHandlers(lifecycleMgr, log) + shellHandlers.SetHub(gateway.Hub) + shellHandlers.RegisterHandlers(gateway.Dispatcher) + lifecycleMgr.SetShellStreamStarter(shellHandlers) + log.Info("Registered Shell WebSocket handlers") } // Start the WebSocket hub diff --git a/apps/backend/go.mod b/apps/backend/go.mod index aae8826eee..0d4cf92198 100644 --- a/apps/backend/go.mod +++ b/apps/backend/go.mod @@ -6,11 +6,12 @@ toolchain go1.24.11 require ( github.com/coder/acp-go-sdk v0.6.3 + github.com/creack/pty v1.1.18 github.com/docker/docker v28.5.2+incompatible + github.com/fsnotify/fsnotify v1.6.0 github.com/gin-gonic/gin v1.9.1 github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 - github.com/jackc/pgx/v5 v5.8.0 github.com/mattn/go-sqlite3 v1.14.33 github.com/nats-io/nats.go v1.31.0 github.com/spf13/viper v1.16.0 @@ -31,7 +32,6 @@ require ( github.com/docker/go-connections v0.6.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect - github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/gabriel-vasile/mimetype v1.4.2 // indirect github.com/gin-contrib/sse v0.1.0 // indirect github.com/go-logr/logr v1.4.3 // indirect @@ -41,9 +41,6 @@ require ( github.com/go-playground/validator/v10 v10.14.0 // indirect github.com/goccy/go-json v0.10.2 // indirect github.com/hashicorp/hcl v1.0.0 // indirect - github.com/jackc/pgpassfile v1.0.0 // indirect - github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect - github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.17.0 // indirect github.com/klauspost/cpuid/v2 v2.2.4 // indirect @@ -82,7 +79,6 @@ require ( golang.org/x/arch v0.3.0 // indirect golang.org/x/crypto v0.44.0 // indirect golang.org/x/net v0.47.0 // indirect - golang.org/x/sync v0.18.0 // indirect golang.org/x/sys v0.39.0 // indirect golang.org/x/text v0.31.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect diff --git a/apps/backend/go.sum b/apps/backend/go.sum index 9244644b2e..99ef3d9625 100644 --- a/apps/backend/go.sum +++ b/apps/backend/go.sum @@ -68,6 +68,8 @@ github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151X github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY= +github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -184,14 +186,6 @@ github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= -github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= -github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= -github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= -github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= -github.com/jackc/pgx/v5 v5.8.0 h1:TYPDoleBBme0xGSAX3/+NujXXtpZn9HBONkQC7IEZSo= -github.com/jackc/pgx/v5 v5.8.0/go.mod h1:QVeDInX2m9VyzvNeiCJVjCkNFqzsNb43204HshNSZKw= -github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= -github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU= @@ -424,8 +418,6 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= -golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= diff --git a/apps/backend/internal/agent/agentctl/client.go b/apps/backend/internal/agent/agentctl/client.go index 9c93b89af5..4ec5d4a916 100644 --- a/apps/backend/internal/agent/agentctl/client.go +++ b/apps/backend/internal/agent/agentctl/client.go @@ -26,6 +26,7 @@ type Client struct { permissionConn *websocket.Conn gitStatusConn *websocket.Conn fileChangesConn *websocket.Conn + shellConn *websocket.Conn mu sync.RWMutex } @@ -372,10 +373,146 @@ func (c *Client) CloseFileChangesStream() { } } -// Close closes all connections (ACP, permissions, git status, file changes) +// Close closes all connections (ACP, permissions, git status, file changes, shell) func (c *Client) Close() { c.CloseACPStream() c.ClosePermissionStream() c.CloseGitStatusStream() c.CloseFileChangesStream() + c.CloseShellStream() +} + +// ShellStatusResponse from agentctl shell status endpoint +type ShellStatusResponse struct { + Running bool `json:"running"` + Pid int `json:"pid"` + Shell string `json:"shell"` + Cwd string `json:"cwd"` + StartedAt string `json:"started_at"` +} + +// ShellStatus gets the status of the embedded shell session +func (c *Client) ShellStatus(ctx context.Context) (*ShellStatusResponse, error) { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/shell/status", nil) + if err != nil { + return nil, err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("shell status failed: %s", resp.Status) + } + + var result ShellStatusResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, err + } + + return &result, nil +} + +// ShellMessage represents a message to/from the shell +type ShellMessage struct { + Type string `json:"type"` // "input", "output", "ping", "pong", "exit" + Data string `json:"data,omitempty"` + Code int `json:"code,omitempty"` // For exit type +} + +// ShellBufferResponse is the response from the shell buffer endpoint +type ShellBufferResponse struct { + Data string `json:"data"` +} + +// ShellBuffer returns the buffered shell output +func (c *Client) ShellBuffer(ctx context.Context) (string, error) { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/shell/buffer", nil) + if err != nil { + return "", err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return "", err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("shell buffer failed: %s", resp.Status) + } + + var result ShellBufferResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + return result.Data, nil +} + +// StreamShell connects to the shell WebSocket stream and returns channels for I/O +func (c *Client) StreamShell(ctx context.Context) (<-chan ShellMessage, chan<- ShellMessage, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.shellConn != nil { + return nil, nil, fmt.Errorf("shell stream already connected") + } + + wsURL := "ws" + c.baseURL[4:] + "/api/v1/shell/stream" + conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) + if err != nil { + return nil, nil, fmt.Errorf("failed to connect shell stream: %w", err) + } + c.shellConn = conn + + outputCh := make(chan ShellMessage, 256) + inputCh := make(chan ShellMessage, 64) + + // Read from WebSocket + go func() { + defer close(outputCh) + for { + var msg ShellMessage + if err := conn.ReadJSON(&msg); err != nil { + if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Debug("shell stream read error", zap.Error(err)) + } + return + } + select { + case outputCh <- msg: + default: + // Channel full, skip + } + } + }() + + // Write to WebSocket + go func() { + for msg := range inputCh { + if err := conn.WriteJSON(msg); err != nil { + c.logger.Debug("shell stream write error", zap.Error(err)) + return + } + } + }() + + return outputCh, inputCh, nil +} + +// CloseShellStream closes the shell stream connection +func (c *Client) CloseShellStream() { + c.mu.Lock() + defer c.mu.Unlock() + + if c.shellConn != nil { + if err := c.shellConn.Close(); err != nil { + c.logger.Debug("failed to close shell stream", zap.Error(err)) + } + c.shellConn = nil + } } diff --git a/apps/backend/internal/agent/handlers/shell_handlers.go b/apps/backend/internal/agent/handlers/shell_handlers.go new file mode 100644 index 0000000000..786649bcf7 --- /dev/null +++ b/apps/backend/internal/agent/handlers/shell_handlers.go @@ -0,0 +1,308 @@ +// Package handlers provides WebSocket and HTTP handlers for agent operations. +package handlers + +import ( + "context" + "fmt" + "sync" + + "github.com/kandev/kandev/internal/agent/agentctl" + "github.com/kandev/kandev/internal/agent/lifecycle" + "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +// ShellHandlers provides WebSocket handlers for shell terminal operations +type ShellHandlers struct { + lifecycleMgr *lifecycle.Manager + logger *logger.Logger + hub ShellOutputBroadcaster + + // Track active shell streams per task + activeStreams map[string]context.CancelFunc + mu sync.RWMutex + + // Input channels for sending input to shell streams + inputChannels map[string]chan<- agentctl.ShellMessage + inputMu sync.RWMutex +} + +// NewShellHandlers creates a new ShellHandlers instance +func NewShellHandlers(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *ShellHandlers { + return &ShellHandlers{ + lifecycleMgr: lifecycleMgr, + logger: log.WithFields(zap.String("component", "shell_handlers")), + activeStreams: make(map[string]context.CancelFunc), + inputChannels: make(map[string]chan<- agentctl.ShellMessage), + } +} + +// SetHub sets the hub for broadcasting shell output +func (h *ShellHandlers) SetHub(hub ShellOutputBroadcaster) { + h.hub = hub +} + +// RegisterHandlers registers shell handlers with the WebSocket dispatcher +func (h *ShellHandlers) RegisterHandlers(d *ws.Dispatcher) { + d.RegisterFunc(ws.ActionShellStatus, h.wsShellStatus) + d.RegisterFunc(ws.ActionShellSubscribe, h.wsShellSubscribe) + d.RegisterFunc(ws.ActionShellInput, h.wsShellInput) +} + +// ShellStatusRequest for shell.status action +type ShellStatusRequest struct { + TaskID string `json:"task_id"` +} + +// ShellInputRequest for shell.input action +type ShellInputRequest struct { + TaskID string `json:"task_id"` + Data string `json:"data"` +} + +// wsShellStatus returns the status of a shell session for a task +func (h *ShellHandlers) wsShellStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req ShellStatusRequest + if err := msg.ParsePayload(&req); err != nil { + return nil, fmt.Errorf("invalid payload: %w", err) + } + + if req.TaskID == "" { + return nil, fmt.Errorf("task_id is required") + } + + // Get the agent instance for this task + instance, ok := h.lifecycleMgr.GetInstanceByTaskID(req.TaskID) + if !ok { + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "available": false, + "error": "no agent running for this task", + }) + } + + // Get shell status from agentctl + client := instance.GetAgentCtlClient() + if client == nil { + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "available": false, + "error": "agent client not available", + }) + } + + status, err := client.ShellStatus(ctx) + if err != nil { + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "available": false, + "error": err.Error(), + }) + } + + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "available": true, + "running": status.Running, + "pid": status.Pid, + "shell": status.Shell, + "cwd": status.Cwd, + "started_at": status.StartedAt, + }) +} + +// ShellSubscribeRequest for shell.subscribe action +type ShellSubscribeRequest struct { + TaskID string `json:"task_id"` +} + +// wsShellSubscribe subscribes to shell output for a task and starts the shell stream if needed +func (h *ShellHandlers) wsShellSubscribe(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req ShellSubscribeRequest + if err := msg.ParsePayload(&req); err != nil { + return nil, fmt.Errorf("invalid payload: %w", err) + } + + if req.TaskID == "" { + return nil, fmt.Errorf("task_id is required") + } + + // Verify the agent instance exists for this task + instance, ok := h.lifecycleMgr.GetInstanceByTaskID(req.TaskID) + if !ok { + return nil, fmt.Errorf("no agent running for task %s", req.TaskID) + } + + // Start the shell stream (idempotent - does nothing if already running) + if err := h.StartShellStream(ctx, req.TaskID); err != nil { + return nil, fmt.Errorf("failed to start shell stream: %w", err) + } + + // Get buffered output to include in response + // This ensures client gets current shell state without duplicate broadcasts + buffer := "" + if client := instance.GetAgentCtlClient(); client != nil { + if b, err := client.ShellBuffer(ctx); err == nil { + buffer = b + } + } + + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "task_id": req.TaskID, + "buffer": buffer, + }) +} + +// wsShellInput sends input to a shell session +func (h *ShellHandlers) wsShellInput(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req ShellInputRequest + if err := msg.ParsePayload(&req); err != nil { + return nil, fmt.Errorf("invalid payload: %w", err) + } + + if req.TaskID == "" { + return nil, fmt.Errorf("task_id is required") + } + + // Verify the agent instance exists for this task + if _, ok := h.lifecycleMgr.GetInstanceByTaskID(req.TaskID); !ok { + return nil, fmt.Errorf("no agent running for task %s", req.TaskID) + } + + // Send input via the shell stream + if err := h.SendShellInput(req.TaskID, req.Data); err != nil { + return nil, err + } + + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) +} + +// StartShellStream starts streaming shell output for a task using the configured hub. +// This implements the lifecycle.ShellStreamStarter interface. +func (h *ShellHandlers) StartShellStream(ctx context.Context, taskID string) error { + if h.hub == nil { + return fmt.Errorf("hub not configured") + } + return h.StartShellStreamWithHub(ctx, taskID, h.hub) +} + +// StartShellStreamWithHub starts streaming shell output for a task to a specific hub +func (h *ShellHandlers) StartShellStreamWithHub(ctx context.Context, taskID string, hub ShellOutputBroadcaster) error { + h.mu.Lock() + if _, exists := h.activeStreams[taskID]; exists { + h.mu.Unlock() + return nil // Already streaming + } + + streamCtx, cancel := context.WithCancel(ctx) + h.activeStreams[taskID] = cancel + h.mu.Unlock() + + go h.runShellStream(streamCtx, taskID, hub) + return nil +} + +// StopShellStream stops the shell stream for a task +func (h *ShellHandlers) StopShellStream(taskID string) { + h.mu.Lock() + if cancel, exists := h.activeStreams[taskID]; exists { + cancel() + delete(h.activeStreams, taskID) + } + h.mu.Unlock() +} + +// ShellOutputBroadcaster interface for broadcasting shell output +type ShellOutputBroadcaster interface { + BroadcastToTask(taskID string, msg *ws.Message) +} + +// runShellStream runs the shell output stream for a task +func (h *ShellHandlers) runShellStream(ctx context.Context, taskID string, hub ShellOutputBroadcaster) { + defer func() { + h.mu.Lock() + delete(h.activeStreams, taskID) + h.mu.Unlock() + }() + + instance, ok := h.lifecycleMgr.GetInstanceByTaskID(taskID) + if !ok { + h.logger.Debug("no agent instance for shell stream", zap.String("task_id", taskID)) + return + } + + client := instance.GetAgentCtlClient() + if client == nil { + h.logger.Debug("no client for shell stream", zap.String("task_id", taskID)) + return + } + + outputCh, inputCh, err := client.StreamShell(ctx) + if err != nil { + h.logger.Error("failed to start shell stream", zap.String("task_id", taskID), zap.Error(err)) + return + } + + // Store input channel for sending input + h.storeInputChannel(taskID, inputCh) + defer h.removeInputChannel(taskID) + + h.logger.Info("shell stream started", zap.String("task_id", taskID)) + + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-outputCh: + if !ok { + return + } + // Broadcast shell output to task subscribers + notification, err := ws.NewNotification(ws.ActionShellOutput, map[string]interface{}{ + "task_id": taskID, + "type": msg.Type, + "data": msg.Data, + "code": msg.Code, + }) + if err != nil { + h.logger.Debug("failed to create shell output notification", zap.Error(err)) + continue + } + hub.BroadcastToTask(taskID, notification) + } + } +} + +func (h *ShellHandlers) storeInputChannel(taskID string, ch chan<- agentctl.ShellMessage) { + h.inputMu.Lock() + h.inputChannels[taskID] = ch + h.inputMu.Unlock() +} + +func (h *ShellHandlers) removeInputChannel(taskID string) { + h.inputMu.Lock() + if ch, exists := h.inputChannels[taskID]; exists { + close(ch) + delete(h.inputChannels, taskID) + } + h.inputMu.Unlock() +} + +// SendShellInput sends input to an active shell stream +func (h *ShellHandlers) SendShellInput(taskID, data string) error { + h.inputMu.RLock() + ch, exists := h.inputChannels[taskID] + h.inputMu.RUnlock() + + if !exists { + return fmt.Errorf("no active shell stream for task %s", taskID) + } + + select { + case ch <- agentctl.ShellMessage{Type: "input", Data: data}: + return nil + default: + return fmt.Errorf("shell input channel full") + } +} + diff --git a/apps/backend/internal/agent/handlers/shell_handlers_test.go b/apps/backend/internal/agent/handlers/shell_handlers_test.go new file mode 100644 index 0000000000..342919b6fc --- /dev/null +++ b/apps/backend/internal/agent/handlers/shell_handlers_test.go @@ -0,0 +1,307 @@ +package handlers + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/kandev/kandev/internal/agent/lifecycle" + "github.com/kandev/kandev/internal/agent/registry" + "github.com/kandev/kandev/internal/common/config" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// MockEventBus implements bus.EventBus for testing +type MockEventBus struct { + PublishedEvents []*bus.Event +} + +func (m *MockEventBus) Publish(ctx context.Context, subject string, event *bus.Event) error { + m.PublishedEvents = append(m.PublishedEvents, event) + return nil +} + +func (m *MockEventBus) Subscribe(subject string, handler bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} + +func (m *MockEventBus) QueueSubscribe(subject, queue string, handler bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} + +func (m *MockEventBus) Request(ctx context.Context, subject string, event *bus.Event, timeout time.Duration) (*bus.Event, error) { + return nil, nil +} + +func (m *MockEventBus) Close() {} + +func (m *MockEventBus) IsConnected() bool { + return true +} + +// testAgentConfig returns a default AgentConfig for testing +func testAgentConfig() config.AgentConfig { + return config.AgentConfig{ + Runtime: "docker", + StandaloneHost: "localhost", + StandalonePort: 9999, + } +} + +func newTestRegistry() *registry.Registry { + log := newTestLogger() + reg := registry.NewRegistry(log) + reg.LoadDefaults() + return reg +} + +func newTestLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{ + Level: "error", + Format: "json", + }) + return log +} + +func TestNewShellHandlers(t *testing.T) { + log := newTestLogger() + + // NewShellHandlers accepts *lifecycle.Manager, but we can pass nil for basic construction test + handlers := NewShellHandlers(nil, log) + + if handlers == nil { + t.Fatal("expected non-nil handlers") + } + if handlers.lifecycleMgr != nil { + t.Error("expected nil lifecycleMgr when nil passed") + } + if handlers.logger == nil { + t.Error("expected non-nil logger") + } + if handlers.activeStreams == nil { + t.Error("expected non-nil activeStreams map") + } + if handlers.inputChannels == nil { + t.Error("expected non-nil inputChannels map") + } +} + +func TestRegisterHandlers(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + dispatcher := ws.NewDispatcher() + handlers.RegisterHandlers(dispatcher) + + // Check that shell.status is registered + if !dispatcher.HasHandler(ws.ActionShellStatus) { + t.Error("expected shell.status handler to be registered") + } + + // Check that shell.input is registered + if !dispatcher.HasHandler(ws.ActionShellInput) { + t.Error("expected shell.input handler to be registered") + } +} + +func TestWsShellStatus_InvalidPayload(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + // Create a message with invalid payload + msg := &ws.Message{ + ID: "test-1", + Action: ws.ActionShellStatus, + Payload: json.RawMessage(`{invalid json`), + } + + _, err := handlers.wsShellStatus(context.Background(), msg) + if err == nil { + t.Error("expected error for invalid payload") + } +} + +func TestWsShellStatus_MissingTaskID(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + // Create a message with empty task_id + msg, _ := ws.NewRequest("test-1", ws.ActionShellStatus, ShellStatusRequest{TaskID: ""}) + + _, err := handlers.wsShellStatus(context.Background(), msg) + if err == nil { + t.Error("expected error for missing task_id") + } + if err.Error() != "task_id is required" { + t.Errorf("expected 'task_id is required' error, got: %v", err) + } +} + +func TestWsShellInput_InvalidPayload(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + msg := &ws.Message{ + ID: "test-1", + Action: ws.ActionShellInput, + Payload: json.RawMessage(`{invalid json`), + } + + _, err := handlers.wsShellInput(context.Background(), msg) + if err == nil { + t.Error("expected error for invalid payload") + } +} + +func TestWsShellInput_MissingTaskID(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + msg, _ := ws.NewRequest("test-1", ws.ActionShellInput, ShellInputRequest{TaskID: "", Data: "test"}) + + _, err := handlers.wsShellInput(context.Background(), msg) + if err == nil { + t.Error("expected error for missing task_id") + } + if err.Error() != "task_id is required" { + t.Errorf("expected 'task_id is required' error, got: %v", err) + } +} + +// newTestManager creates a lifecycle.Manager for testing +func newTestManager() *lifecycle.Manager { + log := newTestLogger() + reg := newTestRegistry() + eventBus := &MockEventBus{} + return lifecycle.NewManager(nil, reg, eventBus, testAgentConfig(), log) +} + +func TestWsShellStatus_NoInstanceFound(t *testing.T) { + log := newTestLogger() + mgr := newTestManager() + handlers := NewShellHandlers(mgr, log) + + msg, _ := ws.NewRequest("test-1", ws.ActionShellStatus, ShellStatusRequest{TaskID: "non-existent-task"}) + + resp, err := handlers.wsShellStatus(context.Background(), msg) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Should return a response indicating no agent available + var payload map[string]interface{} + if err := resp.ParsePayload(&payload); err != nil { + t.Fatalf("failed to parse payload: %v", err) + } + + if payload["available"] != false { + t.Errorf("expected available=false, got %v", payload["available"]) + } + if payload["error"] != "no agent running for this task" { + t.Errorf("expected 'no agent running for this task', got %v", payload["error"]) + } +} + +func TestWsShellStatus_NoClientAvailable(t *testing.T) { + log := newTestLogger() + mgr := newTestManager() + handlers := NewShellHandlers(mgr, log) + + // When no instance exists for the task, handler returns "no agent running" + // Note: Testing the "agent client not available" path requires injecting an + // instance with nil agentctl client, which requires access to lifecycle.Manager + // internal maps. This test verifies the handler works correctly with the manager. + msg, _ := ws.NewRequest("test-1", ws.ActionShellStatus, ShellStatusRequest{TaskID: "test-task-id"}) + + resp, err := handlers.wsShellStatus(context.Background(), msg) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Since there's no instance, we expect the "no agent running" error + var payload map[string]interface{} + if err := resp.ParsePayload(&payload); err != nil { + t.Fatalf("failed to parse payload: %v", err) + } + + if payload["available"] != false { + t.Errorf("expected available=false, got %v", payload["available"]) + } +} + +func TestWsShellInput_NoInstanceFound(t *testing.T) { + log := newTestLogger() + mgr := newTestManager() + handlers := NewShellHandlers(mgr, log) + + msg, _ := ws.NewRequest("test-1", ws.ActionShellInput, ShellInputRequest{ + TaskID: "non-existent-task", + Data: "test input", + }) + + _, err := handlers.wsShellInput(context.Background(), msg) + if err == nil { + t.Error("expected error for non-existent task") + } + expectedErr := "no agent running for task non-existent-task" + if err.Error() != expectedErr { + t.Errorf("expected '%s', got: %v", expectedErr, err) + } +} + +func TestSendShellInput_NoActiveStream(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + err := handlers.SendShellInput("non-existent-task", "test input") + if err == nil { + t.Error("expected error for non-existent stream") + } + expectedErr := "no active shell stream for task non-existent-task" + if err.Error() != expectedErr { + t.Errorf("expected '%s', got: %v", expectedErr, err) + } +} + +func TestStopShellStream_NonExistent(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + // StopShellStream should not panic when called with non-existent taskID + handlers.StopShellStream("non-existent-task") + // No panic = success +} + +func TestNewShellHandlers_WithManager(t *testing.T) { + log := newTestLogger() + mgr := newTestManager() + + handlers := NewShellHandlers(mgr, log) + + if handlers == nil { + t.Fatal("expected non-nil handlers") + } + if handlers.lifecycleMgr != mgr { + t.Error("expected lifecycleMgr to be set to provided manager") + } +} + +func TestRegisterHandlers_DispatcherReceivesMessages(t *testing.T) { + log := newTestLogger() + handlers := NewShellHandlers(nil, log) + + dispatcher := ws.NewDispatcher() + handlers.RegisterHandlers(dispatcher) + + // Test that dispatcher can dispatch to shell.status + msg, _ := ws.NewRequest("test-1", ws.ActionShellStatus, ShellStatusRequest{TaskID: ""}) + _, err := dispatcher.Dispatch(context.Background(), msg) + + // Should get "task_id is required" error since we passed empty task ID + if err == nil { + t.Error("expected error from dispatcher") + } +} diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 95081d8b30..29e16ef9bf 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -118,6 +118,9 @@ type Manager struct { agentCfg config.AgentConfig standaloneCtl *agentctl.StandaloneCtl // Client for standalone agentctl control API + // Shell stream starter for auto-starting shell streams + shellStreamStarter ShellStreamStarter + // Track active instances instances map[string]*AgentInstance // by instance ID byTask map[string]string // taskID -> instanceID @@ -186,6 +189,16 @@ func (m *Manager) SetWorktreeManager(worktreeMgr *worktree.Manager) { m.worktreeMgr = worktreeMgr } +// ShellStreamStarter is called to start shell streaming for an agent instance +type ShellStreamStarter interface { + StartShellStream(ctx context.Context, taskID string) error +} + +// SetShellStreamStarter sets the shell stream starter +func (m *Manager) SetShellStreamStarter(starter ShellStreamStarter) { + m.shellStreamStarter = starter +} + // Start starts the lifecycle manager background tasks func (m *Manager) Start(ctx context.Context) error { m.logger.Info("starting lifecycle manager", @@ -836,6 +849,10 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta // 6. Set up file changes stream to track filesystem changes go m.handleFileChangesStream(instance) + // Note: Shell stream is started on-demand when a client sends shell.subscribe + // This ensures the buffered output (including initial prompt) is received + // when a client is actually listening + // Wait for the updates stream to connect before sending prompt // This prevents race conditions where notifications are sent before streams are ready select { @@ -1004,6 +1021,7 @@ func (m *Manager) reconnectStreams(instance *AgentInstance) { go m.handlePermissionStream(instance) go m.handleGitStatusStream(instance) go m.handleFileChangesStream(instance) + // Note: Shell stream is started on-demand via shell.subscribe // Mark the instance as READY so it can accept prompts m.mu.Lock() diff --git a/apps/backend/internal/agentctl/api/server.go b/apps/backend/internal/agentctl/api/server.go index 460a39ab31..ea3b3cf44a 100644 --- a/apps/backend/internal/agentctl/api/server.go +++ b/apps/backend/internal/agentctl/api/server.go @@ -83,6 +83,11 @@ func (s *Server) setupRoutes() { // Workspace file operations (simple HTTP) api.GET("/workspace/tree", s.handleFileTree) api.GET("/workspace/file/content", s.handleFileContent) + + // Shell access + api.GET("/shell/status", s.handleShellStatus) + api.GET("/shell/stream", s.handleShellStreamWS) + api.GET("/shell/buffer", s.handleShellBuffer) } } @@ -183,3 +188,131 @@ func (s *Server) handleStop(c *gin.Context) { Message: "agent stopped", }) } + +// ShellStatusResponse represents shell status +type ShellStatusResponse struct { + Available bool `json:"available"` + Running bool `json:"running"` + Pid int `json:"pid,omitempty"` + Shell string `json:"shell,omitempty"` + Cwd string `json:"cwd,omitempty"` + StartedAt string `json:"started_at,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleShellStatus(c *gin.Context) { + shell := s.procMgr.Shell() + if shell == nil { + c.JSON(http.StatusOK, ShellStatusResponse{ + Available: false, + Error: "shell not available", + }) + return + } + + status := shell.Status() + c.JSON(http.StatusOK, ShellStatusResponse{ + Available: true, + Running: status.Running, + Pid: status.Pid, + Shell: status.Shell, + Cwd: status.Cwd, + StartedAt: status.StartedAt.Format("2006-01-02T15:04:05Z07:00"), + }) +} + +// ShellMessage represents a shell I/O message +type ShellMessage struct { + Type string `json:"type"` // "input", "output", "ping", "pong", "exit" + Data string `json:"data,omitempty"` + Code int `json:"code,omitempty"` +} + +// ShellBufferResponse is the response for GET /api/v1/shell/buffer +type ShellBufferResponse struct { + Data string `json:"data"` +} + +func (s *Server) handleShellBuffer(c *gin.Context) { + shell := s.procMgr.Shell() + if shell == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "shell not available"}) + return + } + + buffered := shell.GetBufferedOutput() + c.JSON(http.StatusOK, ShellBufferResponse{ + Data: string(buffered), + }) +} + +func (s *Server) handleShellStreamWS(c *gin.Context) { + shell := s.procMgr.Shell() + if shell == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "shell not available"}) + return + } + + conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + s.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer func() { + if err := conn.Close(); err != nil { + s.logger.Debug("failed to close shell websocket", zap.Error(err)) + } + }() + + s.logger.Info("Shell stream WebSocket connected") + + // Create output channel and subscribe to shell output + outputCh := make(chan []byte, 256) + shell.Subscribe(outputCh) + defer shell.Unsubscribe(outputCh) + + // Note: Buffered output is sent via GET /api/v1/shell/buffer when clients subscribe + // This avoids duplicate output when clients reconnect + + // Done channel to signal goroutine shutdown + done := make(chan struct{}) + defer close(done) + + // Read input from WebSocket and write to shell + go func() { + for { + select { + case <-done: + return + default: + var msg ShellMessage + if err := conn.ReadJSON(&msg); err != nil { + s.logger.Debug("shell WebSocket read error", zap.Error(err)) + return + } + + switch msg.Type { + case "input": + if _, err := shell.Write([]byte(msg.Data)); err != nil { + s.logger.Debug("shell write error", zap.Error(err)) + } + case "ping": + // Respond with pong + if err := conn.WriteJSON(ShellMessage{Type: "pong"}); err != nil { + s.logger.Debug("shell pong write error", zap.Error(err)) + return + } + } + } + } + }() + + // Write output from shell to WebSocket + for data := range outputCh { + msg := ShellMessage{Type: "output", Data: string(data)} + if err := conn.WriteJSON(msg); err != nil { + s.logger.Debug("shell WebSocket write error", zap.Error(err)) + return + } + } +} diff --git a/apps/backend/internal/agentctl/config/config.go b/apps/backend/internal/agentctl/config/config.go index 432d77d6c8..45b9a98a62 100644 --- a/apps/backend/internal/agentctl/config/config.go +++ b/apps/backend/internal/agentctl/config/config.go @@ -34,6 +34,9 @@ type Config struct { // Defaults provides default values for new instances Defaults InstanceDefaults + // Shell configuration + ShellEnabled bool // Enable auto-shell feature (default: true) + // Logging configuration LogLevel string LogFormat string @@ -96,6 +99,9 @@ type InstanceConfig struct { // AutoApprovePermissions auto-approves permission requests AutoApprovePermissions bool + // ShellEnabled enables auto-shell feature + ShellEnabled bool + // LogLevel for this instance LogLevel string @@ -120,8 +126,9 @@ func Load() *Config { AutoApprovePermissions: getEnvBool("AGENTCTL_AUTO_APPROVE_PERMISSIONS", false), HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5), }, - LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"), - LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"), + ShellEnabled: getEnvBool("AGENTCTL_SHELL_ENABLED", true), + LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"), + LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"), } } @@ -135,6 +142,7 @@ func (c *Config) NewInstanceConfig(port int, overrides *InstanceOverrides) *Inst WorkDir: c.Defaults.WorkDir, AutoStart: c.Defaults.AutoStart, AutoApprovePermissions: c.Defaults.AutoApprovePermissions, + ShellEnabled: c.ShellEnabled, LogLevel: c.LogLevel, LogFormat: c.LogFormat, } diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go index 1823c3a257..2fae6a6ccb 100644 --- a/apps/backend/internal/agentctl/process/manager.go +++ b/apps/backend/internal/agentctl/process/manager.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/agentctl/adapter" "github.com/kandev/kandev/internal/agentctl/config" + "github.com/kandev/kandev/internal/agentctl/shell" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/pkg/agent" "go.uber.org/zap" @@ -70,6 +71,9 @@ type Manager struct { // Workspace tracker for git status and file changes workspaceTracker *WorkspaceTracker + // Embedded shell session (auto-created when agent starts) + shell *shell.Session + // Protocol adapter for agent communication adapter adapter.AgentAdapter @@ -212,6 +216,19 @@ func (m *Manager) Start(ctx context.Context) error { // Start workspace tracker with background context (not tied to HTTP request) m.workspaceTracker.Start(context.Background()) + // Auto-create shell session if enabled + if m.cfg.ShellEnabled { + shellCfg := shell.DefaultConfig(m.cfg.WorkDir) + shellSession, err := shell.NewSession(shellCfg, m.logger) + if err != nil { + m.logger.Warn("failed to create shell session", zap.Error(err)) + // Non-fatal: agent can still work without shell + } else { + m.shell = shellSession + m.logger.Info("shell session auto-created") + } + } + m.status.Store(StatusRunning) m.logger.Info("agent process started", zap.Int("pid", m.cmd.Process.Pid)) @@ -305,6 +322,14 @@ func (m *Manager) Stop(ctx context.Context) error { m.logger.Info("stopping agent process") m.status.Store(StatusStopping) + // Stop shell session + if m.shell != nil { + if err := m.shell.Stop(); err != nil { + m.logger.Debug("failed to stop shell session", zap.Error(err)) + } + m.shell = nil + } + // Stop workspace tracker if m.workspaceTracker != nil { m.workspaceTracker.Stop() @@ -569,3 +594,9 @@ func (m *Manager) GetPendingPermissions() []*PendingPermission { return result } +// Shell returns the embedded shell session, or nil if not available +func (m *Manager) Shell() *shell.Session { + m.mu.RLock() + defer m.mu.RUnlock() + return m.shell +} diff --git a/apps/backend/internal/agentctl/shell/session.go b/apps/backend/internal/agentctl/shell/session.go new file mode 100644 index 0000000000..57bb7ac44d --- /dev/null +++ b/apps/backend/internal/agentctl/shell/session.go @@ -0,0 +1,343 @@ +// Package shell provides an embedded PTY-based shell session for agentctl. +// The shell session is automatically created when agentctl starts and destroyed when it stops. +package shell + +import ( + "fmt" + "io" + "os" + "os/exec" + "runtime" + "sync" + "time" + + "github.com/creack/pty" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Session represents the single embedded shell session for an agentctl instance. +// It is automatically created when agentctl starts and destroyed when it stops. +type Session struct { + logger *logger.Logger + workDir string + shell string // Shell command (detected based on OS) + shellArgs []string // Shell arguments + + // PTY and process + pty *os.File + cmd *exec.Cmd + + // State + running bool + startedAt time.Time + mu sync.RWMutex + + // Output subscribers (WebSocket connections) + subscribers map[chan<- []byte]struct{} + subMu sync.RWMutex + + // Ring buffer for recent output (for new subscribers) + outputBuffer []byte + bufferMu sync.RWMutex + + // Lifecycle + stopCh chan struct{} + doneCh chan struct{} +} + +// Maximum size of output buffer (16KB should be enough for recent history) +const maxOutputBufferSize = 16 * 1024 + +// Config holds shell session configuration +type Config struct { + WorkDir string // Working directory (default: workspace) + Cols int // Initial terminal columns (default: 80) + Rows int // Initial terminal rows (default: 24) +} + +// DefaultConfig returns the default shell configuration +func DefaultConfig(workDir string) Config { + return Config{ + WorkDir: workDir, + Cols: 80, + Rows: 24, + } +} + +// Status represents the shell status +type Status struct { + Running bool + Pid int + Shell string + Cwd string + StartedAt time.Time +} + +// detectShell returns the appropriate shell for the current OS +func detectShell() (string, []string) { + if runtime.GOOS == "windows" { + // Prefer PowerShell if available + if _, err := exec.LookPath("pwsh.exe"); err == nil { + return "pwsh.exe", []string{"-NoLogo", "-NoExit"} + } + if _, err := exec.LookPath("powershell.exe"); err == nil { + return "powershell.exe", []string{"-NoLogo", "-NoExit"} + } + return "cmd.exe", nil + } + + // Unix-like systems (Linux, macOS) + if shell := os.Getenv("SHELL"); shell != "" { + return shell, []string{"-l"} // Login shell + } + + // Try common shells + shells := []string{"/bin/bash", "/bin/zsh", "/bin/sh"} + for _, sh := range shells { + if _, err := os.Stat(sh); err == nil { + return sh, []string{"-l"} + } + } + + return "/bin/sh", nil +} + +// NewSession creates and starts a new shell session. +// This is called automatically when agentctl starts. +func NewSession(cfg Config, log *logger.Logger) (*Session, error) { + shell, args := detectShell() + + s := &Session{ + logger: log.WithFields(zap.String("component", "shell")), + workDir: cfg.WorkDir, + shell: shell, + shellArgs: args, + subscribers: make(map[chan<- []byte]struct{}), + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } + + if err := s.start(cfg); err != nil { + return nil, err + } + + return s, nil +} + +// start initializes and starts the shell process +func (s *Session) start(cfg Config) error { + s.mu.Lock() + defer s.mu.Unlock() + + s.cmd = exec.Command(s.shell, s.shellArgs...) + s.cmd.Dir = cfg.WorkDir + s.cmd.Env = buildShellEnv(cfg.WorkDir) + + // Start with PTY + var err error + s.pty, err = pty.StartWithSize(s.cmd, &pty.Winsize{ + Cols: uint16(cfg.Cols), + Rows: uint16(cfg.Rows), + }) + if err != nil { + return fmt.Errorf("failed to start PTY: %w", err) + } + + s.running = true + s.startedAt = time.Now() + + s.logger.Info("shell session started", + zap.String("shell", s.shell), + zap.String("cwd", cfg.WorkDir), + zap.Int("pid", s.cmd.Process.Pid)) + + // Start output reader + go s.readOutput() + + // Wait for process exit + go s.waitForExit() + + return nil +} + +// Stop gracefully stops the shell session. +// Called automatically when agentctl stops. +func (s *Session) Stop() error { + s.mu.Lock() + if !s.running { + s.mu.Unlock() + return nil + } + s.running = false + s.mu.Unlock() + + s.logger.Info("stopping shell session") + + // Signal stop + close(s.stopCh) + + // Close PTY (sends SIGHUP to process on Unix) + if s.pty != nil { + _ = s.pty.Close() + } + + // Wait for exit with timeout + select { + case <-s.doneCh: + s.logger.Info("shell session stopped gracefully") + case <-time.After(5 * time.Second): + s.logger.Warn("shell session stop timeout, force killing") + if s.cmd != nil && s.cmd.Process != nil { + _ = s.cmd.Process.Kill() + } + } + + return nil +} + +// Write sends input to the shell +func (s *Session) Write(data []byte) (int, error) { + s.mu.RLock() + defer s.mu.RUnlock() + + if !s.running || s.pty == nil { + return 0, fmt.Errorf("shell not running") + } + + return s.pty.Write(data) +} + +// Subscribe adds a subscriber for shell output +func (s *Session) Subscribe(ch chan<- []byte) { + s.subMu.Lock() + defer s.subMu.Unlock() + s.subscribers[ch] = struct{}{} +} + +// Unsubscribe removes a subscriber +func (s *Session) Unsubscribe(ch chan<- []byte) { + s.subMu.Lock() + defer s.subMu.Unlock() + delete(s.subscribers, ch) +} + +// Status returns the current shell status +func (s *Session) Status() Status { + s.mu.RLock() + defer s.mu.RUnlock() + + pid := 0 + if s.cmd != nil && s.cmd.Process != nil { + pid = s.cmd.Process.Pid + } + + return Status{ + Running: s.running, + Pid: pid, + Shell: s.shell, + Cwd: s.workDir, + StartedAt: s.startedAt, + } +} + +// readOutput continuously reads from PTY and broadcasts to subscribers +func (s *Session) readOutput() { + buf := make([]byte, 4096) + + for { + select { + case <-s.stopCh: + return + default: + } + + n, err := s.pty.Read(buf) + if err != nil { + if err != io.EOF { + s.logger.Debug("shell read error", zap.Error(err)) + } + return + } + + if n > 0 { + data := make([]byte, n) + copy(data, buf[:n]) + s.broadcast(data) + } + } +} + +// broadcast sends data to all subscribers and stores in buffer +func (s *Session) broadcast(data []byte) { + // Store in ring buffer for new subscribers + s.appendToBuffer(data) + + s.subMu.RLock() + defer s.subMu.RUnlock() + + for ch := range s.subscribers { + select { + case ch <- data: + default: + // Subscriber channel full, skip + } + } +} + +// appendToBuffer adds data to the output ring buffer +func (s *Session) appendToBuffer(data []byte) { + s.bufferMu.Lock() + defer s.bufferMu.Unlock() + + s.outputBuffer = append(s.outputBuffer, data...) + + // Trim if exceeds max size (keep the most recent data) + if len(s.outputBuffer) > maxOutputBufferSize { + s.outputBuffer = s.outputBuffer[len(s.outputBuffer)-maxOutputBufferSize:] + } +} + +// GetBufferedOutput returns the buffered recent output +func (s *Session) GetBufferedOutput() []byte { + s.bufferMu.RLock() + defer s.bufferMu.RUnlock() + + if len(s.outputBuffer) == 0 { + return nil + } + + // Return a copy to avoid race conditions + result := make([]byte, len(s.outputBuffer)) + copy(result, s.outputBuffer) + return result +} + +// waitForExit waits for the shell process to exit +func (s *Session) waitForExit() { + defer close(s.doneCh) + + if s.cmd != nil { + _ = s.cmd.Wait() + } + + s.mu.Lock() + s.running = false + s.mu.Unlock() + + s.logger.Info("shell process exited") +} + +// buildShellEnv creates the environment for the shell process +func buildShellEnv(workDir string) []string { + env := os.Environ() + + // Set working directory related vars + env = append(env, "PWD="+workDir) + + // Set terminal type + env = append(env, "TERM=xterm-256color") + + return env +} + diff --git a/apps/backend/internal/agentctl/shell/session_test.go b/apps/backend/internal/agentctl/shell/session_test.go new file mode 100644 index 0000000000..f419f587d2 --- /dev/null +++ b/apps/backend/internal/agentctl/shell/session_test.go @@ -0,0 +1,546 @@ +package shell + +import ( + "os" + "runtime" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +func newTestLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{ + Level: "error", // Suppress logs during tests + Format: "console", + }) + return log +} + +// TestDefaultConfig tests that DefaultConfig returns expected values +func TestDefaultConfig(t *testing.T) { + workDir := "/test/dir" + cfg := DefaultConfig(workDir) + + if cfg.WorkDir != workDir { + t.Errorf("expected WorkDir %q, got %q", workDir, cfg.WorkDir) + } + if cfg.Cols != 80 { + t.Errorf("expected Cols 80, got %d", cfg.Cols) + } + if cfg.Rows != 24 { + t.Errorf("expected Rows 24, got %d", cfg.Rows) + } +} + +// TestDetectShell tests shell detection for the current OS +func TestDetectShell(t *testing.T) { + shell, args := detectShell() + + if shell == "" { + t.Error("detectShell returned empty shell") + } + + // Verify detected shell exists on Unix systems + if runtime.GOOS != "windows" { + if _, err := os.Stat(shell); err != nil { + // May be in PATH, try LookPath + if _, err := os.Stat(shell); err != nil && shell != os.Getenv("SHELL") { + t.Logf("Warning: detected shell %q may not exist: %v", shell, err) + } + } + // Unix shells should have -l flag for login shell + if len(args) > 0 && args[0] != "-l" { + t.Errorf("expected Unix shell args to contain '-l', got %v", args) + } + } else { + // Windows should detect cmd.exe, powershell.exe, or pwsh.exe + validShells := map[string]bool{ + "cmd.exe": true, + "powershell.exe": true, + "pwsh.exe": true, + } + if !validShells[shell] { + t.Errorf("unexpected Windows shell: %q", shell) + } + } +} + +// TestDetectShellWithSHELLEnv tests shell detection respects SHELL env var on Unix +func TestDetectShellWithSHELLEnv(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("SHELL env var is not used on Windows") + } + + originalShell := os.Getenv("SHELL") + defer func() { _ = os.Setenv("SHELL", originalShell) }() + + // Test with custom SHELL + _ = os.Setenv("SHELL", "/bin/custom-shell") + shell, args := detectShell() + + if shell != "/bin/custom-shell" { + t.Errorf("expected shell from SHELL env, got %q", shell) + } + if len(args) == 0 || args[0] != "-l" { + t.Errorf("expected args to contain '-l', got %v", args) + } +} + +// TestNewSession tests creating a new session +func TestNewSession(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + // PTY may not work in all CI environments + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Verify session is running + status := session.Status() + if !status.Running { + t.Error("expected session to be running") + } + if status.Pid == 0 { + t.Error("expected non-zero PID") + } + if status.Cwd != workDir { + t.Errorf("expected Cwd %q, got %q", workDir, status.Cwd) + } + if status.Shell == "" { + t.Error("expected non-empty Shell") + } + if status.StartedAt.IsZero() { + t.Error("expected non-zero StartedAt") + } +} + +// TestSessionStop tests stopping a session +func TestSessionStop(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + + // Stop the session + if err := session.Stop(); err != nil { + t.Errorf("Stop failed: %v", err) + } + + // Verify session is stopped + status := session.Status() + if status.Running { + t.Error("expected session to not be running after Stop") + } + + // Calling Stop again should be idempotent + if err := session.Stop(); err != nil { + t.Errorf("second Stop failed: %v", err) + } +} + +// TestSessionWrite tests writing to the shell +func TestSessionWrite(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Write a simple command + data := []byte("echo hello\n") + n, err := session.Write(data) + if err != nil { + t.Errorf("Write failed: %v", err) + } + if n != len(data) { + t.Errorf("expected to write %d bytes, wrote %d", len(data), n) + } +} + +// TestSessionWriteNotRunning tests writing to a stopped session +func TestSessionWriteNotRunning(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + + // Stop the session first + _ = session.Stop() + + // Write should fail + _, err = session.Write([]byte("echo hello\n")) + if err == nil { + t.Error("expected Write to fail on stopped session") + } +} + +// TestSessionSubscribeUnsubscribe tests the subscriber pattern +func TestSessionSubscribeUnsubscribe(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Create subscriber channel + ch := make(chan []byte, 100) + + // Subscribe + session.Subscribe(ch) + + // Write something to trigger output + _, _ = session.Write([]byte("echo test\n")) + + // Wait a bit for output + time.Sleep(100 * time.Millisecond) + + // Unsubscribe + session.Unsubscribe(ch) + + // The channel should have received some data (shell prompt + echo output) + select { + case data := <-ch: + if len(data) == 0 { + t.Error("expected non-empty output") + } + default: + // May not receive data immediately, that's okay + t.Log("no output received (may be timing-dependent)") + } +} + +// TestSessionMultipleSubscribers tests multiple subscribers receive output +func TestSessionMultipleSubscribers(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Create multiple subscriber channels + ch1 := make(chan []byte, 100) + ch2 := make(chan []byte, 100) + ch3 := make(chan []byte, 100) + + session.Subscribe(ch1) + session.Subscribe(ch2) + session.Subscribe(ch3) + + // Write something + _, _ = session.Write([]byte("echo multi\n")) + + // Wait for output + time.Sleep(100 * time.Millisecond) + + // All channels should potentially receive data + // Just verify no panics or errors occurred + + session.Unsubscribe(ch1) + session.Unsubscribe(ch2) + session.Unsubscribe(ch3) +} + +// TestSessionStatus tests Status returns correct information +func TestSessionStatus(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + status := session.Status() + + // Verify all fields + if !status.Running { + t.Error("expected Running to be true") + } + if status.Pid <= 0 { + t.Errorf("expected positive PID, got %d", status.Pid) + } + if status.Shell == "" { + t.Error("expected non-empty Shell") + } + if status.Cwd != workDir { + t.Errorf("expected Cwd %q, got %q", workDir, status.Cwd) + } + if status.StartedAt.IsZero() { + t.Error("expected non-zero StartedAt") + } + if time.Since(status.StartedAt) > 5*time.Second { + t.Error("StartedAt seems too old") + } +} + +// TestSessionStatusAfterStop tests Status after stopping +func TestSessionStatusAfterStop(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + + _ = session.Stop() + + status := session.Status() + if status.Running { + t.Error("expected Running to be false after Stop") + } +} + +// TestBuildShellEnv tests the buildShellEnv function +func TestBuildShellEnv(t *testing.T) { + workDir := "/test/workspace" + env := buildShellEnv(workDir) + + // Check that env is not empty + if len(env) == 0 { + t.Error("expected non-empty environment") + } + + // Check for PWD + pwdFound := false + termFound := false + for _, e := range env { + if e == "PWD="+workDir { + pwdFound = true + } + if e == "TERM=xterm-256color" { + termFound = true + } + } + + if !pwdFound { + t.Error("expected PWD to be set in environment") + } + if !termFound { + t.Error("expected TERM to be set in environment") + } +} + +// TestSessionConcurrentSubscribe tests concurrent subscribe/unsubscribe +func TestSessionConcurrentSubscribe(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Concurrently subscribe and unsubscribe + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ch := make(chan []byte, 10) + session.Subscribe(ch) + time.Sleep(10 * time.Millisecond) + session.Unsubscribe(ch) + }() + } + + wg.Wait() +} + +// TestSessionConcurrentWrite tests concurrent writes +func TestSessionConcurrentWrite(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Concurrently write + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + _, _ = session.Write([]byte("echo test\n")) + }(i) + } + + wg.Wait() +} + +// TestSessionConcurrentStatus tests concurrent status reads +func TestSessionConcurrentStatus(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("PTY is not supported on Windows") + } + + if os.Getenv("CI") != "" { + t.Skip("Skipping PTY test in CI environment") + } + + log := newTestLogger() + workDir := t.TempDir() + cfg := DefaultConfig(workDir) + + session, err := NewSession(cfg, log) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + defer func() { _ = session.Stop() }() + + // Concurrently read status + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + status := session.Status() + if !status.Running { + t.Error("expected session to be running") + } + }() + } + + wg.Wait() +} + +// TestDetectShellFallback tests shell fallback when SHELL is unset +func TestDetectShellFallback(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("SHELL env var fallback is for Unix only") + } + + originalShell := os.Getenv("SHELL") + defer func() { _ = os.Setenv("SHELL", originalShell) }() + + // Unset SHELL + _ = os.Unsetenv("SHELL") + + shell, _ := detectShell() + + // Should fall back to one of the common shells + validShells := map[string]bool{ + "/bin/bash": true, + "/bin/zsh": true, + "/bin/sh": true, + } + if !validShells[shell] { + t.Logf("Detected shell: %q (may be valid if it exists on this system)", shell) + } +} + diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index b34a12271b..108f80d405 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -164,17 +164,21 @@ func (h *Hub) BroadcastToTask(taskID string, msg *ws.Message) { return } + // Copy client list while holding the lock to avoid race conditions h.mu.RLock() - clients := h.taskSubscribers[taskID] - numClients := len(clients) + subscriberMap := h.taskSubscribers[taskID] + clients := make([]*Client, 0, len(subscriberMap)) + for client := range subscriberMap { + clients = append(clients, client) + } h.mu.RUnlock() h.logger.Debug("BroadcastToTask", zap.String("task_id", taskID), zap.String("action", msg.Action), - zap.Int("subscriber_count", numClients)) + zap.Int("subscriber_count", len(clients))) - for client := range clients { + for _, client := range clients { if client.sendBytes(data) { h.logger.Debug("Sent message to client", zap.String("client_id", client.ID), @@ -195,17 +199,21 @@ func (h *Hub) BroadcastToUser(userID string, msg *ws.Message) { return } + // Copy client list while holding the lock to avoid race conditions h.mu.RLock() - clients := h.userSubscribers[userID] - numClients := len(clients) + subscriberMap := h.userSubscribers[userID] + clients := make([]*Client, 0, len(subscriberMap)) + for client := range subscriberMap { + clients = append(clients, client) + } h.mu.RUnlock() h.logger.Debug("BroadcastToUser", zap.String("user_id", userID), zap.String("action", msg.Action), - zap.Int("subscriber_count", numClients)) + zap.Int("subscriber_count", len(clients))) - for client := range clients { + for _, client := range clients { if client.sendBytes(data) { h.logger.Debug("Sent message to client", zap.String("client_id", client.ID), diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 0ec36c9c95..8983475bc1 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -64,6 +64,7 @@ type simulatedInstance struct { taskID string agentProfileID string status v1.AgentStatus + statusMu sync.Mutex // Protects status field stopCh chan struct{} } @@ -168,18 +169,24 @@ func (s *SimulatedAgentManagerClient) runAgentSimulation(instance *simulatedInst } // Publish agent ready (finished prompt, waiting for follow-up or completion) + instance.statusMu.Lock() instance.status = v1.AgentStatusReady + instance.statusMu.Unlock() s.publishAgentEvent(events.AgentReady, instance) } // publishAgentEvent publishes an agent lifecycle event func (s *SimulatedAgentManagerClient) publishAgentEvent(eventType string, instance *simulatedInstance) { + instance.statusMu.Lock() + status := instance.status + instance.statusMu.Unlock() + data := map[string]interface{}{ "instance_id": instance.id, "task_id": instance.taskID, "agent_profile_id": instance.agentProfileID, "container_id": "sim-container-" + instance.id[:8], - "status": string(instance.status), + "status": string(status), "started_at": time.Now(), "progress": 50, } @@ -261,7 +268,9 @@ func (s *SimulatedAgentManagerClient) StopAgent(ctx context.Context, agentInstan } close(instance.stopCh) + instance.statusMu.Lock() instance.status = v1.AgentStatusStopped + instance.statusMu.Unlock() s.publishAgentEvent(events.AgentStopped, instance) return nil @@ -277,11 +286,15 @@ func (s *SimulatedAgentManagerClient) GetAgentStatus(ctx context.Context, agentI return nil, fmt.Errorf("agent instance %q not found", agentInstanceID) } + instance.statusMu.Lock() + status := instance.status + instance.statusMu.Unlock() + return &v1.AgentInstance{ ID: instance.id, TaskID: instance.taskID, AgentProfileID: instance.agentProfileID, - Status: instance.status, + Status: status, }, nil } diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 83cf1dad3e..3d710860dc 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -150,6 +150,12 @@ const ( ActionWorkspaceFileContentGet = "workspace.file.get" ActionWorkspaceFileChanges = "workspace.file.changes" // Notification + // Shell actions + ActionShellStatus = "shell.status" // Get shell status + ActionShellSubscribe = "shell.subscribe" // Subscribe to shell output + ActionShellInput = "shell.input" // Send input to shell + ActionShellOutput = "shell.output" // Shell output notification + // User actions ActionUserGet = "user.get" ActionUserSettingsUpdate = "user.settings.update" diff --git a/apps/pnpm-lock.yaml b/apps/pnpm-lock.yaml index c7b53f54bb..7543a81b80 100644 --- a/apps/pnpm-lock.yaml +++ b/apps/pnpm-lock.yaml @@ -68,7 +68,7 @@ importers: version: 9.39.2(jiti@2.6.1) eslint-config-next: specifier: 16.1.1 - version: 16.1.1(@typescript-eslint/parser@8.53.0(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3))(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3) + version: 16.1.1(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3) tailwindcss: specifier: ^4.0.0 version: 4.1.18 @@ -277,6 +277,12 @@ importers: '@uiw/react-codemirror': specifier: ^4.25.4 version: 4.25.4(@babel/runtime@7.28.6)(@codemirror/autocomplete@6.20.0)(@codemirror/language@6.12.1)(@codemirror/lint@6.9.2)(@codemirror/search@6.6.0)(@codemirror/state@6.5.3)(@codemirror/theme-one-dark@6.1.3)(@codemirror/view@6.39.10)(codemirror@6.0.2)(react-dom@19.2.3(react@19.2.3))(react@19.2.3) + '@xterm/addon-fit': + specifier: ^0.11.0 + version: 0.11.0 + '@xterm/xterm': + specifier: ^6.0.0 + version: 6.0.0 class-variance-authority: specifier: ^0.7.1 version: 0.7.1 @@ -367,7 +373,7 @@ importers: version: 9.39.2(jiti@2.6.1) eslint-config-next: specifier: 16.1.1 - version: 16.1.1(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3) + version: 16.1.1(@typescript-eslint/parser@8.53.0(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3))(eslint@9.39.2(jiti@2.6.1))(typescript@5.9.3) happy-dom: specifier: ^20.1.0 version: 20.1.0 @@ -2504,6 +2510,12 @@ packages: '@vue/shared@3.5.26': resolution: {integrity: sha512-7Z6/y3uFI5PRoKeorTOSXKcDj0MSasfNNltcslbFrPpcw6aXRUALq4IfJlaTRspiWIUOEZbrpM+iQGmCOiWe4A==} + '@xterm/addon-fit@0.11.0': + resolution: {integrity: sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g==} + + '@xterm/xterm@6.0.0': + resolution: {integrity: sha512-TQwDdQGtwwDt+2cgKDLn0IRaSxYu1tSUjgKarSDkUM0ZNiSRXFpjxEsvc/Zgc5kq5omJ+V0a8/kIM2WD3sMOYg==} + accepts@2.0.0: resolution: {integrity: sha512-5cvg6CtKwfgdmVqY1WIiXKc3Q1bkRqGLi+2W/6ao+6Y7gu/RCwRuAhGEzh5B4KlszSuTLgZYuqFqo5bImjNKng==} engines: {node: '>= 0.6'} @@ -7528,6 +7540,10 @@ snapshots: '@vue/shared@3.5.26': {} + '@xterm/addon-fit@0.11.0': {} + + '@xterm/xterm@6.0.0': {} + accepts@2.0.0: dependencies: mime-types: 3.0.2 diff --git a/apps/web/components/task/shell-terminal.tsx b/apps/web/components/task/shell-terminal.tsx new file mode 100644 index 0000000000..9ed7908945 --- /dev/null +++ b/apps/web/components/task/shell-terminal.tsx @@ -0,0 +1,147 @@ +'use client'; + +import { useEffect, useRef, useCallback } from 'react'; +import { Terminal } from '@xterm/xterm'; +import { FitAddon } from '@xterm/addon-fit'; +import '@xterm/xterm/css/xterm.css'; +import { useAppStore, useAppStoreApi } from '@/components/state-provider'; +import { getWebSocketClient } from '@/lib/ws/connection'; + +type ShellTerminalProps = { + taskId: string; +}; + +export function ShellTerminal({ taskId }: ShellTerminalProps) { + const terminalRef = useRef(null); + const xtermRef = useRef(null); + const fitAddonRef = useRef(null); + const lastOutputLengthRef = useRef(0); + const subscriptionIdRef = useRef(0); + const storeApi = useAppStoreApi(); + + const shellOutput = useAppStore((state) => state.shell.outputs[taskId] || ''); + + const send = useCallback( + (action: string, payload: Record) => { + const client = getWebSocketClient(); + if (client) { + client.send({ type: 'request', action, payload }); + } + }, + [] + ); + + // Initialize terminal + useEffect(() => { + if (!terminalRef.current || xtermRef.current) return; + + const terminal = new Terminal({ + cursorBlink: true, + fontSize: 13, + fontFamily: 'Menlo, Monaco, "Courier New", monospace', + theme: { + background: '#1e1e1e', + foreground: '#d4d4d4', + cursor: '#d4d4d4', + cursorAccent: '#1e1e1e', + selectionBackground: '#264f78', + black: '#1e1e1e', + red: '#f44747', + green: '#6a9955', + yellow: '#dcdcaa', + blue: '#569cd6', + magenta: '#c586c0', + cyan: '#4ec9b0', + white: '#d4d4d4', + brightBlack: '#808080', + brightRed: '#f44747', + brightGreen: '#6a9955', + brightYellow: '#dcdcaa', + brightBlue: '#569cd6', + brightMagenta: '#c586c0', + brightCyan: '#4ec9b0', + brightWhite: '#ffffff', + }, + }); + + const fitAddon = new FitAddon(); + terminal.loadAddon(fitAddon); + terminal.open(terminalRef.current); + fitAddon.fit(); + + xtermRef.current = terminal; + fitAddonRef.current = fitAddon; + + // Handle user input + terminal.onData((data) => { + send('shell.input', { task_id: taskId, data }); + }); + + // Handle resize + const resizeObserver = new ResizeObserver(() => { + fitAddon.fit(); + }); + resizeObserver.observe(terminalRef.current); + + return () => { + resizeObserver.disconnect(); + terminal.dispose(); + xtermRef.current = null; + fitAddonRef.current = null; + }; + }, [taskId, send]); + + // Write new output to terminal + useEffect(() => { + if (!xtermRef.current) return; + + // Only write new data since last update + const newData = shellOutput.slice(lastOutputLengthRef.current); + if (newData) { + xtermRef.current.write(newData); + lastOutputLengthRef.current = shellOutput.length; + } + }, [shellOutput]); + + // Subscribe to shell on mount to start the shell stream + useEffect(() => { + // Increment subscription ID to invalidate any pending requests + const currentSubscriptionId = ++subscriptionIdRef.current; + + // Clear any stale output before subscribing to get fresh buffer + storeApi.getState().clearShellOutput(taskId); + lastOutputLengthRef.current = 0; + + // Also clear the terminal display + if (xtermRef.current) { + xtermRef.current.clear(); + } + + const client = getWebSocketClient(); + if (client) { + client + .request<{ success: boolean; buffer?: string }>('shell.subscribe', { task_id: taskId }) + .then((response) => { + // Only process if this is still the current subscription (handles React Strict Mode) + if (subscriptionIdRef.current !== currentSubscriptionId) return; + + // Write buffered output from response + if (response.buffer) { + storeApi.getState().appendShellOutput(taskId, response.buffer); + } + }) + .catch((err) => { + if (subscriptionIdRef.current !== currentSubscriptionId) return; + console.error('Failed to subscribe to shell:', err); + }); + } + }, [taskId, storeApi]); + + return ( +
+ ); +} + diff --git a/apps/web/components/task/task-layout.tsx b/apps/web/components/task/task-layout.tsx index 62625841a5..0340debf34 100644 --- a/apps/web/components/task/task-layout.tsx +++ b/apps/web/components/task/task-layout.tsx @@ -72,7 +72,7 @@ export const TaskLayout = memo(function TaskLayout({ - +
diff --git a/apps/web/components/task/task-right-panel.tsx b/apps/web/components/task/task-right-panel.tsx index b3ab393056..98659535b1 100644 --- a/apps/web/components/task/task-right-panel.tsx +++ b/apps/web/components/task/task-right-panel.tsx @@ -8,14 +8,16 @@ import { ResizableHandle, ResizablePanel, ResizablePanelGroup } from '@kandev/ui import { Tabs, TabsContent, TabsList, TabsTrigger } from '@kandev/ui/tabs'; import { getLocalStorage, setLocalStorage } from '@/lib/local-storage'; import { COMMANDS } from '@/components/task/task-data'; +import { ShellTerminal } from '@/components/task/shell-terminal'; type TaskRightPanelProps = { topPanel: ReactNode; + taskId: string; }; const DEFAULT_RIGHT_LAYOUT: [number, number] = [55, 45]; -const TaskRightPanel = memo(function TaskRightPanel({ topPanel }: TaskRightPanelProps) { +const TaskRightPanel = memo(function TaskRightPanel({ topPanel, taskId }: TaskRightPanelProps) { const [rightLayout, setRightLayout] = useState<[number, number]>(() => getLocalStorage('task-layout-right', DEFAULT_RIGHT_LAYOUT) ); @@ -176,11 +178,8 @@ const TaskRightPanel = memo(function TaskRightPanel({ topPanel }: TaskRightPanel {terminalIds.map((id) => ( -
-
-

kan-dev@workspace:~$

-

_

-
+
+
))} diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 56044e681c..62056913cb 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -111,6 +111,21 @@ export type TerminalState = { terminals: Array<{ id: string; output: string[] }>; }; +export type ShellState = { + // Map of taskId to shell output buffer (raw bytes as string) + outputs: Record; + // Map of taskId to shell status + statuses: Record< + string, + { + available: boolean; + running?: boolean; + shell?: string; + cwd?: string; + } + >; +}; + export type FileInfo = { path: string; status: 'modified' | 'added' | 'deleted' | 'untracked' | 'renamed'; @@ -168,6 +183,7 @@ export type AppState = { agentProfiles: AgentProfilesState; userSettings: UserSettingsState; terminal: TerminalState; + shell: ShellState; diffs: DiffState; gitStatus: GitStatusState; connection: ConnectionState; @@ -190,6 +206,12 @@ export type AppState = { setSettingsData: (next: Partial) => void; setUserSettings: (settings: UserSettingsState) => void; setTerminalOutput: (terminalId: string, data: string) => void; + appendShellOutput: (taskId: string, data: string) => void; + setShellStatus: ( + taskId: string, + status: { available: boolean; running?: boolean; shell?: string; cwd?: string } + ) => void; + clearShellOutput: (taskId: string) => void; setConnectionStatus: (status: ConnectionState['status'], error?: string | null) => void; setMessages: ( sessionId: string | null, @@ -226,6 +248,7 @@ const defaultState: AppState = { agentProfiles: { items: [], version: 0 }, userSettings: { workspaceId: null, boardId: null, repositoryIds: [], loaded: false }, terminal: { terminals: [] }, + shell: { outputs: {}, statuses: {} }, diffs: { files: [] }, gitStatus: { taskId: null, @@ -261,6 +284,9 @@ const defaultState: AppState = { setSettingsData: () => undefined, setUserSettings: () => undefined, setTerminalOutput: () => undefined, + appendShellOutput: () => undefined, + setShellStatus: () => undefined, + clearShellOutput: () => undefined, setConnectionStatus: () => undefined, setMessages: () => undefined, setMessagesSessionId: () => undefined, @@ -296,6 +322,9 @@ function mergeInitialState( | 'setSettingsData' | 'setUserSettings' | 'setTerminalOutput' + | 'appendShellOutput' + | 'setShellStatus' + | 'clearShellOutput' | 'setConnectionStatus' | 'setMessages' | 'setMessagesSessionId' @@ -326,6 +355,7 @@ function mergeInitialState( agentProfiles: { ...defaultState.agentProfiles, ...initialState.agentProfiles }, userSettings: { ...defaultState.userSettings, ...initialState.userSettings }, terminal: { ...defaultState.terminal, ...initialState.terminal }, + shell: { ...defaultState.shell, ...initialState.shell }, diffs: { ...defaultState.diffs, ...initialState.diffs }, gitStatus: { ...defaultState.gitStatus, ...initialState.gitStatus }, connection: { ...defaultState.connection, ...initialState.connection }, @@ -360,6 +390,7 @@ export function createAppStore(initialState?: Partial) { if (state.agentProfiles) Object.assign(draft.agentProfiles, state.agentProfiles); if (state.userSettings) Object.assign(draft.userSettings, state.userSettings); if (state.terminal) Object.assign(draft.terminal, state.terminal); + if (state.shell) Object.assign(draft.shell, state.shell); if (state.diffs) Object.assign(draft.diffs, state.diffs); if (state.connection) Object.assign(draft.connection, state.connection); if (state.messages) Object.assign(draft.messages, state.messages); @@ -452,6 +483,18 @@ export function createAppStore(initialState?: Partial) { draft.terminal.terminals.push({ id: terminalId, output: [data] }); } }), + appendShellOutput: (taskId, data) => + set((draft) => { + draft.shell.outputs[taskId] = (draft.shell.outputs[taskId] || '') + data; + }), + setShellStatus: (taskId, status) => + set((draft) => { + draft.shell.statuses[taskId] = status; + }), + clearShellOutput: (taskId) => + set((draft) => { + draft.shell.outputs[taskId] = ''; + }), setConnectionStatus: (status, error = null) => set((draft) => { draft.connection.status = status; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index dbcd9d7b5f..dd72d92025 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -31,7 +31,8 @@ export type BackendMessageType = | 'agent.profile.created' | 'agent.profile.updated' | 'user.settings.updated' - | 'workspace.file.changes'; + | 'workspace.file.changes' + | 'shell.output'; export type BackendMessage = { id?: string; @@ -223,6 +224,13 @@ export type UserSettingsUpdatedPayload = { updated_at?: string; }; +export type ShellOutputPayload = { + task_id: string; + type: 'output' | 'exit'; + data?: string; + code?: number; +}; + export type BackendMessageMap = { 'kanban.update': BackendMessage<'kanban.update', KanbanUpdatePayload>; 'task.created': BackendMessage<'task.created', TaskEventPayload>; @@ -257,6 +265,7 @@ export type BackendMessageMap = { 'agent.profile.updated': BackendMessage<'agent.profile.updated', AgentProfileChangedPayload>; 'user.settings.updated': BackendMessage<'user.settings.updated', UserSettingsUpdatedPayload>; 'workspace.file.changes': BackendMessage<'workspace.file.changes', FileChangeNotificationPayload>; + 'shell.output': BackendMessage<'shell.output', ShellOutputPayload>; }; // Workspace file types diff --git a/apps/web/lib/ws/handlers/terminals.ts b/apps/web/lib/ws/handlers/terminals.ts index d344195fca..29b1666cd4 100644 --- a/apps/web/lib/ws/handlers/terminals.ts +++ b/apps/web/lib/ws/handlers/terminals.ts @@ -7,5 +7,14 @@ export function registerTerminalsHandlers(store: StoreApi): WsHandlers 'terminal.output': (message) => { store.getState().setTerminalOutput(message.payload.terminalId, message.payload.data); }, + 'shell.output': (message) => { + const { task_id, type, data } = message.payload; + if (type === 'output' && data) { + store.getState().appendShellOutput(task_id, data); + } else if (type === 'exit') { + // Shell exited - update status + store.getState().setShellStatus(task_id, { available: false }); + } + }, }; } diff --git a/apps/web/package.json b/apps/web/package.json index b308668628..d9925e3d9f 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -11,6 +11,8 @@ }, "dependencies": { "@base-ui/react": "^1.0.0", + "@codemirror/autocomplete": "^6.18.1", + "@codemirror/commands": "^6.7.0", "@codemirror/lang-cpp": "^6.0.3", "@codemirror/lang-css": "^6.3.1", "@codemirror/lang-go": "^6.0.1", @@ -22,8 +24,6 @@ "@codemirror/lang-python": "^6.2.1", "@codemirror/lang-rust": "^6.0.2", "@codemirror/lang-yaml": "^6.1.2", - "@codemirror/autocomplete": "^6.18.1", - "@codemirror/commands": "^6.7.0", "@codemirror/state": "^6.5.3", "@codemirror/view": "^6.39.10", "@dnd-kit/core": "^6.3.1", @@ -37,6 +37,8 @@ "@tanstack/react-virtual": "^3.13.18", "@uiw/codemirror-theme-vscode": "^4.25.4", "@uiw/react-codemirror": "^4.25.4", + "@xterm/addon-fit": "^0.11.0", + "@xterm/xterm": "^6.0.0", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "cmdk": "^1.1.1",