turbo-editors/turbo-core · session.go

⑂ v1.0.0
session.go · 678 lines · 20.5 KBGo    Blame HistoryRaw
   1package acp
4
5	"encoding/json"
9	"path/filepath"
📦 Turbo Core   13	"rickub.com/turbo-editors/turbo-core/jsonrpc"
   14
16// HandshakeTimeout caps how long the editor waits for an agent to say hello.
17// A cold agent may have a model to reach, so it is generous.
18const HandshakeTimeout = 60 * time.Second
20// CallTimeout caps the short calls. A prompt is deliberately not one of them:
21// a turn takes as long as the model takes.
22const CallTimeout = 30 * time.Second
24// ErrNotReady is returned when something is asked of a session whose handshake
25// has not finished.
26var ErrNotReady = errors.New("acp: the agent is not ready yet")
28// Options are what a session needs from the editor around it.
30// Every callback is called from the connection's reading goroutine, so none of
31// them may draw: OnUpdate exists precisely so that the event loop can come
32// round and draw for itself. They are options rather than fields because Start
33// begins the goroutine that calls them — the same race terminal.ViewOptions was
34// created to fix.
35type Options struct {
36	// Client is what the editor calls itself in the handshake.
37	Client Implementation
39	// OnUpdate is called whenever the conversation changed. It must only ask
40	// the event loop to come round again.
41	OnUpdate func()
43	// OnPermission is called when the agent wants an answer before it acts.
44	// It must record the request and return; the dialog belongs to the event
45	// loop, and Permission.Answer is how the answer gets back.
46	OnPermission func(*Permission)
48	// ReadTextFile returns a file's text as the editor sees it — from an open
49	// buffer when there is one, so the agent reads what you can see rather
50	// than what was last saved. A nil one reads from disk.
51	ReadTextFile func(path string) (string, error)
53	// WriteTextFile puts text into the editor. A nil one writes to disk.
54	WriteTextFile func(path, content string) error
56	// OnLog is given each line the agent writes to its standard error.
57	OnLog func(line string)
60// Permission is the agent asking to do something, waiting for an answer.
62// It is a value the event loop picks up, shows in a dialog, and answers. The
63// reply cannot be made where the request arrived: that is the connection's
64// reading goroutine, and opening a dialog belongs to the goroutine that draws.
65type Permission struct {
66	// Title is what the tool calls itself: "Shell", "Edit file".
67	Title string
68	// Detail is the one line that says what it would actually do.
69	Detail string
70	// Options are the answers the agent will accept, in its own order.
71	Options []PermissionOption
73	request *jsonrpc.Request
74	once    sync.Once
77// Answer tells the agent which option was chosen.
79// Only the first call has any effect, so a dialog that is answered and then
80// torn down by its window closing cannot send two responses to one request —
81// which would desynchronise an agent that matches answers to requests by id.
83// It returns at once and writes on a goroutine of its own. The caller is the
84// event loop, closing a dialog, and writing blocks until the agent reads: an
85// agent that has stopped reading must not be able to freeze the editor on the
86// keystroke that answers it.
88//	permission.Answer(permission.Options[0].OptionID)
89func (p *Permission) Answer(optionID string) {
90	p.settle(PermissionOutcome{Outcome: OutcomeSelected, OptionID: optionID})
93// Cancel tells the agent that nobody chose, which is what closing the window
94// or interrupting the turn means.
96// Cancelled, not refused: nobody declined anything, the conversation simply
97// ended. The distinction is the protocol's own, and an agent that logs why it
98// stopped should log the truth.
99func (p *Permission) Cancel() {
100	p.settle(PermissionOutcome{Outcome: OutcomeCancelled})
103// settle answers the request once, off the caller's goroutine.
104func (p *Permission) settle(outcome PermissionOutcome) {
105	p.once.Do(func() {
106		go p.request.Reply(RequestPermissionResult{Outcome: outcome}, nil)
110// RejectOption returns the option that means "no", preferring the agent's own
111// reject_once, and falling back to the last one it offered.
113// Guessing is better than inventing an id the agent does not know, and the
114// last option is by convention the most refusing one.
115func (p *Permission) RejectOption() string {
116	for _, option := range p.Options {
117		if option.Kind == OptionRejectOnce {
118			return option.OptionID
121	if len(p.Options) == 0 {
122		return ""
124	return p.Options[len(p.Options)-1].OptionID
127// Session is one conversation with one agent: the child process, the
128// connection, and the transcript.
130// It is safe for concurrent use. Everything the window reads goes through a
131// method that takes the lock, because the transcript is built on the reading
132// goroutine and drawn on the one that owns the screen.
133type Session struct {
134	agent   Agent
135	options Options
136	process *process
137	conn    *jsonrpc.Conn
139	mu         sync.Mutex
140	transcript Transcript
141	sessionID  string
142	ready      bool
143	failed     error
144	turn       bool
145	queued     []pending
146	embeds     bool
147	usedTokens int64
148	sizeTokens int64
149	commands   []Command
150	agentInfo  Implementation
151	unknown    int
152	unreadable string
153	stopReason string
156// pending is a prompt typed before the handshake finished, held until it has.
157type pending struct {
158	text     string
159	mentions []Mention
162// Start runs an agent and begins the handshake.
164// It returns as soon as the process is running: the handshake needs the agent
165// to answer, which may mean reaching a model, and the editor must not stop
166// drawing while that happens. Prompt may be called immediately — what is typed
167// before the agent is ready is held and sent when it is.
169//	session, err := acp.Start(agent, ".", acp.Options{OnUpdate: app.Wake})
170//	defer session.Close()
171func Start(agent Agent, projectDir string, options Options) (*Session, error) {
172	child, err := startProcess(agent, projectDir, options.OnLog)
173	if err != nil {
174		return nil, err
177	session := NewSession(traceStream(child.stream), agent, workingDirectory(agent, projectDir), options)
178	session.process = child
179	return session, nil
182// NewSession holds a conversation over a stream that is already open, rooted
183
185// It takes an io.ReadWriteCloser rather than a command, which is what lets the
186// whole client be driven from a test against an agent **in the same process**,
187// over net.Pipe — real framing, real concurrency, real decoding, with no
188// subprocess to reap, no model to reach and no timing to get lucky with. Start
189// is the thin layer that builds that stream out of a child process's pipes,
190// exactly as the language server client is arranged.
192//	agent, client := net.Pipe()
193//	session := acp.NewSession(client, acp.Agent{Name: "test"}, ".", acp.Options{})
194func NewSession(stream io.ReadWriteCloser, agent Agent, cwd string, options Options) *Session {
195	s := &Session{agent: agent, options: options}
196	s.transcript.SetAgentName(agent.Name)
197	s.conn = jsonrpc.NewConn(stream, Framing{}, s.onNotification, s.onRequest)
199	go s.run()
200	go s.handshake(cwd)
201
204// run reads from the agent until the conversation ends.
205func (s *Session) run() {
206	err := s.conn.Run()
208	s.mu.Lock()
209	if err != nil && s.failed == nil {
210		s.failed = err
212	s.ready = false
213	s.turn = false
214	s.mu.Unlock()
216
219// handshake performs the opening exchange and starts a conversation.
220func (s *Session) handshake(cwd string) {
221	ctx, cancel := context.WithTimeout(context.Background(), HandshakeTimeout)
222	defer cancel()
224	if err := s.initialize(ctx); err != nil {
225		s.fail(err)
226
228	if err := s.newSession(ctx, cwd); err != nil {
229		s.fail(err)
230
233	s.mu.Lock()
234	s.ready = true
235	queued := s.queued
236	s.queued = nil
237	s.mu.Unlock()
239
240	for _, prompt := range queued {
241		s.send(prompt.text, prompt.mentions)
245// initialize agrees a protocol version and tells the agent what this editor
246// can do for it.
247func (s *Session) initialize(ctx context.Context) error {
248	params := InitializeParams{
249		ProtocolVersion: ProtocolVersion,
250		ClientCapabilities: ClientCapabilities{
251			FS: FileSystemCapability{ReadTextFile: true, WriteTextFile: true},
253		ClientInfo: s.options.Client,
256	var result InitializeResult
257	if err := s.conn.Call(ctx, MethodInitialize, params, &result); err != nil {
258		return fmt.Errorf("the agent would not start a conversation: %w", err)
260	if result.ProtocolVersion != ProtocolVersion {
261		return fmt.Errorf("the agent speaks protocol version %d; this editor speaks %d",
262			result.ProtocolVersion, ProtocolVersion)
264	if len(result.AuthMethods) > 0 {
265		return fmt.Errorf("the agent wants to be logged in first (%s); log in with its own command and start it again",
266			result.AuthMethods[0].Name)
269	// What the agent calls itself is kept for the status dialog, not used as
270	// the label. docker agent reports "docker agent" — the name of the
271	// *runtime* — while the assistant inside it is called whatever its
272	// configuration says, and the menu entry is what the user chose. A window
273	// titled "Bob (llama.cpp)" whose messages are signed "docker agent" is two
274	// names for one thing.
275	s.mu.Lock()
276	s.agentInfo = result.AgentInfo
277	s.embeds = result.PromptCapabilities().EmbeddedContext
278	s.mu.Unlock()
279	return nil
282// newSession opens a conversation rooted in a directory.
283func (s *Session) newSession(ctx context.Context, cwd string) error {
284	absolute, err := filepath.Abs(cwd)
285	if err != nil {
286		absolute = cwd
289	var result NewSessionResult
290	params := NewSessionParams{Cwd: absolute, McpServers: []any{}}
291	if err := s.conn.Call(ctx, MethodNewSession, params, &result); err != nil {
292		return fmt.Errorf("the agent would not open a session: %w", err)
294	if result.SessionID == "" {
295		return errors.New("the agent opened a session with no id")
298	s.mu.Lock()
299	s.sessionID = result.SessionID
300	s.mu.Unlock()
301	return nil
304// Prompt sends something to the agent, and records it in the conversation.
306// It returns immediately: the turn runs on a goroutine of its own, and what
307// the agent says arrives through OnUpdate. Prompting before the handshake has
308// finished holds the text until it has.
310// Each mention is a file the text names with "@"; it goes to the agent as a
311// content block of its own — the file's text when the agent accepts embedded
312// context, a link to it otherwise — in place of the name. The conversation
313// keeps the text as typed, name included, because that is what you said.
315//	session.Prompt("what does buildMenus do?")
316//	session.Prompt("explain @app/menus.go", acp.Mention{Name: "app/menus.go", Path: "/src/p/app/menus.go"})
317func (s *Session) Prompt(text string, mentions ...Mention) {
318	if text == "" {
319
322	s.mu.Lock()
323	s.transcript.AddUser("You", text)
324	ready, failed := s.ready, s.failed
325	if !ready && failed == nil {
326		s.queued = append(s.queued, pending{text: text, mentions: mentions})
328	s.mu.Unlock()
330
331
332	case failed != nil:
333		s.note(fmt.Sprintf("The agent is not running: %v", failed))
334	case ready:
335		go s.send(text, mentions)
339// send runs one turn and records how it ended.
340func (s *Session) send(text string, mentions []Mention) {
341	s.mu.Lock()
342	id, ready, embeds := s.sessionID, s.ready, s.embeds
343	s.turn = true
344	s.mu.Unlock()
346	if !ready {
347
349
351	params := PromptParams{SessionID: id, Prompt: blocksFor(text, mentions, embeds, s.readFile)}
353	// No timeout: a turn takes as long as the model takes, and cutting one off
354	// after an arbitrary number of seconds would look exactly like a refusal.
355	finished := make(chan struct{})
356	go s.pulse(finished)
358	var result PromptResult
359	err := s.conn.Call(context.Background(), MethodPrompt, params, &result)
360	close(finished)
362	s.mu.Lock()
363	s.turn = false
364	s.stopReason = result.StopReason
365	s.mu.Unlock()
367
368	case err != nil && errors.Is(err, jsonrpc.ErrClosed):
369		s.note("The agent stopped.")
370	case err != nil:
371		s.note(fmt.Sprintf("The turn failed: %v", err))
372	case result.StopReason != "" && result.StopReason != StopEndTurn:
373		s.note("The turn ended: " + result.StopReason)
375
378// pulse wakes the event loop at the spinner's rate while a turn is running.
380// The spinner is drawn from the clock, so something has to *cause* a redraw
381// for it to move — and an agent that is thinking sends nothing for seconds at
382// a time. A dropped tick cannot strand anything, which is why this is allowed
383// to be a ticker at all: it only ever asks for a turn of the loop, and never
384// carries a fact.
385func (s *Session) pulse(finished <-chan struct{}) {
386	ticker := time.NewTicker(SpinnerPeriod)
387	defer ticker.Stop()
390
391		case <-finished:
392
393		case <-ticker.C:
394			s.wake()
399// Cancel interrupts the turn in progress.
401// It returns at once and sends on a goroutine of its own. A notification is
402// nothing to wait for, and writing one blocks until the agent reads it — an
403// agent that has stopped reading would otherwise freeze the editor on the very
404// keystroke meant to get away from it.
405func (s *Session) Cancel() {
406	s.mu.Lock()
407	id, running := s.sessionID, s.turn
408	s.mu.Unlock()
410	if id == "" || !running {
411
413	go func() { _ = s.conn.Notify(MethodCancel, CancelParams{SessionID: id}) }()
416// Close ends the conversation and the process behind it.
418// A session built straight onto a stream has no process; closing the
419// connection closes the stream, which is all there is to end.
420func (s *Session) Close() error {
421	if err := s.conn.Close(); err != nil && s.process == nil {
422		return err
424	if s.process == nil {
425		return nil
427	return s.process.Stop()
430// onNotification folds an update into the conversation.
431func (s *Session) onNotification(method string, params json.RawMessage) {
432	if method != MethodUpdate {
433
436	var notification SessionNotification
437	if err := json.Unmarshal(params, &notification); err != nil {
438		s.unreadableUpdate(params, err)
439
442	s.mu.Lock()
443	known := s.transcript.Apply(notification.Update)
444
445	case !known:
446		s.unknown++
447	case notification.Update.SessionUpdate == UpdateUsage:
448		s.usedTokens, s.sizeTokens = notification.Update.Used, notification.Update.Size
449	case notification.Update.SessionUpdate == UpdateCommands:
450		s.commands = notification.Update.AvailableCommands
452	s.mu.Unlock()
454
457// unreadableUpdate records an update whose shape this client could not decode
458// — a field of the wrong type, most likely — so that the status dialog can
459// say so. Dropping it silently would make an agent that sent something look
460// exactly like an agent that sent nothing.
461func (s *Session) unreadableUpdate(params json.RawMessage, err error) {
462	kind := struct {
463		Update struct {
464			SessionUpdate string `json:"sessionUpdate"`
465		} `json:"update"`
467	_ = json.Unmarshal(params, &kind)
469	s.mu.Lock()
470	s.unknown++
471	s.unreadable = fmt.Sprintf("%s: %v", firstNonEmpty(kind.Update.SessionUpdate, "an update"), err)
472	s.mu.Unlock()
473
476// onRequest answers the questions an agent asks of its client.
478// A permission is *recorded* rather than answered: the answer comes from a
479// dialog, and opening one belongs to the goroutine that draws. The file
480// methods are answered on the spot, because the editor already knows.
481func (s *Session) onRequest(req *jsonrpc.Request) {
482	switch req.Method {
483	case MethodRequestPermission:
484		s.askPermission(req)
485	case MethodReadTextFile:
486		req.Reply(s.readTextFile(req.Params))
487	case MethodWriteTextFile:
488		req.Reply(s.writeTextFile(req.Params))
489
490		req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeMethodNotFound, Message: req.Method})
494// askPermission hands a permission request to the editor.
496// With nobody to ask — a session with no OnPermission — the request is
497// cancelled rather than left hanging: an agent waiting for an answer that can
498// never come would simply stop, with nothing said anywhere.
499func (s *Session) askPermission(req *jsonrpc.Request) {
500	var params RequestPermissionParams
501	if err := json.Unmarshal(req.Params, &params); err != nil {
502		req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeInvalidParams, Message: err.Error()})
503
506	permission := &Permission{
507		Title:   firstNonEmpty(params.ToolCall.Title, params.ToolCall.Kind, "the agent"),
508		Detail:  summarise(params.ToolCall.RawInput),
509		Options: params.Options,
510		request: req,
512	if s.options.OnPermission == nil {
513		permission.Cancel()
514
516	s.options.OnPermission(permission)
517
520// fail records why the session will never be ready, and says so in the window.
521func (s *Session) fail(err error) {
522	s.mu.Lock()
523	s.failed = err
524	s.queued = nil
525	s.mu.Unlock()
527	s.note(err.Error())
530// note adds a line the editor is saying for itself.
531func (s *Session) note(text string) {
532	s.mu.Lock()
533	s.transcript.AddNotice(text)
534	s.mu.Unlock()
535
538// wake asks the event loop to come round, if anybody is listening.
539func (s *Session) wake() {
540	if s.options.OnUpdate != nil {
541		s.options.OnUpdate()
545// AddAgentTextForTest puts a message into the conversation as though the agent
546// had sent it.
548// It exists so that the window's drawing can be tested against a **fixed**
549// conversation. A test that asserted on a screen while a live agent wrote to
550// it would pass or fail by luck, and one such test hid a real fault in this
551// project for a whole session.
552func (s *Session) AddAgentTextForTest(text string) {
553	s.mu.Lock()
554	s.transcript.Apply(Update{
555		SessionUpdate: UpdateAgentMessage,
556		Content:       []byte(`{"type":"text","text":` + quoteJSON(text) + `}`),
558	s.mu.Unlock()
561// quoteJSON renders a string as a JSON string literal.
562func quoteJSON(text string) string {
563	encoded, err := json.Marshal(text)
564	if err != nil {
565		return `""`
567	return string(encoded)
570// Agent returns which agent this session is talking to.
571func (s *Session) Agent() Agent { return s.agent }
573// Entries returns a copy of the conversation so far.
574func (s *Session) Entries() []Entry {
575	s.mu.Lock()
576	defer s.mu.Unlock()
577	return s.transcript.Entries()
580// AgentName returns the label the agent's messages carry, which is the name
581// the agents file gave it. See initialize for why the handshake's own name is
582// not used here.
583func (s *Session) AgentName() string {
584	s.mu.Lock()
585	defer s.mu.Unlock()
586	return s.transcript.AgentName()
589// Ready reports whether the handshake has finished.
590func (s *Session) Ready() bool {
591	s.mu.Lock()
592	defer s.mu.Unlock()
593	return s.ready
596// Running reports whether a turn is in progress.
597func (s *Session) Running() bool {
598	s.mu.Lock()
599	defer s.mu.Unlock()
600	return s.turn
603// Err returns why the session stopped working, or nil.
604func (s *Session) Err() error {
605	s.mu.Lock()
606	defer s.mu.Unlock()
607	if errors.Is(s.failed, io.EOF) {
608		return nil
610	return s.failed
613// Usage returns how much of the agent's context the conversation has used, and
614// how much there is. Both are zero until the agent says.
615func (s *Session) Usage() (used, size int64) {
616	s.mu.Lock()
617	defer s.mu.Unlock()
618	return s.usedTokens, s.sizeTokens
621// EmbedsContext reports whether the agent accepts a mentioned file's text
622// inside the prompt. Before the handshake it is false, and a prompt held until
623// then is built when it is sent, so the answer used is the agent's own.
624func (s *Session) EmbedsContext() bool {
625	s.mu.Lock()
626	defer s.mu.Unlock()
627	return s.embeds
630// Commands returns what the agent said it can be asked to do, in its order.
632// It is what the picker lists when "/" is typed at the start of the box, and
633// it changes whenever the agent sends another available_commands_update — an
634// agent may add or take away commands as the conversation goes.
635func (s *Session) Commands() []Command {
636	s.mu.Lock()
637	defer s.mu.Unlock()
639	out := make([]Command, len(s.commands))
640	copy(out, s.commands)
641	return out
644// Unknown returns how many updates arrived that this client does not
645// understand. It is in the status dialog so that "the protocol moved on" is
646// visible rather than silent.
647func (s *Session) Unknown() int {
648	s.mu.Lock()
649	defer s.mu.Unlock()
650	return s.unknown
653// Unreadable returns the last update this client could not decode, as "kind:
654// error", or "" when every update so far was read. It is one line of the
655// status dialog, and the reason TraceEnv exists.
656func (s *Session) Unreadable() string {
657	s.mu.Lock()
658	defer s.mu.Unlock()
659	return s.unreadable
662// AgentInfo returns what the agent called itself in the handshake: the name,
663// title and version of the *program*. It is what the status dialog shows, so
664// that "which build of the agent is this?" has an answer.
665func (s *Session) AgentInfo() Implementation {
666	s.mu.Lock()
667	defer s.mu.Unlock()
668	return s.agentInfo
671// Log returns what the agent has written to its standard error, and nothing
672// for a session that has no process of its own.
673func (s *Session) Log() []string {
674	if s.process == nil {
675		return nil
677	return s.process.Log()