kandev/apps/backend/internal/agentctl/server/process/interactive_runner.go

// Package process provides background process execution and output streaming for agentctl.
//
// InteractiveRunner extends the pattern from ProcessRunner to support interactive
// CLI passthrough sessions where users interact directly with agent CLIs through
// a PTY-backed terminal.

package process

import (
	"context"
	"fmt"
	"io"
	"os"
	"os/exec"
	"regexp"
	"sync"
	"syscall"
	"time"

	"github.com/creack/pty"
	"github.com/google/uuid"
	"github.com/kandev/kandev/internal/agentctl/types"
	"github.com/kandev/kandev/internal/common/logger"
	"go.uber.org/zap"
)

// InteractiveStartRequest contains parameters for starting an interactive passthrough process.
type InteractiveStartRequest struct {
	SessionID         string            `json:"session_id"`                    // Required: Agent session owning this process
	Command           []string          `json:"command"`                       // Required: Command and args to execute
	WorkingDir        string            `json:"working_dir"`                   // Working directory
	Env               map[string]string `json:"env,omitempty"`                 // Additional environment variables
	PromptPattern     string            `json:"prompt_pattern,omitempty"`      // Regex pattern to detect agent prompt for turn completion
	IdleTimeoutMs     int               `json:"idle_timeout_ms,omitempty"`     // Idle timeout in ms for turn detection
	BufferMaxBytes    int64             `json:"buffer_max_bytes,omitempty"`    // Max output buffer size
	StatusDetector    string            `json:"status_detector,omitempty"`     // Status detector type: "claude_code", "codex", ""
	CheckIntervalMs   int               `json:"check_interval_ms,omitempty"`   // How often to check state (default 100ms)
	StabilityWindowMs int               `json:"stability_window_ms,omitempty"` // State stability window (default 0)
	ImmediateStart    bool              `json:"immediate_start,omitempty"`     // Start immediately with default dimensions (don't wait for resize)
	DefaultCols       int               `json:"default_cols,omitempty"`        // Default columns if ImmediateStart (default 120)
	DefaultRows       int               `json:"default_rows,omitempty"`        // Default rows if ImmediateStart (default 40)
}

// InteractiveProcessInfo represents the state of an interactive process.
type InteractiveProcessInfo struct {
	ID         string               `json:"id"`
	SessionID  string               `json:"session_id"`
	Command    []string             `json:"command"`
	WorkingDir string               `json:"working_dir"`
	Status     types.ProcessStatus  `json:"status"`
	ExitCode   *int                 `json:"exit_code,omitempty"`
	StartedAt  time.Time            `json:"started_at"`
	UpdatedAt  time.Time            `json:"updated_at"`
	Output     []ProcessOutputChunk `json:"output,omitempty"`
}

// DirectOutputWriter is a writer that receives raw PTY output.
// When set, output bypasses the event bus and goes directly to this writer.
type DirectOutputWriter interface {
	io.Writer
	io.Closer
}

// interactiveProcess represents a running interactive PTY process.
type interactiveProcess struct {
	info   InteractiveProcessInfo
	cmd    *exec.Cmd
	ptmx   *os.File // PTY master file descriptor (using creack/pty)
	buffer *ringBuffer

	// Turn detection
	promptPattern *regexp.Regexp
	idleTimeout   time.Duration
	idleTimer     *time.Timer
	idleTimerMu   sync.Mutex

	// Status tracking (vt10x-based TUI detection)
	statusTracker *StatusTracker
	lastState     AgentState

	// Deferred start - process created lazily on first resize
	// This ensures PTY is created at exact frontend dimensions
	started       bool
	startOnce     sync.Once
	startCmd      []string
	startDir      string
	startEnv      map[string]string
	startReq      InteractiveStartRequest // Full request for deferred initialization

	// Direct output - when set, raw output goes here instead of event bus
	directOutput   DirectOutputWriter
	directOutputMu sync.RWMutex

	// Lifecycle
	stopOnce   sync.Once
	stopSignal chan struct{}
	mu         sync.Mutex
}

// TurnCompleteCallback is called when turn detection determines the agent is waiting for input.
type TurnCompleteCallback func(sessionID string)

