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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
|
package acp
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"path/filepath"
"sync"
"time"
"codeberg.org/turbo-editors/turbo-core/jsonrpc"
)
// HandshakeTimeout caps how long the editor waits for an agent to say hello.
// A cold agent may have a model to reach, so it is generous.
const HandshakeTimeout = 60 * time.Second
// CallTimeout caps the short calls. A prompt is deliberately not one of them:
// a turn takes as long as the model takes.
const CallTimeout = 30 * time.Second
// ErrNotReady is returned when something is asked of a session whose handshake
// has not finished.
var ErrNotReady = errors.New("acp: the agent is not ready yet")
// Options are what a session needs from the editor around it.
//
// Every callback is called from the connection's reading goroutine, so none of
// them may draw: OnUpdate exists precisely so that the event loop can come
// round and draw for itself. They are options rather than fields because Start
// begins the goroutine that calls them — the same race terminal.ViewOptions was
// created to fix.
type Options struct {
// Client is what the editor calls itself in the handshake.
Client Implementation
// OnUpdate is called whenever the conversation changed. It must only ask
// the event loop to come round again.
OnUpdate func()
// OnPermission is called when the agent wants an answer before it acts.
// It must record the request and return; the dialog belongs to the event
// loop, and Permission.Answer is how the answer gets back.
OnPermission func(*Permission)
// ReadTextFile returns a file's text as the editor sees it — from an open
// buffer when there is one, so the agent reads what you can see rather
// than what was last saved. A nil one reads from disk.
ReadTextFile func(path string) (string, error)
// WriteTextFile puts text into the editor. A nil one writes to disk.
WriteTextFile func(path, content string) error
// OnLog is given each line the agent writes to its standard error.
OnLog func(line string)
}
// Permission is the agent asking to do something, waiting for an answer.
//
// It is a value the event loop picks up, shows in a dialog, and answers. The
// reply cannot be made where the request arrived: that is the connection's
// reading goroutine, and opening a dialog belongs to the goroutine that draws.
type Permission struct {
// Title is what the tool calls itself: "Shell", "Edit file".
Title string
// Detail is the one line that says what it would actually do.
Detail string
// Options are the answers the agent will accept, in its own order.
Options []PermissionOption
request *jsonrpc.Request
once sync.Once
}
// Answer tells the agent which option was chosen.
//
// Only the first call has any effect, so a dialog that is answered and then
// torn down by its window closing cannot send two responses to one request —
// which would desynchronise an agent that matches answers to requests by id.
//
// It returns at once and writes on a goroutine of its own. The caller is the
// event loop, closing a dialog, and writing blocks until the agent reads: an
// agent that has stopped reading must not be able to freeze the editor on the
// keystroke that answers it.
//
// permission.Answer(permission.Options[0].OptionID)
func (p *Permission) Answer(optionID string) {
p.settle(PermissionOutcome{Outcome: OutcomeSelected, OptionID: optionID})
}
// Cancel tells the agent that nobody chose, which is what closing the window
// or interrupting the turn means.
//
// Cancelled, not refused: nobody declined anything, the conversation simply
// ended. The distinction is the protocol's own, and an agent that logs why it
// stopped should log the truth.
func (p *Permission) Cancel() {
p.settle(PermissionOutcome{Outcome: OutcomeCancelled})
}
// settle answers the request once, off the caller's goroutine.
func (p *Permission) settle(outcome PermissionOutcome) {
p.once.Do(func() {
go p.request.Reply(RequestPermissionResult{Outcome: outcome}, nil)
})
}
// RejectOption returns the option that means "no", preferring the agent's own
// reject_once, and falling back to the last one it offered.
//
// Guessing is better than inventing an id the agent does not know, and the
// last option is by convention the most refusing one.
func (p *Permission) RejectOption() string {
for _, option := range p.Options {
if option.Kind == OptionRejectOnce {
return option.OptionID
}
}
if len(p.Options) == 0 {
return ""
}
return p.Options[len(p.Options)-1].OptionID
}
// Session is one conversation with one agent: the child process, the
// connection, and the transcript.
//
// It is safe for concurrent use. Everything the window reads goes through a
// method that takes the lock, because the transcript is built on the reading
// goroutine and drawn on the one that owns the screen.
type Session struct {
agent Agent
options Options
process *process
conn *jsonrpc.Conn
mu sync.Mutex
transcript Transcript
sessionID string
ready bool
failed error
turn bool
queued []pending
embeds bool
usedTokens int64
sizeTokens int64
commands []Command
agentInfo Implementation
unknown int
unreadable string
stopReason string
}
// pending is a prompt typed before the handshake finished, held until it has.
type pending struct {
text string
mentions []Mention
}
// Start runs an agent and begins the handshake.
//
// It returns as soon as the process is running: the handshake needs the agent
// to answer, which may mean reaching a model, and the editor must not stop
// drawing while that happens. Prompt may be called immediately — what is typed
// before the agent is ready is held and sent when it is.
//
// session, err := acp.Start(agent, ".", acp.Options{OnUpdate: app.Wake})
// defer session.Close()
func Start(agent Agent, projectDir string, options Options) (*Session, error) {
child, err := startProcess(agent, projectDir, options.OnLog)
if err != nil {
return nil, err
}
session := NewSession(traceStream(child.stream), agent, workingDirectory(agent, projectDir), options)
session.process = child
return session, nil
}
// NewSession holds a conversation over a stream that is already open, rooted
// at cwd.
//
// It takes an io.ReadWriteCloser rather than a command, which is what lets the
// whole client be driven from a test against an agent **in the same process**,
// over net.Pipe — real framing, real concurrency, real decoding, with no
// subprocess to reap, no model to reach and no timing to get lucky with. Start
// is the thin layer that builds that stream out of a child process's pipes,
// exactly as the language server client is arranged.
//
// agent, client := net.Pipe()
// session := acp.NewSession(client, acp.Agent{Name: "test"}, ".", acp.Options{})
func NewSession(stream io.ReadWriteCloser, agent Agent, cwd string, options Options) *Session {
s := &Session{agent: agent, options: options}
s.transcript.SetAgentName(agent.Name)
s.conn = jsonrpc.NewConn(stream, Framing{}, s.onNotification, s.onRequest)
go s.run()
go s.handshake(cwd)
return s
}
// run reads from the agent until the conversation ends.
func (s *Session) run() {
err := s.conn.Run()
s.mu.Lock()
if err != nil && s.failed == nil {
s.failed = err
}
s.ready = false
s.turn = false
s.mu.Unlock()
s.wake()
}
// handshake performs the opening exchange and starts a conversation.
func (s *Session) handshake(cwd string) {
ctx, cancel := context.WithTimeout(context.Background(), HandshakeTimeout)
defer cancel()
if err := s.initialize(ctx); err != nil {
s.fail(err)
return
}
if err := s.newSession(ctx, cwd); err != nil {
s.fail(err)
return
}
s.mu.Lock()
s.ready = true
queued := s.queued
s.queued = nil
s.mu.Unlock()
s.wake()
for _, prompt := range queued {
s.send(prompt.text, prompt.mentions)
}
}
// initialize agrees a protocol version and tells the agent what this editor
// can do for it.
func (s *Session) initialize(ctx context.Context) error {
params := InitializeParams{
ProtocolVersion: ProtocolVersion,
ClientCapabilities: ClientCapabilities{
FS: FileSystemCapability{ReadTextFile: true, WriteTextFile: true},
},
ClientInfo: s.options.Client,
}
var result InitializeResult
if err := s.conn.Call(ctx, MethodInitialize, params, &result); err != nil {
return fmt.Errorf("the agent would not start a conversation: %w", err)
}
if result.ProtocolVersion != ProtocolVersion {
return fmt.Errorf("the agent speaks protocol version %d; this editor speaks %d",
result.ProtocolVersion, ProtocolVersion)
}
if len(result.AuthMethods) > 0 {
return fmt.Errorf("the agent wants to be logged in first (%s); log in with its own command and start it again",
result.AuthMethods[0].Name)
}
// What the agent calls itself is kept for the status dialog, not used as
// the label. docker agent reports "docker agent" — the name of the
// *runtime* — while the assistant inside it is called whatever its
// configuration says, and the menu entry is what the user chose. A window
// titled "Bob (llama.cpp)" whose messages are signed "docker agent" is two
// names for one thing.
s.mu.Lock()
s.agentInfo = result.AgentInfo
s.embeds = result.PromptCapabilities().EmbeddedContext
s.mu.Unlock()
return nil
}
// newSession opens a conversation rooted in a directory.
func (s *Session) newSession(ctx context.Context, cwd string) error {
absolute, err := filepath.Abs(cwd)
if err != nil {
absolute = cwd
}
var result NewSessionResult
params := NewSessionParams{Cwd: absolute, McpServers: []any{}}
if err := s.conn.Call(ctx, MethodNewSession, params, &result); err != nil {
return fmt.Errorf("the agent would not open a session: %w", err)
}
if result.SessionID == "" {
return errors.New("the agent opened a session with no id")
}
s.mu.Lock()
s.sessionID = result.SessionID
s.mu.Unlock()
return nil
}
// Prompt sends something to the agent, and records it in the conversation.
//
// It returns immediately: the turn runs on a goroutine of its own, and what
// the agent says arrives through OnUpdate. Prompting before the handshake has
// finished holds the text until it has.
//
// Each mention is a file the text names with "@"; it goes to the agent as a
// content block of its own — the file's text when the agent accepts embedded
// context, a link to it otherwise — in place of the name. The conversation
// keeps the text as typed, name included, because that is what you said.
//
// session.Prompt("what does buildMenus do?")
// session.Prompt("explain @app/menus.go", acp.Mention{Name: "app/menus.go", Path: "/src/p/app/menus.go"})
func (s *Session) Prompt(text string, mentions ...Mention) {
if text == "" {
return
}
s.mu.Lock()
s.transcript.AddUser("You", text)
ready, failed := s.ready, s.failed
if !ready && failed == nil {
s.queued = append(s.queued, pending{text: text, mentions: mentions})
}
s.mu.Unlock()
s.wake()
switch {
case failed != nil:
s.note(fmt.Sprintf("The agent is not running: %v", failed))
case ready:
go s.send(text, mentions)
}
}
// send runs one turn and records how it ended.
func (s *Session) send(text string, mentions []Mention) {
s.mu.Lock()
id, ready, embeds := s.sessionID, s.ready, s.embeds
s.turn = true
s.mu.Unlock()
if !ready {
return
}
s.wake()
params := PromptParams{SessionID: id, Prompt: blocksFor(text, mentions, embeds, s.readFile)}
// No timeout: a turn takes as long as the model takes, and cutting one off
// after an arbitrary number of seconds would look exactly like a refusal.
finished := make(chan struct{})
go s.pulse(finished)
var result PromptResult
err := s.conn.Call(context.Background(), MethodPrompt, params, &result)
close(finished)
s.mu.Lock()
s.turn = false
s.stopReason = result.StopReason
s.mu.Unlock()
switch {
case err != nil && errors.Is(err, jsonrpc.ErrClosed):
s.note("The agent stopped.")
case err != nil:
s.note(fmt.Sprintf("The turn failed: %v", err))
case result.StopReason != "" && result.StopReason != StopEndTurn:
s.note("The turn ended: " + result.StopReason)
}
s.wake()
}
// pulse wakes the event loop at the spinner's rate while a turn is running.
//
// The spinner is drawn from the clock, so something has to *cause* a redraw
// for it to move — and an agent that is thinking sends nothing for seconds at
// a time. A dropped tick cannot strand anything, which is why this is allowed
// to be a ticker at all: it only ever asks for a turn of the loop, and never
// carries a fact.
func (s *Session) pulse(finished <-chan struct{}) {
ticker := time.NewTicker(SpinnerPeriod)
defer ticker.Stop()
for {
select {
case <-finished:
return
case <-ticker.C:
s.wake()
}
}
}
// Cancel interrupts the turn in progress.
//
// It returns at once and sends on a goroutine of its own. A notification is
// nothing to wait for, and writing one blocks until the agent reads it — an
// agent that has stopped reading would otherwise freeze the editor on the very
// keystroke meant to get away from it.
func (s *Session) Cancel() {
s.mu.Lock()
id, running := s.sessionID, s.turn
s.mu.Unlock()
if id == "" || !running {
return
}
go func() { _ = s.conn.Notify(MethodCancel, CancelParams{SessionID: id}) }()
}
// Close ends the conversation and the process behind it.
//
// A session built straight onto a stream has no process; closing the
// connection closes the stream, which is all there is to end.
func (s *Session) Close() error {
if err := s.conn.Close(); err != nil && s.process == nil {
return err
}
if s.process == nil {
return nil
}
return s.process.Stop()
}
// onNotification folds an update into the conversation.
func (s *Session) onNotification(method string, params json.RawMessage) {
if method != MethodUpdate {
return
}
var notification SessionNotification
if err := json.Unmarshal(params, ¬ification); err != nil {
s.unreadableUpdate(params, err)
return
}
s.mu.Lock()
known := s.transcript.Apply(notification.Update)
switch {
case !known:
s.unknown++
case notification.Update.SessionUpdate == UpdateUsage:
s.usedTokens, s.sizeTokens = notification.Update.Used, notification.Update.Size
case notification.Update.SessionUpdate == UpdateCommands:
s.commands = notification.Update.AvailableCommands
}
s.mu.Unlock()
s.wake()
}
// unreadableUpdate records an update whose shape this client could not decode
// — a field of the wrong type, most likely — so that the status dialog can
// say so. Dropping it silently would make an agent that sent something look
// exactly like an agent that sent nothing.
func (s *Session) unreadableUpdate(params json.RawMessage, err error) {
kind := struct {
Update struct {
SessionUpdate string `json:"sessionUpdate"`
} `json:"update"`
}{}
_ = json.Unmarshal(params, &kind)
s.mu.Lock()
s.unknown++
s.unreadable = fmt.Sprintf("%s: %v", firstNonEmpty(kind.Update.SessionUpdate, "an update"), err)
s.mu.Unlock()
s.wake()
}
// onRequest answers the questions an agent asks of its client.
//
// A permission is *recorded* rather than answered: the answer comes from a
// dialog, and opening one belongs to the goroutine that draws. The file
// methods are answered on the spot, because the editor already knows.
func (s *Session) onRequest(req *jsonrpc.Request) {
switch req.Method {
case MethodRequestPermission:
s.askPermission(req)
case MethodReadTextFile:
req.Reply(s.readTextFile(req.Params))
case MethodWriteTextFile:
req.Reply(s.writeTextFile(req.Params))
default:
req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeMethodNotFound, Message: req.Method})
}
}
// askPermission hands a permission request to the editor.
//
// With nobody to ask — a session with no OnPermission — the request is
// cancelled rather than left hanging: an agent waiting for an answer that can
// never come would simply stop, with nothing said anywhere.
func (s *Session) askPermission(req *jsonrpc.Request) {
var params RequestPermissionParams
if err := json.Unmarshal(req.Params, ¶ms); err != nil {
req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeInvalidParams, Message: err.Error()})
return
}
permission := &Permission{
Title: firstNonEmpty(params.ToolCall.Title, params.ToolCall.Kind, "the agent"),
Detail: summarise(params.ToolCall.RawInput),
Options: params.Options,
request: req,
}
if s.options.OnPermission == nil {
permission.Cancel()
return
}
s.options.OnPermission(permission)
s.wake()
}
// fail records why the session will never be ready, and says so in the window.
func (s *Session) fail(err error) {
s.mu.Lock()
s.failed = err
s.queued = nil
s.mu.Unlock()
s.note(err.Error())
}
// note adds a line the editor is saying for itself.
func (s *Session) note(text string) {
s.mu.Lock()
s.transcript.AddNotice(text)
s.mu.Unlock()
s.wake()
}
// wake asks the event loop to come round, if anybody is listening.
func (s *Session) wake() {
if s.options.OnUpdate != nil {
s.options.OnUpdate()
}
}
// AddAgentTextForTest puts a message into the conversation as though the agent
// had sent it.
//
// It exists so that the window's drawing can be tested against a **fixed**
// conversation. A test that asserted on a screen while a live agent wrote to
// it would pass or fail by luck, and one such test hid a real fault in this
// project for a whole session.
func (s *Session) AddAgentTextForTest(text string) {
s.mu.Lock()
s.transcript.Apply(Update{
SessionUpdate: UpdateAgentMessage,
Content: []byte(`{"type":"text","text":` + quoteJSON(text) + `}`),
})
s.mu.Unlock()
}
// quoteJSON renders a string as a JSON string literal.
func quoteJSON(text string) string {
encoded, err := json.Marshal(text)
if err != nil {
return `""`
}
return string(encoded)
}
// Agent returns which agent this session is talking to.
func (s *Session) Agent() Agent { return s.agent }
// Entries returns a copy of the conversation so far.
func (s *Session) Entries() []Entry {
s.mu.Lock()
defer s.mu.Unlock()
return s.transcript.Entries()
}
// AgentName returns the label the agent's messages carry, which is the name
// the agents file gave it. See initialize for why the handshake's own name is
// not used here.
func (s *Session) AgentName() string {
s.mu.Lock()
defer s.mu.Unlock()
return s.transcript.AgentName()
}
// Ready reports whether the handshake has finished.
func (s *Session) Ready() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.ready
}
// Running reports whether a turn is in progress.
func (s *Session) Running() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.turn
}
// Err returns why the session stopped working, or nil.
func (s *Session) Err() error {
s.mu.Lock()
defer s.mu.Unlock()
if errors.Is(s.failed, io.EOF) {
return nil
}
return s.failed
}
// Usage returns how much of the agent's context the conversation has used, and
// how much there is. Both are zero until the agent says.
func (s *Session) Usage() (used, size int64) {
s.mu.Lock()
defer s.mu.Unlock()
return s.usedTokens, s.sizeTokens
}
// EmbedsContext reports whether the agent accepts a mentioned file's text
// inside the prompt. Before the handshake it is false, and a prompt held until
// then is built when it is sent, so the answer used is the agent's own.
func (s *Session) EmbedsContext() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.embeds
}
// Commands returns what the agent said it can be asked to do, in its order.
//
// It is what the picker lists when "/" is typed at the start of the box, and
// it changes whenever the agent sends another available_commands_update — an
// agent may add or take away commands as the conversation goes.
func (s *Session) Commands() []Command {
s.mu.Lock()
defer s.mu.Unlock()
out := make([]Command, len(s.commands))
copy(out, s.commands)
return out
}
// Unknown returns how many updates arrived that this client does not
// understand. It is in the status dialog so that "the protocol moved on" is
// visible rather than silent.
func (s *Session) Unknown() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.unknown
}
// Unreadable returns the last update this client could not decode, as "kind:
// error", or "" when every update so far was read. It is one line of the
// status dialog, and the reason TraceEnv exists.
func (s *Session) Unreadable() string {
s.mu.Lock()
defer s.mu.Unlock()
return s.unreadable
}
// AgentInfo returns what the agent called itself in the handshake: the name,
// title and version of the *program*. It is what the status dialog shows, so
// that "which build of the agent is this?" has an answer.
func (s *Session) AgentInfo() Implementation {
s.mu.Lock()
defer s.mu.Unlock()
return s.agentInfo
}
// Log returns what the agent has written to its standard error, and nothing
// for a session that has no process of its own.
func (s *Session) Log() []string {
if s.process == nil {
return nil
}
return s.process.Log()
}
|