turbo-editors/turbo-core · session.go

⑂ v1.0.3
session.go · 692 lines · 21.3 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
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   159	pastes   []Paste
   160	mentions []Mention
163// Start runs an agent and begins the handshake.
165// It returns as soon as the process is running: the handshake needs the agent
166// to answer, which may mean reaching a model, and the editor must not stop
167// drawing while that happens. Prompt may be called immediately — what is typed
168// before the agent is ready is held and sent when it is.
170//	session, err := acp.Start(agent, ".", acp.Options{OnUpdate: app.Wake})
171//	defer session.Close()
172func Start(agent Agent, projectDir string, options Options) (*Session, error) {
173	child, err := startProcess(agent, projectDir, options.OnLog)
174	if err != nil {
175		return nil, err
178	session := NewSession(traceStream(child.stream), agent, workingDirectory(agent, projectDir), options)
179	session.process = child
180	return session, nil
183// NewSession holds a conversation over a stream that is already open, rooted
184
186// It takes an io.ReadWriteCloser rather than a command, which is what lets the
187// whole client be driven from a test against an agent **in the same process**,
188// over net.Pipe — real framing, real concurrency, real decoding, with no
189// subprocess to reap, no model to reach and no timing to get lucky with. Start
190// is the thin layer that builds that stream out of a child process's pipes,
191// exactly as the language server client is arranged.
193//	agent, client := net.Pipe()
194//	session := acp.NewSession(client, acp.Agent{Name: "test"}, ".", acp.Options{})
195func NewSession(stream io.ReadWriteCloser, agent Agent, cwd string, options Options) *Session {
196	s := &Session{agent: agent, options: options}
197	s.transcript.SetAgentName(agent.Name)
198	s.conn = jsonrpc.NewConn(stream, Framing{}, s.onNotification, s.onRequest)
200	go s.run()
201	go s.handshake(cwd)
202
205// run reads from the agent until the conversation ends.
206func (s *Session) run() {
207	err := s.conn.Run()
209	s.mu.Lock()
210	if err != nil && s.failed == nil {
211		s.failed = err
213	s.ready = false
214	s.turn = false
215	s.mu.Unlock()
217
220// handshake performs the opening exchange and starts a conversation.
221func (s *Session) handshake(cwd string) {
222	ctx, cancel := context.WithTimeout(context.Background(), HandshakeTimeout)
223	defer cancel()
225	if err := s.initialize(ctx); err != nil {
226		s.fail(err)
227
229	if err := s.newSession(ctx, cwd); err != nil {
230		s.fail(err)
231
234	s.mu.Lock()
235	s.ready = true
236	queued := s.queued
237	s.queued = nil
238	s.mu.Unlock()
240
241	for _, prompt := range queued {
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   242		s.send(prompt.text, prompt.pastes, prompt.mentions)
   243
246// initialize agrees a protocol version and tells the agent what this editor
247// can do for it.
248func (s *Session) initialize(ctx context.Context) error {
249	params := InitializeParams{
250		ProtocolVersion: ProtocolVersion,
251		ClientCapabilities: ClientCapabilities{
252			FS: FileSystemCapability{ReadTextFile: true, WriteTextFile: true},
254		ClientInfo: s.options.Client,
257	var result InitializeResult
258	if err := s.conn.Call(ctx, MethodInitialize, params, &result); err != nil {
259		return fmt.Errorf("the agent would not start a conversation: %w", err)
261	if result.ProtocolVersion != ProtocolVersion {
262		return fmt.Errorf("the agent speaks protocol version %d; this editor speaks %d",
263			result.ProtocolVersion, ProtocolVersion)
265	if len(result.AuthMethods) > 0 {
266		return fmt.Errorf("the agent wants to be logged in first (%s); log in with its own command and start it again",
267			result.AuthMethods[0].Name)
270	// What the agent calls itself is kept for the status dialog, not used as
271	// the label. docker agent reports "docker agent" — the name of the
272	// *runtime* — while the assistant inside it is called whatever its
273	// configuration says, and the menu entry is what the user chose. A window
274	// titled "Bob (llama.cpp)" whose messages are signed "docker agent" is two
275	// names for one thing.
276	s.mu.Lock()
277	s.agentInfo = result.AgentInfo
278	s.embeds = result.PromptCapabilities().EmbeddedContext
279	s.mu.Unlock()
280	return nil
283// newSession opens a conversation rooted in a directory.
284func (s *Session) newSession(ctx context.Context, cwd string) error {
285	absolute, err := filepath.Abs(cwd)
286	if err != nil {
287		absolute = cwd
290	var result NewSessionResult
291	params := NewSessionParams{Cwd: absolute, McpServers: []any{}}
292	if err := s.conn.Call(ctx, MethodNewSession, params, &result); err != nil {
293		return fmt.Errorf("the agent would not open a session: %w", err)
295	if result.SessionID == "" {
296		return errors.New("the agent opened a session with no id")
299	s.mu.Lock()
300	s.sessionID = result.SessionID
301	s.mu.Unlock()
302	return nil
305// Prompt sends something to the agent, and records it in the conversation.
307// It returns immediately: the turn runs on a goroutine of its own, and what
308// the agent says arrives through OnUpdate. Prompting before the handshake has
309// finished holds the text until it has.
311// Each mention is a file the text names with "@"; it goes to the agent as a
312// content block of its own — the file's text when the agent accepts embedded
313// context, a link to it otherwise — in place of the name. The conversation
314// keeps the text as typed, name included, because that is what you said.
316//	session.Prompt("what does buildMenus do?")
317//	session.Prompt("explain @app/menus.go", acp.Mention{Name: "app/menus.go", Path: "/src/p/app/menus.go"})
318func (s *Session) Prompt(text string, mentions ...Mention) {
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   319	s.PromptWith(text, nil, mentions...)
322// PromptWith is Prompt for a text in which pasted passages stand as tokens.
324// Each paste's Token is replaced by its Text in what the agent receives —
325// byte for byte, and only inside the text blocks, so a mention is never read
326// out of a paste. The conversation keeps the text with the tokens in it: that
327// is what makes the exchange readable afterwards, which is half the point of
328// keeping a paste aside at all.
330//	session.PromptWith("here is the trace "+paste.Token+" what is failing?", []acp.Paste{paste})
331func (s *Session) PromptWith(text string, pastes []Paste, mentions ...Mention) {
   332	if text == "" {
333
336	s.mu.Lock()
337	s.transcript.AddUser("You", text)
338	ready, failed := s.ready, s.failed
339	if !ready && failed == nil {
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   340		s.queued = append(s.queued, pending{text: text, pastes: pastes, mentions: mentions})
   341
342	s.mu.Unlock()
344
345
346	case failed != nil:
347		s.note(fmt.Sprintf("The agent is not running: %v", failed))
348	case ready:
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   349		go s.send(text, pastes, mentions)
   350
353// send runs one turn and records how it ended.
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   354func (s *Session) send(text string, pastes []Paste, mentions []Mention) {
   355	s.mu.Lock()
356	id, ready, embeds := s.sessionID, s.ready, s.embeds
357	s.turn = true
358	s.mu.Unlock()
360	if !ready {
361
363
📦 Turbo Core — files rewritten outside reload into their windows; the agent window wraps what you type and keeps a long paste aside as a token (#28); configrepo fetches a shared .turbo-<slug> from a forge URL (#25)   365	params := PromptParams{SessionID: id, Prompt: expandPastes(blocksFor(text, mentions, embeds, s.readFile), pastes)}
   366
367	// No timeout: a turn takes as long as the model takes, and cutting one off
368	// after an arbitrary number of seconds would look exactly like a refusal.
369	finished := make(chan struct{})
370	go s.pulse(finished)
372	var result PromptResult
373	err := s.conn.Call(context.Background(), MethodPrompt, params, &result)
374	close(finished)
376	s.mu.Lock()
377	s.turn = false
378	s.stopReason = result.StopReason
379	s.mu.Unlock()
381
382	case err != nil && errors.Is(err, jsonrpc.ErrClosed):
383		s.note("The agent stopped.")
384	case err != nil:
385		s.note(fmt.Sprintf("The turn failed: %v", err))
386	case result.StopReason != "" && result.StopReason != StopEndTurn:
387		s.note("The turn ended: " + result.StopReason)
389
392// pulse wakes the event loop at the spinner's rate while a turn is running.
394// The spinner is drawn from the clock, so something has to *cause* a redraw
395// for it to move — and an agent that is thinking sends nothing for seconds at
396// a time. A dropped tick cannot strand anything, which is why this is allowed
397// to be a ticker at all: it only ever asks for a turn of the loop, and never
398// carries a fact.
399func (s *Session) pulse(finished <-chan struct{}) {
400	ticker := time.NewTicker(SpinnerPeriod)
401	defer ticker.Stop()
404
405		case <-finished:
406
407		case <-ticker.C:
408			s.wake()
413// Cancel interrupts the turn in progress.
415// It returns at once and sends on a goroutine of its own. A notification is
416// nothing to wait for, and writing one blocks until the agent reads it — an
417// agent that has stopped reading would otherwise freeze the editor on the very
418// keystroke meant to get away from it.
419func (s *Session) Cancel() {
420	s.mu.Lock()
421	id, running := s.sessionID, s.turn
422	s.mu.Unlock()
424	if id == "" || !running {
425
427	go func() { _ = s.conn.Notify(MethodCancel, CancelParams{SessionID: id}) }()
430// Close ends the conversation and the process behind it.
432// A session built straight onto a stream has no process; closing the
433// connection closes the stream, which is all there is to end.
434func (s *Session) Close() error {
435	if err := s.conn.Close(); err != nil && s.process == nil {
436		return err
438	if s.process == nil {
439		return nil
441	return s.process.Stop()
444// onNotification folds an update into the conversation.
445func (s *Session) onNotification(method string, params json.RawMessage) {
446	if method != MethodUpdate {
447
450	var notification SessionNotification
451	if err := json.Unmarshal(params, &notification); err != nil {
452		s.unreadableUpdate(params, err)
453
456	s.mu.Lock()
457	known := s.transcript.Apply(notification.Update)
458
459	case !known:
460		s.unknown++
461	case notification.Update.SessionUpdate == UpdateUsage:
462		s.usedTokens, s.sizeTokens = notification.Update.Used, notification.Update.Size
463	case notification.Update.SessionUpdate == UpdateCommands:
464		s.commands = notification.Update.AvailableCommands
466	s.mu.Unlock()
468
471// unreadableUpdate records an update whose shape this client could not decode
472// — a field of the wrong type, most likely — so that the status dialog can
473// say so. Dropping it silently would make an agent that sent something look
474// exactly like an agent that sent nothing.
475func (s *Session) unreadableUpdate(params json.RawMessage, err error) {
476	kind := struct {
477		Update struct {
478			SessionUpdate string `json:"sessionUpdate"`
479		} `json:"update"`
481	_ = json.Unmarshal(params, &kind)
483	s.mu.Lock()
484	s.unknown++
485	s.unreadable = fmt.Sprintf("%s: %v", firstNonEmpty(kind.Update.SessionUpdate, "an update"), err)
486	s.mu.Unlock()
487
490// onRequest answers the questions an agent asks of its client.
492// A permission is *recorded* rather than answered: the answer comes from a
493// dialog, and opening one belongs to the goroutine that draws. The file
494// methods are answered on the spot, because the editor already knows.
495func (s *Session) onRequest(req *jsonrpc.Request) {
496	switch req.Method {
497	case MethodRequestPermission:
498		s.askPermission(req)
499	case MethodReadTextFile:
500		req.Reply(s.readTextFile(req.Params))
501	case MethodWriteTextFile:
502		req.Reply(s.writeTextFile(req.Params))
503
504		req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeMethodNotFound, Message: req.Method})
508// askPermission hands a permission request to the editor.
510// With nobody to ask — a session with no OnPermission — the request is
511// cancelled rather than left hanging: an agent waiting for an answer that can
512// never come would simply stop, with nothing said anywhere.
513func (s *Session) askPermission(req *jsonrpc.Request) {
514	var params RequestPermissionParams
515	if err := json.Unmarshal(req.Params, &params); err != nil {
516		req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeInvalidParams, Message: err.Error()})
517
520	permission := &Permission{
521		Title:   firstNonEmpty(params.ToolCall.Title, params.ToolCall.Kind, "the agent"),
522		Detail:  summarise(params.ToolCall.RawInput),
523		Options: params.Options,
524		request: req,
526	if s.options.OnPermission == nil {
527		permission.Cancel()
528
530	s.options.OnPermission(permission)
531
534// fail records why the session will never be ready, and says so in the window.
535func (s *Session) fail(err error) {
536	s.mu.Lock()
537	s.failed = err
538	s.queued = nil
539	s.mu.Unlock()
541	s.note(err.Error())
544// note adds a line the editor is saying for itself.
545func (s *Session) note(text string) {
546	s.mu.Lock()
547	s.transcript.AddNotice(text)
548	s.mu.Unlock()
549
552// wake asks the event loop to come round, if anybody is listening.
553func (s *Session) wake() {
554	if s.options.OnUpdate != nil {
555		s.options.OnUpdate()
559// AddAgentTextForTest puts a message into the conversation as though the agent
560// had sent it.
562// It exists so that the window's drawing can be tested against a **fixed**
563// conversation. A test that asserted on a screen while a live agent wrote to
564// it would pass or fail by luck, and one such test hid a real fault in this
565// project for a whole session.
566func (s *Session) AddAgentTextForTest(text string) {
567	s.mu.Lock()
568	s.transcript.Apply(Update{
569		SessionUpdate: UpdateAgentMessage,
570		Content:       []byte(`{"type":"text","text":` + quoteJSON(text) + `}`),
572	s.mu.Unlock()
575// quoteJSON renders a string as a JSON string literal.
576func quoteJSON(text string) string {
577	encoded, err := json.Marshal(text)
578	if err != nil {
579		return `""`
581	return string(encoded)
584// Agent returns which agent this session is talking to.
585func (s *Session) Agent() Agent { return s.agent }
587// Entries returns a copy of the conversation so far.
588func (s *Session) Entries() []Entry {
589	s.mu.Lock()
590	defer s.mu.Unlock()
591	return s.transcript.Entries()
594// AgentName returns the label the agent's messages carry, which is the name
595// the agents file gave it. See initialize for why the handshake's own name is
596// not used here.
597func (s *Session) AgentName() string {
598	s.mu.Lock()
599	defer s.mu.Unlock()
600	return s.transcript.AgentName()
603// Ready reports whether the handshake has finished.
604func (s *Session) Ready() bool {
605	s.mu.Lock()
606	defer s.mu.Unlock()
607	return s.ready
610// Running reports whether a turn is in progress.
611func (s *Session) Running() bool {
612	s.mu.Lock()
613	defer s.mu.Unlock()
614	return s.turn
617// Err returns why the session stopped working, or nil.
618func (s *Session) Err() error {
619	s.mu.Lock()
620	defer s.mu.Unlock()
621	if errors.Is(s.failed, io.EOF) {
622		return nil
624	return s.failed
627// Usage returns how much of the agent's context the conversation has used, and
628// how much there is. Both are zero until the agent says.
629func (s *Session) Usage() (used, size int64) {
630	s.mu.Lock()
631	defer s.mu.Unlock()
632	return s.usedTokens, s.sizeTokens
635// EmbedsContext reports whether the agent accepts a mentioned file's text
636// inside the prompt. Before the handshake it is false, and a prompt held until
637// then is built when it is sent, so the answer used is the agent's own.
638func (s *Session) EmbedsContext() bool {
639	s.mu.Lock()
640	defer s.mu.Unlock()
641	return s.embeds
644// Commands returns what the agent said it can be asked to do, in its order.
646// It is what the picker lists when "/" is typed at the start of the box, and
647// it changes whenever the agent sends another available_commands_update — an
648// agent may add or take away commands as the conversation goes.
649func (s *Session) Commands() []Command {
650	s.mu.Lock()
651	defer s.mu.Unlock()
653	out := make([]Command, len(s.commands))
654	copy(out, s.commands)
655	return out
658// Unknown returns how many updates arrived that this client does not
659// understand. It is in the status dialog so that "the protocol moved on" is
660// visible rather than silent.
661func (s *Session) Unknown() int {
662	s.mu.Lock()
663	defer s.mu.Unlock()
664	return s.unknown
667// Unreadable returns the last update this client could not decode, as "kind:
668// error", or "" when every update so far was read. It is one line of the
669// status dialog, and the reason TraceEnv exists.
670func (s *Session) Unreadable() string {
671	s.mu.Lock()
672	defer s.mu.Unlock()
673	return s.unreadable
676// AgentInfo returns what the agent called itself in the handshake: the name,
677// title and version of the *program*. It is what the status dialog shows, so
678// that "which build of the agent is this?" has an answer.
679func (s *Session) AgentInfo() Implementation {
680	s.mu.Lock()
681	defer s.mu.Unlock()
682	return s.agentInfo
685// Log returns what the agent has written to its standard error, and nothing
686// for a session that has no process of its own.
687func (s *Session) Log() []string {
688	if s.process == nil {
689		return nil
691	return s.process.Log()