1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

package acp

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"sync"
)

// MaxLogLines is how much of an agent's standard error is kept for the status
// dialog. It is where a misconfigured model endpoint reports itself, and it is
// also where a chatty agent would otherwise grow without bound.
const MaxLogLines = 200

// process is a running agent: the child, its pipes as one stream, and whatever
// it has said on standard error.
type process struct {
	command *exec.Cmd
	stream  io.ReadWriteCloser

	mu   sync.Mutex
	log  []string
	over int // lines dropped past the cap
}

// pipes joins a child's stdin and stdout into the single stream jsonrpc wants.
//
// Closing it closes the input, which is how an agent is told to finish: every
// well-behaved one exits on end of input, so the process ends without a signal.
type pipes struct {
	io.ReadCloser
	io.WriteCloser
}

// Close shuts both halves, reporting the first failure.
func (p pipes) Close() error {
	writeErr := p.WriteCloser.Close()
	readErr := p.ReadCloser.Close()
	if writeErr != nil {
		return writeErr
	}
	return readErr
}

// startProcess runs an agent and returns it, with its pipes joined into one
// stream and its standard error being drained.
//
// The child's environment is the editor's with the agent's own added on top, so
// a credential already in the environment reaches the agent without being
// written into a file.
func startProcess(agent Agent, projectDir string, onLog func(string)) (*process, error) {
	command := exec.Command(agent.Command, agent.Args...)
	command.Dir = workingDirectory(agent, projectDir)
	command.Env = environment(agent)

	stdin, err := command.StdinPipe()
	if err != nil {
		return nil, fmt.Errorf("acp: connecting to %s: %w", agent.Command, err)
	}
	stdout, err := command.StdoutPipe()
	if err != nil {
		return nil, fmt.Errorf("acp: connecting to %s: %w", agent.Command, err)
	}
	stderr, err := command.StderrPipe()
	if err != nil {
		return nil, fmt.Errorf("acp: connecting to %s: %w", agent.Command, err)
	}

	if err := command.Start(); err != nil {
		return nil, fmt.Errorf("acp: starting %s: %w", agent.CommandLine(), err)
	}

	p := &process{command: command, stream: pipes{ReadCloser: stdout, WriteCloser: stdin}}
	go p.drain(stderr, onLog)
	return p, nil
}

// workingDirectory returns where the agent should run: its own cwd if it named
// one, and the project otherwise.
func workingDirectory(agent Agent, projectDir string) string {
	if agent.Cwd == "" {
		return projectDir
	}
	if filepath.IsAbs(agent.Cwd) {
		return agent.Cwd
	}
	return filepath.Join(projectDir, agent.Cwd)
}

// environment returns the child's environment: the editor's, plus the agent's.
func environment(agent Agent) []string {
	if len(agent.Env) == 0 {
		return nil // nil means "inherit", which is what os/exec does with it
	}

	env := os.Environ()
	for name, value := range agent.Env {
		env = append(env, name+"="+value)
	}
	return env
}

// drain reads the agent's standard error, keeping the last lines of it.
//
// It must be read rather than ignored: a full pipe blocks the child, and an
// agent that logs enough would stop answering with no error anywhere. It is
// also not protocol — `docker agent` prints a welcome banner here — so nothing
// read from it ever reaches the JSON-RPC layer.
func (p *process) drain(stderr io.Reader, onLog func(string)) {
	scanner := bufio.NewScanner(stderr)
	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)

	for scanner.Scan() {
		line := scanner.Text()
		p.record(line)
		if onLog != nil {
			onLog(line)
		}
	}
}

// record keeps one line of the log, dropping the oldest past the cap.
func (p *process) record(line string) {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.log = append(p.log, line)
	if len(p.log) > MaxLogLines {
		p.log = p.log[len(p.log)-MaxLogLines:]
		p.over++
	}
}

// Log returns what the agent has said on standard error, most recent last.
func (p *process) Log() []string {
	p.mu.Lock()
	defer p.mu.Unlock()

	out := make([]string, len(p.log))
	copy(out, p.log)
	return out
}

// Dropped returns how many log lines were lost to the cap.
func (p *process) Dropped() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.over
}

// Stop ends the agent.
//
// Closing the stream closes its standard input, which is how the protocol says
// to finish: an agent reading end of input exits. The process is then killed
// anyway, because an agent that ignores it must not outlive the window — and
// Wait is what stops it becoming a zombie.
func (p *process) Stop() error {
	err := p.stream.Close()
	if p.command.Process != nil {
		_ = p.command.Process.Kill()
	}
	_ = p.command.Wait()
	return err
}