// OutputCallback is called when process output is received.
// Used when running without a WorkspaceTracker (e.g., standalone passthrough mode).
type OutputCallback func(output *types.ProcessOutput)

// StatusCallback is called when process status changes.
// Used when running without a WorkspaceTracker (e.g., standalone passthrough mode).
type StatusCallback func(status *types.ProcessStatusUpdate)

// AgentStateCallback is called when agent TUI state changes (working, waiting, etc.).
type AgentStateCallback func(sessionID string, state AgentState)

// InteractiveRunner manages interactive PTY-based processes with stdin support.
type InteractiveRunner struct {
	logger           *logger.Logger
	workspaceTracker *WorkspaceTracker
	bufferMaxBytes   int64
	turnCompleteCallback TurnCompleteCallback
	outputCallback       OutputCallback
	statusCallback       StatusCallback
	stateCallback        AgentStateCallback

	mu        sync.RWMutex
	processes map[string]*interactiveProcess
}

// NewInteractiveRunner creates a new interactive process runner.
func NewInteractiveRunner(workspaceTracker *WorkspaceTracker, log *logger.Logger, bufferMaxBytes int64) *InteractiveRunner {
	return &InteractiveRunner{
		logger:           log.WithFields(zap.String("component", "interactive-runner")),
		workspaceTracker: workspaceTracker,
		bufferMaxBytes:   bufferMaxBytes,
		processes:        make(map[string]*interactiveProcess),
	}
}

// SetTurnCompleteCallback sets the callback to invoke when turn detection fires.
func (r *InteractiveRunner) SetTurnCompleteCallback(cb TurnCompleteCallback) {
	r.turnCompleteCallback = cb
}

// SetOutputCallback sets the callback to invoke when process output is received.
// This is used when running without a WorkspaceTracker.
func (r *InteractiveRunner) SetOutputCallback(cb OutputCallback) {
	r.outputCallback = cb
}

// SetStatusCallback sets the callback to invoke when process status changes.
// This is used when running without a WorkspaceTracker.
func (r *InteractiveRunner) SetStatusCallback(cb StatusCallback) {
	r.statusCallback = cb
}

// SetStateCallback sets the callback to invoke when agent TUI state changes.
func (r *InteractiveRunner) SetStateCallback(cb AgentStateCallback) {
	r.stateCallback = cb
}

// createStatusDetector creates the appropriate detector based on the detector type.
// The idle detector is the default - it relies on the idle timer mechanism for turn detection.
func createStatusDetector(detectorType string) StatusDetector {
	return NewIdleDetector()
}

