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
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201

// Package agent runs an ACP (Agent Client Protocol) code agent as a
// subprocess and exposes its session lifecycle to the rest of ori.
//
// It is agent-agnostic: any executable speaking ACP on stdio works (the
// Claude Code adapter, Gemini CLI, a test mock, ...). Events flowing from the
// agent (streamed updates, permission requests) are delivered to a Handler
// supplied by the caller.
package agent

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/exec"

	acp "github.com/coder/acp-go-sdk"
)

// Handler receives the events an ACP agent sends during a session.
//
// HandleSessionUpdate is called for every streamed update (message chunks,
// tool calls, plan changes, ...). HandlePermissionRequest must block until a
// decision is available and return it; returning an error rejects the
// request.
type Handler interface {
	HandleSessionUpdate(ctx context.Context, notification acp.SessionNotification)
	HandlePermissionRequest(ctx context.Context, request acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error)
}

// Options configure Start and Connect.
type Options struct {
	// Command is the agent command line: executable first, then arguments.
	// Only used by Start.
	Command []string
	// Cwd is the absolute working directory given to the ACP session.
	Cwd string
	// Handler receives session updates and permission requests. Required.
	Handler Handler
	// Logger receives connection diagnostics. Optional.
	Logger *slog.Logger
	// Stderr receives the agent process's stderr. Defaults to os.Stderr.
	// Only used by Start.
	Stderr io.Writer
}

// Session is a live conversation with a running ACP agent.
type Session struct {
	id       acp.SessionId
	conn     *acp.ClientSideConnection
	initResp acp.InitializeResponse
	shutdown func() error
}

// ID returns the ACP session identifier.
func (s *Session) ID() string { return string(s.id) }

// InitializeResult returns the capabilities the agent advertised during the
// ACP initialize handshake.
func (s *Session) InitializeResult() acp.InitializeResponse { return s.initResp }

// PromptText sends a plain-text user prompt and blocks until the turn ends,
// while updates stream to the Handler. It returns the reason the turn
// stopped.
//
// Example:
//
//	stop, err := session.PromptText(ctx, "Explain this repository")
//	if err == nil && stop == acp.StopReasonEndTurn {
//		fmt.Println("turn finished")
//	}
func (s *Session) PromptText(ctx context.Context, text string) (acp.StopReason, error) {
	return s.Prompt(ctx, []acp.ContentBlock{acp.TextBlock(text)})
}

// Prompt sends a user prompt made of arbitrary content blocks and blocks
// until the turn ends. Updates stream to the Handler while it runs.
func (s *Session) Prompt(ctx context.Context, blocks []acp.ContentBlock) (acp.StopReason, error) {
	resp, err := s.conn.Prompt(ctx, acp.PromptRequest{SessionId: s.id, Prompt: blocks})
	if err != nil {
		return "", fmt.Errorf("prompt: %w", err)
	}
	return resp.StopReason, nil
}

// Cancel asks the agent to stop the current turn. The pending Prompt call
// then returns with acp.StopReasonCancelled.
func (s *Session) Cancel(ctx context.Context) error {
	if err := s.conn.Cancel(ctx, acp.CancelNotification{SessionId: s.id}); err != nil {
		return fmt.Errorf("cancel: %w", err)
	}
	return nil
}

// Done closes when the agent disconnects, whatever the cause.
func (s *Session) Done() <-chan struct{} { return s.conn.Done() }

// Close terminates the agent. It is safe to call more than once.
func (s *Session) Close() error {
	if s.shutdown == nil {
		return nil
	}
	shutdown := s.shutdown
	s.shutdown = nil
	return shutdown()
}

// Connect performs the ACP handshake (initialize + session/new) over an
// existing transport: in writes to the agent, out reads from it. It is the
// transport-independent core of Start and what tests use with in-memory
// pipes.
func Connect(ctx context.Context, in io.Writer, out io.Reader, opts Options) (*Session, error) {
	if opts.Handler == nil {
		return nil, errors.New("agent: Options.Handler is required")
	}

	conn := acp.NewClientSideConnection(&client{handler: opts.Handler}, in, out)
	if opts.Logger != nil {
		conn.SetLogger(opts.Logger)
	}

	initResp, err := conn.Initialize(ctx, acp.InitializeRequest{
		ProtocolVersion: acp.ProtocolVersionNumber,
		ClientCapabilities: acp.ClientCapabilities{
			Fs: acp.FileSystemCapabilities{ReadTextFile: true, WriteTextFile: true},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("initialize: %w", err)
	}

	newResp, err := conn.NewSession(ctx, acp.NewSessionRequest{Cwd: opts.Cwd, McpServers: []acp.McpServer{}})
	if err != nil {
		return nil, fmt.Errorf("new session: %w", err)
	}

	return &Session{id: newResp.SessionId, conn: conn, initResp: initResp}, nil
}

// Start spawns opts.Command as a subprocess speaking ACP on its stdio, then
// performs the handshake. Closing the returned Session kills the process.
//
// Example:
//
//	session, err := agent.Start(ctx, agent.Options{
//		Command: []string{"npx", "-y", "@agentclientprotocol/claude-agent-acp"},
//		Cwd:     "/work/my-project",
//		Handler: myHandler,
//	})
func Start(ctx context.Context, opts Options) (*Session, error) {
	if len(opts.Command) == 0 {
		return nil, errors.New("agent: Options.Command is required")
	}

	cmd, stdin, stdout, err := spawnProcess(opts)
	if err != nil {
		return nil, err
	}

	session, err := Connect(ctx, stdin, stdout, opts)
	if err != nil {
		_ = killProcess(cmd)
		return nil, err
	}
	session.shutdown = func() error { return killProcess(cmd) }
	return session, nil
}

// killProcess terminates the agent process and reaps it; the "killed" error
// from Wait is the expected outcome, not a failure.
func killProcess(cmd *exec.Cmd) error {
	if err := cmd.Process.Kill(); err != nil {
		return err
	}
	_ = cmd.Wait()
	return nil
}

// spawnProcess launches the agent command with its stdio wired for ACP.
func spawnProcess(opts Options) (*exec.Cmd, io.WriteCloser, io.ReadCloser, error) {
	cmd := exec.Command(opts.Command[0], opts.Command[1:]...)
	cmd.Dir = opts.Cwd
	cmd.Stderr = opts.Stderr
	if cmd.Stderr == nil {
		cmd.Stderr = os.Stderr
	}
	stdin, err := cmd.StdinPipe()
	if err != nil {
		return nil, nil, nil, fmt.Errorf("agent stdin: %w", err)
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, nil, nil, fmt.Errorf("agent stdout: %w", err)
	}
	if err := cmd.Start(); err != nil {
		return nil, nil, nil, fmt.Errorf("start agent %q: %w", opts.Command[0], err)
	}
	return cmd, stdin, stdout, nil
}