// Package kubernetes provides Kubernetes-based implementations of port interfaces. package kubernetes import ( "bufio" "context" "fmt" "io" "os/exec" "sync" "time" "github.com/orchard9/rdev/internal/domain" "github.com/orchard9/rdev/internal/port" ) // Executor implements port.CommandExecutor using kubectl exec. type Executor struct { namespace string mu sync.RWMutex // Track active commands for cancellation activeCommands map[domain.CommandID]context.CancelFunc activeMu sync.Mutex } // NewExecutor creates a new Kubernetes command executor. func NewExecutor(namespace string) *Executor { return &Executor{ namespace: namespace, activeCommands: make(map[domain.CommandID]context.CancelFunc), } } // Ensure Executor implements port.CommandExecutor at compile time. var _ port.CommandExecutor = (*Executor)(nil) // execSimpleTimeout is the timeout for ExecSimple single-command executions. const execSimpleTimeout = 30 * time.Second // Execute runs a command in the target pod and streams output to the handler. func (e *Executor) Execute(ctx context.Context, cmd *domain.Command, podName string, handler domain.OutputHandler) (*domain.CommandResult, error) { e.mu.RLock() namespace := e.namespace e.mu.RUnlock() // Create cancellable context for this command cmdCtx, cancel := context.WithCancel(ctx) defer cancel() // Track for potential cancellation e.activeMu.Lock() e.activeCommands[cmd.ID] = cancel e.activeMu.Unlock() defer func() { e.activeMu.Lock() delete(e.activeCommands, cmd.ID) e.activeMu.Unlock() }() startTime := time.Now() var args []string switch cmd.Type { case domain.CommandTypeClaude: // claude -p --dangerously-skip-permissions --output-format stream-json [--resume id] "prompt" // Always use stream-json so callers receive structured JSONL events including session_id. if len(cmd.Args) == 0 { return &domain.CommandResult{ CommandID: cmd.ID, ExitCode: 1, Error: fmt.Errorf("claude command requires a prompt (cmd.Args is empty)"), }, nil } args = []string{ "exec", "-n", namespace, podName, "--", "claude", "-p", "--dangerously-skip-permissions", "--output-format", "stream-json", } if cmd.ResumeSessionID != "" { args = append(args, "--resume", cmd.ResumeSessionID) } args = append(args, cmd.Args[0]) // prompt is first arg case domain.CommandTypeShell: // bash -c "command" if len(cmd.Args) == 0 { return &domain.CommandResult{ CommandID: cmd.ID, ExitCode: 1, Error: fmt.Errorf("shell command requires a command string (cmd.Args is empty)"), }, nil } args = []string{ "exec", "-n", namespace, podName, "--", "bash", "-c", cmd.Args[0], // command is first arg } case domain.CommandTypeGit: // git args = append([]string{ "exec", "-n", namespace, podName, "--", "git", "-C", "/workspace", }, cmd.Args...) default: return &domain.CommandResult{ CommandID: cmd.ID, ExitCode: 1, Error: fmt.Errorf("unknown command type: %s", cmd.Type), }, nil } // Create the kubectl command kubectl := exec.CommandContext(cmdCtx, "kubectl", args...) // Get stdout and stderr pipes stdout, err := kubectl.StdoutPipe() if err != nil { return &domain.CommandResult{ CommandID: cmd.ID, ExitCode: 1, Error: fmt.Errorf("stdout pipe: %w", err), }, nil } stderr, err := kubectl.StderrPipe() if err != nil { return &domain.CommandResult{ CommandID: cmd.ID, ExitCode: 1, Error: fmt.Errorf("stderr pipe: %w", err), }, nil } // Start the command if err := kubectl.Start(); err != nil { return &domain.CommandResult{ CommandID: cmd.ID, ExitCode: 1, Error: fmt.Errorf("start: %w", err), }, nil } // Stream output concurrently var wg sync.WaitGroup wg.Add(2) go func() { defer wg.Done() streamOutput(stdout, "stdout", handler) }() go func() { defer wg.Done() streamOutput(stderr, "stderr", handler) }() // Wait for output to be consumed wg.Wait() // Wait for command to complete err = kubectl.Wait() duration := time.Since(startTime) result := &domain.CommandResult{ CommandID: cmd.ID, DurationMs: duration.Milliseconds(), } if err != nil { if exitError, ok := err.(*exec.ExitError); ok { result.ExitCode = exitError.ExitCode() } else { result.ExitCode = 1 result.Error = err } } return result, nil } // streamOutput reads from a reader and sends each line to the handler. func streamOutput(r io.Reader, stream string, handler domain.OutputHandler) { scanner := bufio.NewScanner(r) // Increase buffer size for long lines buf := make([]byte, 0, 64*1024) scanner.Buffer(buf, 1024*1024) for scanner.Scan() { handler(domain.OutputLine{ Stream: stream, Line: scanner.Text(), Timestamp: time.Now(), }) } } // Cancel attempts to cancel a running command. func (e *Executor) Cancel(ctx context.Context, cmdID domain.CommandID) error { e.activeMu.Lock() defer e.activeMu.Unlock() cancel, exists := e.activeCommands[cmdID] if !exists { return domain.ErrCommandNotFound } cancel() return nil } // PodExists checks if a pod exists and is running. func (e *Executor) PodExists(ctx context.Context, podName string) (bool, error) { e.mu.RLock() namespace := e.namespace e.mu.RUnlock() cmd := exec.CommandContext(ctx, "kubectl", "get", "pod", podName, "-n", namespace, "-o", "jsonpath={.status.phase}", ) output, err := cmd.Output() if err != nil { // Pod doesn't exist or error return false, nil } return string(output) == "Running", nil } // CheckConnection verifies connectivity to the Kubernetes cluster. func (e *Executor) CheckConnection(ctx context.Context) error { cmd := exec.CommandContext(ctx, "kubectl", "cluster-info", "--request-timeout=5s") return cmd.Run() } // ExecSimple executes a shell command and returns the output as a string. // This is a convenience method for simple commands that don't need streaming. func (e *Executor) ExecSimple(ctx context.Context, podName, command string) (string, error) { e.mu.RLock() namespace := e.namespace e.mu.RUnlock() ctx, cancel := context.WithTimeout(ctx, execSimpleTimeout) defer cancel() args := []string{ "exec", "-n", namespace, podName, "-c", "claudebox", "--", "bash", "-c", command, } cmd := exec.CommandContext(ctx, "kubectl", args...) output, err := cmd.CombinedOutput() if err != nil { return string(output), err } return string(output), nil }