// Start creates an interactive process entry and defers PTY creation until first resize.
// This ensures the PTY is created at exact frontend dimensions, preventing redraw issues.
func (r *InteractiveRunner) Start(ctx context.Context, req InteractiveStartRequest) (*InteractiveProcessInfo, error) {
	if req.SessionID == "" {
		return nil, fmt.Errorf("session_id is required")
	}
	if len(req.Command) == 0 {
		return nil, fmt.Errorf("command is required")
	}

	id := uuid.New().String()
	now := time.Now().UTC()

	bufferMaxBytes := req.BufferMaxBytes
	if bufferMaxBytes <= 0 {
		bufferMaxBytes = r.bufferMaxBytes
	}

	// Compile prompt pattern if provided
	var promptPattern *regexp.Regexp
	if req.PromptPattern != "" {
		var compileErr error
		promptPattern, compileErr = regexp.Compile(req.PromptPattern)
		if compileErr != nil {
			r.logger.Warn("failed to compile prompt pattern, turn detection may not work",
				zap.String("pattern", req.PromptPattern),
				zap.Error(compileErr))
		}
	}

	idleTimeout := time.Duration(req.IdleTimeoutMs) * time.Millisecond
	if idleTimeout <= 0 {
		idleTimeout = 5 * time.Second // Default 5 seconds
	}

	// Create process struct WITHOUT spawning PTY yet
	// PTY will be created on first resize when we know the exact dimensions
	proc := &interactiveProcess{
		info: InteractiveProcessInfo{
			ID:         id,
			SessionID:  req.SessionID,
			Command:    req.Command,
			WorkingDir: req.WorkingDir,
			Status:     types.ProcessStatusRunning,
			StartedAt:  now,
			UpdatedAt:  now,
		},
		buffer:        newRingBuffer(bufferMaxBytes),
		promptPattern: promptPattern,
		idleTimeout:   idleTimeout,
		lastState:     StateUnknown,
		stopSignal:    make(chan struct{}),
		// Store start parameters for deferred initialization
		started:  false,
		startCmd: req.Command,
		startDir: req.WorkingDir,
		startEnv: req.Env,
		startReq: req,
	}

	r.mu.Lock()
	r.processes[id] = proc
	r.mu.Unlock()

	// If immediate start is requested, start with default dimensions
	if req.ImmediateStart {
		cols := req.DefaultCols
		rows := req.DefaultRows
		if cols <= 0 {
			cols = 120 // Default width
		}
		if rows <= 0 {
			rows = 40 // Default height
		}

		var startErr error
		proc.startOnce.Do(func() {
			r.logger.Info("immediate start - starting process with default dimensions",
				zap.String("process_id", id),
				zap.String("session_id", req.SessionID),
				zap.Int("cols", cols),
				zap.Int("rows", rows))
			startErr = r.startProcess(proc, cols, rows)
		})
		if startErr != nil {
			r.mu.Lock()
			delete(r.processes, id)
			r.mu.Unlock()
			return nil, fmt.Errorf("failed to start process: %w", startErr)
		}

		r.logger.Info("interactive process started immediately",
			zap.String("process_id", id),
			zap.String("session_id", req.SessionID),
			zap.Strings("command", req.Command),
			zap.String("working_dir", req.WorkingDir),
		)
	} else {
		r.logger.Info("interactive process created (waiting for terminal dimensions)",
			zap.String("process_id", id),
			zap.String("session_id", req.SessionID),
			zap.Strings("command", req.Command),
			zap.String("working_dir", req.WorkingDir),
		)
	}

	r.publishStatus(proc)

	info := proc.snapshot(false)
	return &info, nil
}

// startProcess actually spawns the PTY and process. Called on first resize.
func (r *InteractiveRunner) startProcess(proc *interactiveProcess, cols, rows int) error {
	req := proc.startReq

	// Build command - use Background context so the process lives beyond the request
	// The process lifecycle is managed by Stop() and wait(), not by context cancellation
	cmd := exec.Command(proc.startCmd[0], proc.startCmd[1:]...)
	if proc.startDir != "" {
		cmd.Dir = proc.startDir
	}
	cmd.Env = mergeEnv(proc.startEnv)
	// Note: Do NOT set Setpgid when using PTY - it conflicts with terminal control
	// The PTY session handles process group management

	// Start process in PTY with exact dimensions from frontend using creack/pty
	// This is the battle-tested PTY library that properly handles resize/SIGWINCH
	ptmx, err := pty.StartWithSize(cmd, &pty.Winsize{
		Cols: uint16(cols),
		Rows: uint16(rows),
	})
	if err != nil {
		return fmt.Errorf("failed to start pty: %w", err)
	}

	// Create status tracker if a detector is configured
	var statusTracker *StatusTracker
	if req.StatusDetector != "" {
		detector := createStatusDetector(req.StatusDetector)
		config := StatusTrackerConfig{
			Rows:            rows,
			Cols:            cols,
			CheckInterval:   time.Duration(req.CheckIntervalMs) * time.Millisecond,
			StabilityWindow: time.Duration(req.StabilityWindowMs) * time.Millisecond,
		}
		if config.CheckInterval <= 0 {
			config.CheckInterval = 100 * time.Millisecond
		}
		// Create callback that will invoke the runner's state callback
		stateCallback := func(sessionID string, state AgentState) {
			if r.stateCallback != nil {
				r.stateCallback(sessionID, state)
			}
		}
		statusTracker = NewStatusTracker(req.SessionID, detector, stateCallback, config, r.logger)
		r.logger.Debug("status tracker created",
			zap.String("session_id", req.SessionID),
			zap.String("detector", req.StatusDetector))
	}

	proc.mu.Lock()
	proc.ptmx = ptmx
	proc.cmd = cmd
	proc.statusTracker = statusTracker
	proc.started = true
	proc.mu.Unlock()

	pid := 0
	if cmd.Process != nil {
		pid = cmd.Process.Pid
	}
	r.logger.Info("interactive process started at exact dimensions",
		zap.String("process_id", proc.info.ID),
		zap.String("session_id", proc.info.SessionID),
		zap.Int("cols", cols),
		zap.Int("rows", rows),
		zap.Int("pid", pid),
	)

	// Start output reading and process waiting goroutines
	go r.readOutput(proc)
	go r.wait(proc)

	return nil
}

