Add multiple terminals per task and saved commands that start one

internal/terminal/terminal.go

@@ -2,12 +2,14 @@ package terminal
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"log/slog"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"slices"
 	"strings"
 	"sync"
 	"syscall"
@@ -16,6 +18,7 @@ import (
 	"github.com/creack/pty"
 
 	"g.fc.run/ai/byre/internal/api"
+	"g.fc.run/ai/byre/internal/ids"
 )
 
 const (
@@ -25,6 +28,9 @@ const (
 	promptMark  = "\x1b]133;D;"
 	lineMark    = "\x1b]7701;"
 	carryMax    = 64 << 10
+	keepSize    = 256 << 10
+	Quiet       = 500 * time.Millisecond
+	MaxWait     = 10 * time.Second
 )
 
 var lineMarks = [][]byte{[]byte(promptMark), []byte(lineMark)}
@@ -52,12 +58,15 @@ fi
 
 type Terminals struct {
 	zdotdir string
+	changed func(taskID string)
 
 	mu     sync.Mutex
-	byTask map[string]*Terminal
+	byTask map[string][]*Terminal
 }
 
 type Terminal struct {
+	ID      string
+	Name    string
 	ptmx    *os.File
 	cmd     *exec.Cmd
 	Prompts bool
@@ -80,7 +89,7 @@ type wake struct {
 	out []byte
 }
 
-func New(zdotdir string) (*Terminals, error) {
+func New(zdotdir string, changed func(taskID string)) (*Terminals, error) {
 	err := os.MkdirAll(zdotdir, 0o755)
 	if err != nil {
 		return nil, err
@@ -89,23 +98,50 @@ func New(zdotdir string) (*Terminals, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Terminals{zdotdir: zdotdir, byTask: map[string]*Terminal{}}, nil
+	return &Terminals{zdotdir: zdotdir, changed: changed, byTask: map[string][]*Terminal{}}, nil
 }
 
-func (ts *Terminals) Get(taskID string) *Terminal {
+func (ts *Terminals) List(taskID string) []*Terminal {
 	ts.mu.Lock()
 	defer ts.mu.Unlock()
-	return ts.byTask[taskID]
+	return slices.Clone(ts.byTask[taskID])
 }
 
-func (ts *Terminals) Attach(t api.Task, size api.TerminalSize) (*Terminal, error) {
+func (ts *Terminals) Get(taskID, id string) *Terminal {
 	ts.mu.Lock()
 	defer ts.mu.Unlock()
-
-	term := ts.byTask[t.Id]
-	if term != nil {
-		return term, nil
+	for _, term := range ts.byTask[taskID] {
+		if term.ID == id {
+			return term
+		}
 	}
+	return nil
+}
+
+func (ts *Terminals) First(taskID string) *Terminal {
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
+	if len(ts.byTask[taskID]) == 0 {
+		return nil
+	}
+	return ts.byTask[taskID][0]
+}
+
+func (ts *Terminals) Start(t api.Task, name, command string, size api.TerminalSize) (*Terminal, error) {
+	term, err := ts.start(t, name, size)
+	if err != nil {
+		return nil, err
+	}
+	ts.changed(t.Id)
+	if command != "" {
+		go term.typeWhenReady(command + "\r")
+	}
+	return term, nil
+}
+
+func (ts *Terminals) start(t api.Task, name string, size api.TerminalSize) (*Terminal, error) {
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
 
 	shell := os.Getenv("SHELL")
 	if shell == "" {
@@ -123,27 +159,49 @@ func (ts *Terminals) Attach(t api.Task, size api.TerminalSize) (*Terminal, error
 		return nil, fmt.Errorf("start %s: %w", shell, err)
 	}
 
-	term = &Terminal{ptmx: ptmx, cmd: cmd, Prompts: prompts, clients: map[chan []byte]bool{}, started: time.Now()}
-	ts.byTask[t.Id] = term
-	slog.Info("terminal started", "task", t.Id, "shell", shell, "pid", cmd.Process.Pid)
+	term := &Terminal{ID: ids.New(ids.Terminal), Name: name, ptmx: ptmx, cmd: cmd, Prompts: prompts, clients: map[chan []byte]bool{}, started: time.Now()}
+	ts.byTask[t.Id] = append(ts.byTask[t.Id], term)
+	slog.Info("terminal started", "task", t.Id, "terminal", term.ID, "name", name, "shell", shell, "pid", cmd.Process.Pid)
 	go func() {
 		term.pump()
-		ts.mu.Lock()
-		if ts.byTask[t.Id] == term {
-			delete(ts.byTask, t.Id)
+		if ts.remove(t.Id, term) {
+			ts.changed(t.Id)
 		}
-		ts.mu.Unlock()
-		slog.Info("terminal ended", "task", t.Id, "exit", term.exit)
+		slog.Info("terminal ended", "task", t.Id, "terminal", term.ID, "exit", term.exit)
 	}()
 	return term, nil
 }
 
+func (ts *Terminals) remove(taskID string, term *Terminal) bool {
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
+	i := slices.Index(ts.byTask[taskID], term)
+	if i < 0 {
+		return false
+	}
+	ts.byTask[taskID] = slices.Delete(ts.byTask[taskID], i, i+1)
+	if len(ts.byTask[taskID]) == 0 {
+		delete(ts.byTask, taskID)
+	}
+	return true
+}
+
+func (ts *Terminals) Kill(taskID, id string) bool {
+	term := ts.Get(taskID, id)
+	if term == nil || !ts.remove(taskID, term) {
+		return false
+	}
+	term.kill()
+	ts.changed(taskID)
+	return true
+}
+
 func (ts *Terminals) Close(taskID string) {
 	ts.mu.Lock()
-	term := ts.byTask[taskID]
+	terms := ts.byTask[taskID]
 	delete(ts.byTask, taskID)
 	ts.mu.Unlock()
-	if term != nil {
+	for _, term := range terms {
 		term.kill()
 	}
 }
@@ -151,10 +209,12 @@ func (ts *Terminals) Close(taskID string) {
 func (ts *Terminals) CloseAll() {
 	ts.mu.Lock()
 	all := ts.byTask
-	ts.byTask = map[string]*Terminal{}
+	ts.byTask = map[string][]*Terminal{}
 	ts.mu.Unlock()
-	for _, term := range all {
-		term.kill()
+	for _, terms := range all {
+		for _, term := range terms {
+			term.kill()
+		}
 	}
 }
 
@@ -349,6 +409,70 @@ func (term *Terminal) Activity() (started, lastOutput time.Time) {
 	return term.started, term.lastOutput
 }
 
+func (term *Terminal) Started() time.Time {
+	term.mu.Lock()
+	defer term.mu.Unlock()
+	return term.started
+}
+
+func (term *Terminal) typeWhenReady(text string) {
+	backlog, out := term.Subscribe()
+	if out == nil {
+		return
+	}
+	_, err := term.AwaitPrompt(context.Background(), out, backlog)
+	term.Unsubscribe(out)
+	if err != nil {
+		return
+	}
+	err = term.Write([]byte(text))
+	if err != nil {
+		slog.Warn("type saved command", "terminal", term.ID, "err", err)
+	}
+}
+
+func (term *Terminal) Settling() bool {
+	return term.promptWait() > 0
+}
+
+func (term *Terminal) promptWait() time.Duration {
+	started, last := term.Activity()
+	wait := time.Until(started.Add(MaxWait))
+	if !last.IsZero() {
+		wait = min(wait, time.Until(last.Add(Quiet)))
+	}
+	return wait
+}
+
+func (term *Terminal) AwaitPrompt(ctx context.Context, out chan []byte, got []byte) ([]byte, error) {
+	for {
+		wait := term.promptWait()
+		if wait <= 0 {
+			return got, nil
+		}
+		timer := time.NewTimer(wait)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return nil, ctx.Err()
+		case <-timer.C:
+		case b, open := <-out:
+			timer.Stop()
+			if !open {
+				return got, nil
+			}
+			got = KeepTail(append(got, b...))
+		}
+	}
+}
+
+func KeepTail(b []byte) []byte {
+	if len(b) > 2*keepSize {
+		return bytes.Clone(b[len(b)-keepSize:])
+	}
+	return b
+}
+
 func (term *Terminal) Write(b []byte) error {
 	_, err := term.ptmx.Write(b)
 	return err