// WriteStdin writes data to the process stdin (through PTY).
func (r *InteractiveRunner) WriteStdin(processID string, data string) error {
	proc, ok := r.get(processID)
	if !ok {
		return fmt.Errorf("process not found: %s", processID)
	}

	proc.mu.Lock()
	started := proc.started
	ptyInstance := proc.ptmx
	proc.mu.Unlock()

	if !started {
		return fmt.Errorf("process not started yet - waiting for terminal dimensions")
	}

	if ptyInstance == nil {
		return fmt.Errorf("process stdin not available")
	}

	_, err := ptyInstance.Write([]byte(data))
	if err != nil {
		return fmt.Errorf("failed to write to stdin: %w", err)
	}

	// Reset idle timer when user sends input
	r.resetIdleTimer(proc)

	return nil
}

// Stop terminates an interactive process.
func (r *InteractiveRunner) Stop(ctx context.Context, processID string) error {
	proc, ok := r.get(processID)
	if !ok {
		return fmt.Errorf("process not found: %s", processID)
	}

	// Signal output reader to exit
	proc.stopOnce.Do(func() {
		close(proc.stopSignal)
	})

	// Stop idle timer
	proc.idleTimerMu.Lock()
	if proc.idleTimer != nil {
		proc.idleTimer.Stop()
	}
	proc.idleTimerMu.Unlock()

	// Close PTY (this will cause the process to receive SIGHUP)
	proc.mu.Lock()
	if proc.ptmx != nil {
		_ = proc.ptmx.Close()
	}
	proc.mu.Unlock()

	// Terminate the process directly (PTY handles its own session management)
	if proc.cmd != nil && proc.cmd.Process != nil {
		_ = proc.cmd.Process.Signal(syscall.SIGTERM)

		// Wait for graceful exit, then force kill
		done := make(chan struct{})
		go func() {
			_ = proc.cmd.Wait()
			close(done)
		}()

		select {
		case <-ctx.Done():
			_ = proc.cmd.Process.Kill()
		case <-time.After(2 * time.Second):
			_ = proc.cmd.Process.Kill()
		case <-done:
			// Process exited cleanly
		}
	}

	return nil
}

// Get retrieves process information by ID.
func (r *InteractiveRunner) Get(id string, includeOutput bool) (*InteractiveProcessInfo, bool) {
	proc, ok := r.get(id)
	if !ok {
		return nil, false
	}
	info := proc.snapshot(includeOutput)
	return &info, true
}

// GetBySession retrieves process information by session ID.
func (r *InteractiveRunner) GetBySession(sessionID string) (*InteractiveProcessInfo, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()

	for _, proc := range r.processes {
		if proc.info.SessionID == sessionID {
			info := proc.snapshot(false)
			return &info, true
		}
	}
	return nil, false
}

// GetBuffer returns the buffered output for a process.
func (r *InteractiveRunner) GetBuffer(processID string) ([]ProcessOutputChunk, bool) {
	proc, ok := r.get(processID)
	if !ok {
		return nil, false
	}
	return proc.buffer.snapshot(), true
}

// ResizeBySession resizes the PTY for a process by session ID.
// On first resize, this triggers lazy process start at the exact frontend dimensions.
func (r *InteractiveRunner) ResizeBySession(sessionID string, cols, rows uint16) error {
	r.mu.RLock()
	var proc *interactiveProcess
	for _, p := range r.processes {
		if p.info.SessionID == sessionID {
			proc = p
			break
		}
	}
	r.mu.RUnlock()

	if proc == nil {
		return fmt.Errorf("no process found for session %s", sessionID)
	}

	// Lazy start: spawn process on first resize when we have exact dimensions
	var startErr error
	proc.startOnce.Do(func() {
		r.logger.Info("first resize received - starting process",
			zap.String("session_id", sessionID),
			zap.Uint16("cols", cols),
			zap.Uint16("rows", rows))
		startErr = r.startProcess(proc, int(cols), int(rows))
	})
	if startErr != nil {
		return fmt.Errorf("failed to start process on first resize: %w", startErr)
	}

	proc.mu.Lock()
	ptyInstance := proc.ptmx
	statusTracker := proc.statusTracker
	proc.mu.Unlock()

	// If process started, resize the PTY using creack/pty's Setsize
	// This properly sets the window size and sends SIGWINCH to the process
	if ptyInstance != nil {
		if err := pty.Setsize(ptyInstance, &pty.Winsize{
			Cols: cols,
			Rows: rows,
		}); err != nil {
			return fmt.Errorf("failed to resize PTY: %w", err)
		}
	}

	// Also resize the status tracker's virtual terminal
	if statusTracker != nil {
		statusTracker.Resize(int(cols), int(rows))
	}

	r.logger.Debug("resized PTY",
		zap.String("session_id", sessionID),
		zap.Uint16("cols", cols),
		zap.Uint16("rows", rows))

	return nil
}

func (r *InteractiveRunner) get(id string) (*interactiveProcess, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	proc, ok := r.processes[id]
	return proc, ok
}

func (r *InteractiveRunner) readOutput(proc *interactiveProcess) {
	buf := make([]byte, 32768) // 32KB buffer for better throughput
	recentOutput := ""         // Keep recent output for prompt pattern matching
	firstRead := true

	for {
		select {
		case <-proc.stopSignal:
			r.logger.Debug("readOutput: stop signal received",
				zap.String("process_id", proc.info.ID))
			return
		default:
		}

		proc.mu.Lock()
		ptyInstance := proc.ptmx
		proc.mu.Unlock()

		if ptyInstance == nil {
			r.logger.Debug("readOutput: pty is nil, exiting",
				zap.String("process_id", proc.info.ID))
			return
		}

		n, err := ptyInstance.Read(buf)
		if firstRead {
			r.logger.Info("readOutput: first read attempt",
				zap.String("process_id", proc.info.ID),
				zap.Int("bytes", n),
				zap.Error(err))
			firstRead = false
		}
		if n > 0 {
			data := buf[:n]
			dataStr := string(data)

			// Feed to status tracker for TUI state detection
			if proc.statusTracker != nil {
				proc.statusTracker.Write(data)

				// Periodically check state (debounced by ShouldCheck)
				if proc.statusTracker.ShouldCheck() {
					newState := proc.statusTracker.CheckAndUpdate()
					if newState != proc.lastState {
						proc.lastState = newState
						r.handleStateChange(proc, newState)
					}
				}
			}

			// Check if we have a direct output writer (binary WebSocket mode)
			proc.directOutputMu.RLock()
			directWriter := proc.directOutput
			proc.directOutputMu.RUnlock()

			if directWriter != nil {
				// Direct mode: write raw bytes to the WebSocket, skip event bus
				if _, writeErr := directWriter.Write(data); writeErr != nil {
					r.logger.Debug("direct output write error",
						zap.String("process_id", proc.info.ID),
						zap.Error(writeErr))
					// Don't return - the writer might have closed but process continues
				}
			} else {
				// Normal mode: buffer and stream output via event bus
				chunk := ProcessOutputChunk{
					Stream:    "stdout",
					Data:      dataStr,
					Timestamp: time.Now().UTC(),
				}
				proc.buffer.append(chunk)
				r.publishOutput(proc, chunk)
			}

			// Update recent output for prompt pattern matching (keep last 1KB)
			// Trim before appending to prevent temporary memory spikes with large outputs
			maxSize := 1024
			if len(recentOutput)+len(dataStr) > maxSize {
				// Calculate how much to keep from existing buffer
				keepFromExisting := maxSize - len(dataStr)
				if keepFromExisting < 0 {
					keepFromExisting = 0
				}
				if keepFromExisting > 0 && len(recentOutput) > keepFromExisting {
					recentOutput = recentOutput[len(recentOutput)-keepFromExisting:]
				} else if keepFromExisting == 0 {
					recentOutput = ""
				}
			}
			recentOutput += dataStr
			// Final trim in case dataStr itself was larger than maxSize
			if len(recentOutput) > maxSize {
				recentOutput = recentOutput[len(recentOutput)-maxSize:]
			}

			// Check prompt pattern for turn completion
			if proc.promptPattern != nil && proc.promptPattern.MatchString(recentOutput) {
				r.emitTurnComplete(proc)
				recentOutput = "" // Reset after match
			}

			// Reset idle timer on any output
			r.resetIdleTimer(proc)
		}
		if err != nil {
			r.logger.Debug("interactive process output read ended",
				zap.String("process_id", proc.info.ID),
				zap.Error(err))
			return
		}
	}
}

func (r *InteractiveRunner) resetIdleTimer(proc *interactiveProcess) {
	proc.idleTimerMu.Lock()
	defer proc.idleTimerMu.Unlock()

	if proc.idleTimer != nil {
		proc.idleTimer.Stop()
	}

	if proc.idleTimeout > 0 {
		proc.idleTimer = time.AfterFunc(proc.idleTimeout, func() {
			r.emitTurnComplete(proc)
		})
	}
}

func (r *InteractiveRunner) emitTurnComplete(proc *interactiveProcess) {
	if r.turnCompleteCallback != nil {
		r.turnCompleteCallback(proc.info.SessionID)
	}
	r.logger.Debug("turn complete detected",
		zap.String("process_id", proc.info.ID),
		zap.String("session_id", proc.info.SessionID))
}

// handleStateChange processes agent state changes detected by the status tracker.
func (r *InteractiveRunner) handleStateChange(proc *interactiveProcess, state AgentState) {
	r.logger.Debug("agent state changed",
		zap.String("process_id", proc.info.ID),
		zap.String("session_id", proc.info.SessionID),
		zap.String("state", string(state)))

	// WaitingInput state triggers turn complete
	if state == StateWaitingInput {
		r.emitTurnComplete(proc)
	}

	// Invoke the state callback for external handling
	if r.stateCallback != nil {
		r.stateCallback(proc.info.SessionID, state)
	}
}

// wait blocks until the process exits and then cleans up.
// Note: cmd.Wait() is intentionally blocking without a timeout. This is the correct
// behavior because:
// 1. Wait() is required to reap the process and prevent zombies
// 2. Stuck processes should be terminated via Stop() which sends SIGTERM/SIGKILL
// 3. Adding a timeout here would leave the process unreachable and create leaks
func (r *InteractiveRunner) wait(proc *interactiveProcess) {
	err := proc.cmd.Wait()
	exitCode := 0
	status := types.ProcessStatusExited
	var signalName string
	if err != nil {
		if exitErr, ok := err.(*exec.ExitError); ok {
			if waitStatus, ok := exitErr.Sys().(syscall.WaitStatus); ok {
				if waitStatus.Signaled() {
					signalName = waitStatus.Signal().String()
					exitCode = 128 + int(waitStatus.Signal())
				} else {
					exitCode = waitStatus.ExitStatus()
				}
			} else {
				exitCode = 1
			}
		} else {
			exitCode = 1
		}
		status = types.ProcessStatusFailed
	}

	r.logger.Info("interactive process exited",
		zap.String("process_id", proc.info.ID),
		zap.String("session_id", proc.info.SessionID),
		zap.String("status", string(status)),
		zap.Int("exit_code", exitCode),
		zap.String("signal", signalName),
		zap.Error(err),
	)

	// Stop idle timer
	proc.idleTimerMu.Lock()
	if proc.idleTimer != nil {
		proc.idleTimer.Stop()
	}
	proc.idleTimerMu.Unlock()

	// Update process info
	proc.mu.Lock()
	proc.info.Status = status
	proc.info.ExitCode = &exitCode
	proc.info.UpdatedAt = time.Now().UTC()
	proc.mu.Unlock()

	// Close PTY
	proc.mu.Lock()
	if proc.ptmx != nil {
		_ = proc.ptmx.Close()
		proc.ptmx = nil
	}
	proc.mu.Unlock()

	r.publishStatus(proc)

	// Remove from tracking
	r.mu.Lock()
	delete(r.processes, proc.info.ID)
	r.mu.Unlock()
}

func (r *InteractiveRunner) publishOutput(proc *interactiveProcess, chunk ProcessOutputChunk) {
	// No gating needed - process starts at exact frontend dimensions via lazy start
	proc.mu.Lock()
	info := proc.info
	proc.mu.Unlock()

	output := &types.ProcessOutput{
		SessionID: info.SessionID,
		ProcessID: info.ID,
		Kind:      types.ProcessKindAgentPassthrough,
		Stream:    chunk.Stream,
		Data:      chunk.Data,
		Timestamp: chunk.Timestamp,
	}

	// Use WorkspaceTracker if available, otherwise use callback
	if r.workspaceTracker != nil {
		r.workspaceTracker.notifyWorkspaceStreamProcessOutput(output)
	} else if r.outputCallback != nil {
		r.outputCallback(output)
	}
}

func (r *InteractiveRunner) publishStatus(proc *interactiveProcess) {
	proc.mu.Lock()
	info := proc.info
	proc.mu.Unlock()

	// Convert []string command to single string for status update
	cmdStr := ""
	if len(info.Command) > 0 {
		cmdStr = info.Command[0]
	}

	update := &types.ProcessStatusUpdate{
		SessionID:  info.SessionID,
		ProcessID:  info.ID,
		Kind:       types.ProcessKindAgentPassthrough,
		Command:    cmdStr,
		WorkingDir: info.WorkingDir,
		Status:     info.Status,
		ExitCode:   info.ExitCode,
		Timestamp:  time.Now().UTC(),
	}

	// Use WorkspaceTracker if available, otherwise use callback
	if r.workspaceTracker != nil {
		r.workspaceTracker.notifyWorkspaceStreamProcessStatus(update)
	} else if r.statusCallback != nil {
		r.statusCallback(update)
	}
}

func (p *interactiveProcess) snapshot(includeOutput bool) InteractiveProcessInfo {
	p.mu.Lock()
	defer p.mu.Unlock()
	info := p.info
	if includeOutput && p.buffer != nil {
		info.Output = p.buffer.snapshot()
	}
	return info
}

// SetDirectOutput sets a direct output writer for a process.
// When set, PTY output bypasses the event bus and goes directly to this writer.
// This is used for the dedicated binary WebSocket in passthrough mode.
// Returns error if process not found.
func (r *InteractiveRunner) SetDirectOutput(processID string, writer DirectOutputWriter) error {
	proc, ok := r.get(processID)
	if !ok {
		return fmt.Errorf("process not found: %s", processID)
	}

	proc.directOutputMu.Lock()
	proc.directOutput = writer
	proc.directOutputMu.Unlock()

	r.logger.Info("direct output set for process",
		zap.String("process_id", processID),
		zap.String("session_id", proc.info.SessionID))

	return nil
}

// ClearDirectOutput clears the direct output writer for a process.
// Output will return to the normal event bus path.
func (r *InteractiveRunner) ClearDirectOutput(processID string) error {
	proc, ok := r.get(processID)
	if !ok {
		return fmt.Errorf("process not found: %s", processID)
	}

	proc.directOutputMu.Lock()
	proc.directOutput = nil
	proc.directOutputMu.Unlock()

	r.logger.Info("direct output cleared for process",
		zap.String("process_id", processID),
		zap.String("session_id", proc.info.SessionID))

	return nil
}

// GetPtyWriter returns a writer for sending input to the PTY.
// This is used for the dedicated binary WebSocket in passthrough mode.
func (r *InteractiveRunner) GetPtyWriter(processID string) (io.Writer, error) {
	proc, ok := r.get(processID)
	if !ok {
		return nil, fmt.Errorf("process not found: %s", processID)
	}

	proc.mu.Lock()
	defer proc.mu.Unlock()

	if !proc.started {
		return nil, fmt.Errorf("process not started yet - waiting for terminal dimensions")
	}

	if proc.ptmx == nil {
		return nil, fmt.Errorf("PTY not available")
	}

	return proc.ptmx, nil
}