527 lines
13 KiB
Go
527 lines
13 KiB
Go
package fs
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"syscall"
|
|
|
|
"ollie/cmd/olliesrv/internal/agent"
|
|
"ollie/paths"
|
|
"ollie/cmd/olliesrv/internal/session"
|
|
)
|
|
|
|
func requestAgentNew(s *SessionNode, data []byte) ([]byte, error) {
|
|
ag, err := session.CreateAgent(s.Name(), strings.Fields(string(data)))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create agent: %w", err)
|
|
}
|
|
// Create AgentLog for this agent and wire up the event handler
|
|
al := NewAgentLog("")
|
|
s.SetAgentLog(ag.ID(), al)
|
|
wireAgentEvents(s.ID(), ag, al)
|
|
return []byte(ag.ID() + "\n"), nil
|
|
}
|
|
|
|
func writeAgentPrompt(s *SessionNode, ag *agent.Agent, al *AgentLog, data []byte) error {
|
|
input := strings.TrimSpace(string(data))
|
|
if input == "" {
|
|
return nil
|
|
}
|
|
if input == "/invalidate" {
|
|
s.InvalidateModelsCache()
|
|
return nil
|
|
}
|
|
go func() {
|
|
ag.Submit(s.Ctx(), input)
|
|
al.EnsureTrailingNewline()
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
func writeAgentFifoIn(ag *agent.Agent, data []byte) error {
|
|
input := strings.TrimSpace(string(data))
|
|
if input == "" {
|
|
return nil
|
|
}
|
|
ag.Queue(input)
|
|
return nil
|
|
}
|
|
|
|
func readAgentFifoOut(ag *agent.Agent) ([]byte, error) {
|
|
item, ok := ag.PopQueue()
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
return []byte(item), nil
|
|
}
|
|
|
|
func statAgentChat() os.FileInfo {
|
|
return &SyntheticFileInfo{Name_: "chat", Mode_: 0444, Size_: 64 * 1024}
|
|
}
|
|
|
|
func readAgentChat(al *AgentLog) ([]byte, error) {
|
|
al.mu.RLock()
|
|
defer al.mu.RUnlock()
|
|
data := make([]byte, len(al.log))
|
|
copy(data, al.log)
|
|
return data, nil
|
|
}
|
|
|
|
func readAgentChatText(al *AgentLog) ([]byte, error) {
|
|
al.mu.RLock()
|
|
defer al.mu.RUnlock()
|
|
return stripMarkers(al.log), nil
|
|
}
|
|
|
|
func streamAgentChat(s *SessionNode, al *AgentLog, cctx context.Context, base string) ([]byte, string, error) {
|
|
merged, cancel := mergeCtx(s.Ctx(), cctx)
|
|
defer cancel()
|
|
return streamChat(al, merged, base)
|
|
}
|
|
|
|
func streamAgentChatText(s *SessionNode, al *AgentLog, cctx context.Context, base string) ([]byte, string, error) {
|
|
merged, cancel := mergeCtx(s.Ctx(), cctx)
|
|
defer cancel()
|
|
for {
|
|
data, nextBase, err := streamChat(al, merged, base)
|
|
if err != nil || len(data) == 0 {
|
|
return data, nextBase, err
|
|
}
|
|
stripped := stripMarkers(data)
|
|
if len(stripped) > 0 {
|
|
return stripped, nextBase, nil
|
|
}
|
|
// All data was markers/fences — advance base and wait for more.
|
|
base = nextBase
|
|
}
|
|
}
|
|
|
|
// stripMarkers removes [[[...]]] marker lines and source fence delimiter lines
|
|
// from raw chat data. Content inside fences is preserved — only the ``` lines
|
|
// themselves are stripped. Partial lines (no trailing \n) are always passed
|
|
// through — they are streaming content, never markers.
|
|
func stripMarkers(data []byte) []byte {
|
|
var out []byte
|
|
for len(data) > 0 {
|
|
// Find next newline
|
|
nl := bytes.IndexByte(data, '\n')
|
|
if nl < 0 {
|
|
// Partial line at end — always content, emit it
|
|
out = append(out, data...)
|
|
break
|
|
}
|
|
line := data[:nl+1] // includes \n
|
|
data = data[nl+1:]
|
|
|
|
// Drop [[[...]]] marker lines
|
|
trimmed := bytes.TrimRight(line, "\n")
|
|
if len(trimmed) >= 7 && bytes.HasPrefix(trimmed, []byte("[[[")) && bytes.HasSuffix(trimmed, []byte("]]]")) {
|
|
continue
|
|
}
|
|
|
|
// Drop bare fence delimiter lines (``` or ```lang).
|
|
// These are the source fences wrapping tool output.
|
|
// Keep fences with 4+ backticks (````), those are user/assistant markdown.
|
|
stripped := bytes.TrimSpace(trimmed)
|
|
if bytes.HasPrefix(stripped, []byte("```")) && !bytes.HasPrefix(stripped, []byte("````")) {
|
|
continue
|
|
}
|
|
|
|
out = append(out, line...)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func blockAgentStateWait(s *SessionNode, ag *agent.Agent, cctx context.Context, base string) ([]byte, string, error) {
|
|
merged, cancel := mergeCtx(s.Ctx(), cctx)
|
|
defer cancel()
|
|
if base == "" {
|
|
base = ag.State()
|
|
}
|
|
v, ok := ag.WaitChange(merged, agent.WatchState, base)
|
|
if !ok {
|
|
st := ag.State()
|
|
return []byte(st + "\n"), st, nil
|
|
}
|
|
return []byte(v + "\n"), v, nil
|
|
}
|
|
|
|
func readAgentLog(al *AgentLog) ([]byte, error) {
|
|
const maxWindow = 64 * 1024
|
|
al.mu.RLock()
|
|
log := al.log
|
|
start := 0
|
|
if len(log) > maxWindow {
|
|
start = len(log) - maxWindow
|
|
}
|
|
data := make([]byte, len(log)-start)
|
|
copy(data, log[start:])
|
|
al.mu.RUnlock()
|
|
return data, nil
|
|
}
|
|
|
|
func readAgentPlan(al *AgentLog) ([]byte, error) {
|
|
al.mu.RLock()
|
|
data := make([]byte, len(al.plan))
|
|
copy(data, al.plan)
|
|
al.mu.RUnlock()
|
|
return data, nil
|
|
}
|
|
|
|
func writeAgentPlan(al *AgentLog, data []byte) error {
|
|
al.mu.Lock()
|
|
al.plan = make([]byte, len(data))
|
|
copy(al.plan, data)
|
|
al.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
func readAgentCfg(s *SessionNode, ag *agent.Agent, al *AgentLog) ([]byte, error) {
|
|
al.mu.RLock()
|
|
defer al.mu.RUnlock()
|
|
p := ag.GenParams()
|
|
var sb strings.Builder
|
|
fmt.Fprintf(&sb, "name=%s\n", s.ID())
|
|
fmt.Fprintf(&sb, "backend=%s\n", ag.BackendName())
|
|
fmt.Fprintf(&sb, "model=%s\n", ag.ModelName())
|
|
fmt.Fprintf(&sb, "id=%s\n", ag.ID())
|
|
fmt.Fprintf(&sb, "profile=%s\n", ag.Profile())
|
|
fmt.Fprintf(&sb, "displayName=%s\n", ag.Name())
|
|
fmt.Fprintf(&sb, "cwd=%s\n", ag.Cwd())
|
|
fmt.Fprintf(&sb, "remote=%s\n", al.remote)
|
|
fmt.Fprintf(&sb, "maxTokens=%d\n", p.MaxTokens)
|
|
if p.Temperature != nil {
|
|
fmt.Fprintf(&sb, "temperature=%g\n", *p.Temperature)
|
|
}
|
|
if p.TopP != nil {
|
|
fmt.Fprintf(&sb, "topP=%g\n", *p.TopP)
|
|
}
|
|
if p.TopK != nil {
|
|
fmt.Fprintf(&sb, "topK=%d\n", *p.TopK)
|
|
}
|
|
if len(p.Stop) > 0 {
|
|
fmt.Fprintf(&sb, "stop=%s\n", strings.Join(p.Stop, ","))
|
|
}
|
|
return []byte(sb.String()), nil
|
|
}
|
|
|
|
func writeAgentCfg(ag *agent.Agent, data []byte) error {
|
|
input := strings.TrimSpace(string(data))
|
|
if input == "" {
|
|
return nil
|
|
}
|
|
parts := strings.SplitN(input, "=", 2)
|
|
if len(parts) != 2 {
|
|
return fmt.Errorf("invalid cfg format (expected key=value)")
|
|
}
|
|
switch parts[0] {
|
|
case "name":
|
|
if strings.TrimSpace(parts[1]) != "" {
|
|
ag.SetName(strings.TrimSpace(parts[1]))
|
|
}
|
|
default:
|
|
return fmt.Errorf("unknown cfg key: %s", parts[0])
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func agentCtlHandler(s *SessionNode, ag *agent.Agent) func([]byte) ([]byte, error) {
|
|
handlers := map[string]func([]string) ([]byte, error){
|
|
"kill": func(_ []string) ([]byte, error) {
|
|
if s.Session != nil {
|
|
s.Session.RemoveAgent(ag.ID())
|
|
session.PublishEvent("session."+s.ID()+".agent."+ag.ID()+".kill", "")
|
|
}
|
|
return []byte("ok\n"), nil
|
|
},
|
|
"stop": func(_ []string) ([]byte, error) {
|
|
ag.Interrupt(agent.ErrInterrupted)
|
|
return []byte("ok\n"), nil
|
|
},
|
|
"compact": func(_ []string) ([]byte, error) {
|
|
if err := ag.Compact(s.Ctx()); err != nil {
|
|
return nil, err
|
|
}
|
|
return []byte("ok\n"), nil
|
|
},
|
|
"clear": func(_ []string) ([]byte, error) {
|
|
if err := ag.Clear(); err != nil {
|
|
return nil, err
|
|
}
|
|
return []byte("ok\n"), nil
|
|
},
|
|
"inject": func(args []string) ([]byte, error) {
|
|
text := strings.Join(args, " ")
|
|
if text == "" {
|
|
return nil, fmt.Errorf("inject requires text")
|
|
}
|
|
ag.InjectRewrite(text)
|
|
return []byte("ok\n"), nil
|
|
},
|
|
"i": func(args []string) ([]byte, error) {
|
|
text := strings.Join(args, " ")
|
|
if text == "" {
|
|
return nil, fmt.Errorf("inject requires text")
|
|
}
|
|
ag.InjectRewrite(text)
|
|
return []byte("ok\n"), nil
|
|
},
|
|
"agent": func(args []string) ([]byte, error) {
|
|
if len(args) == 0 {
|
|
return []byte(ag.Profile() + "\n"), nil
|
|
}
|
|
if err := ag.SwitchProfile(args[0]); err != nil {
|
|
return nil, err
|
|
}
|
|
return []byte(args[0] + "\n"), nil
|
|
},
|
|
"model": func(args []string) ([]byte, error) {
|
|
if len(args) == 0 {
|
|
if be := ag.Backend(); be != nil {
|
|
return []byte(be.Model() + "\n"), nil
|
|
}
|
|
return nil, nil
|
|
}
|
|
if be := ag.Backend(); be != nil {
|
|
be.SetModel(strings.Join(args, " "))
|
|
}
|
|
return []byte(strings.Join(args, " ") + "\n"), nil
|
|
},
|
|
"models": func(_ []string) ([]byte, error) {
|
|
return []byte(s.CachedListModels() + "\n"), nil
|
|
},
|
|
"tools": func(_ []string) ([]byte, error) {
|
|
return readAgentTools(ag)
|
|
},
|
|
"tool_load": func(args []string) ([]byte, error) {
|
|
if len(args) == 0 {
|
|
return nil, fmt.Errorf("tool_load requires a tool name")
|
|
}
|
|
ts := ag.ToolServer()
|
|
if ts == nil {
|
|
return nil, fmt.Errorf("no tool server")
|
|
}
|
|
if err := ts.LoadTool(args[0]); err != nil {
|
|
return nil, err
|
|
}
|
|
ag.RefreshTools()
|
|
return []byte(args[0] + "\n"), nil
|
|
},
|
|
"backend": func(args []string) ([]byte, error) {
|
|
if len(args) == 0 {
|
|
if be := ag.Backend(); be != nil {
|
|
return []byte(be.Name() + "\n"), nil
|
|
}
|
|
return nil, nil
|
|
}
|
|
return nil, fmt.Errorf("backend switching not supported via ctl; use cfg")
|
|
},
|
|
"name": func(args []string) ([]byte, error) {
|
|
if len(args) == 0 {
|
|
return []byte(ag.Name() + "\n"), nil
|
|
}
|
|
ag.SetName(strings.Join(args, " "))
|
|
return []byte(strings.Join(args, " ") + "\n"), nil
|
|
},
|
|
"cwd": func(args []string) ([]byte, error) {
|
|
if len(args) == 0 {
|
|
return []byte(ag.Cwd() + "\n"), nil
|
|
}
|
|
dir := paths.ExpandHome(strings.Join(args, " "))
|
|
if _, err := os.Stat(dir); err != nil {
|
|
return nil, fmt.Errorf("cwd: %w", err)
|
|
}
|
|
ag.SetCWD(dir)
|
|
return []byte(dir + "\n"), nil
|
|
},
|
|
"systemprompt": func(_ []string) ([]byte, error) {
|
|
return []byte(ag.SystemPrompt() + "\n"), nil
|
|
},
|
|
"help": func(_ []string) ([]byte, error) {
|
|
return []byte(`stop — interrupt running agent
|
|
compact — compact context
|
|
clear — clear history
|
|
inject <text> — inject prompt mid-turn (overwrites pending)
|
|
agent [name] — show/switch agent profile
|
|
model [name] — show/switch model
|
|
models — list available models
|
|
tools — list loaded tools
|
|
tool_load <n> — load a tool
|
|
cwd [path] — show/change working directory
|
|
name [name] — show/rename agent
|
|
backend — show backend
|
|
systemprompt — show rendered system prompt
|
|
help — this message
|
|
`), nil
|
|
},
|
|
}
|
|
return func(data []byte) ([]byte, error) {
|
|
return dispatch(handlers, data)
|
|
}
|
|
}
|
|
|
|
func readAgentStats(ag *agent.Agent) ([]byte, error) {
|
|
s := "usage=" + ag.UsageStr() + "\n" +
|
|
"cost=" + ag.CostStr() + "\n" +
|
|
"ctxsz=" + ag.CtxSz() + "\n"
|
|
return []byte(s), nil
|
|
}
|
|
|
|
func readAgentTools(ag *agent.Agent) ([]byte, error) {
|
|
ts := ag.ToolServer()
|
|
if ts == nil {
|
|
return []byte("(no tool server)\n"), nil
|
|
}
|
|
loaded, err := ts.ListTools()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var sb strings.Builder
|
|
for _, ti := range loaded {
|
|
if ti.Description != "" {
|
|
fmt.Fprintf(&sb, "%-20s %s\n", ti.Name, ti.Description)
|
|
} else {
|
|
sb.WriteString(ti.Name + "\n")
|
|
}
|
|
}
|
|
return []byte(sb.String()), nil
|
|
}
|
|
|
|
func writeAgentTools(ag *agent.Agent, data []byte) error {
|
|
name := strings.TrimSpace(string(data))
|
|
if name == "" {
|
|
return nil
|
|
}
|
|
ts := ag.ToolServer()
|
|
if ts == nil {
|
|
return fmt.Errorf("no tool server")
|
|
}
|
|
if err := ts.LoadTool(name); err != nil {
|
|
return err
|
|
}
|
|
ag.RefreshTools()
|
|
return nil
|
|
}
|
|
|
|
func readAgentName(ag *agent.Agent) ([]byte, error) {
|
|
return []byte(ag.Name() + "\n"), nil
|
|
}
|
|
|
|
func writeAgentName(s *SessionNode, ag *agent.Agent, data []byte) error {
|
|
newName := strings.TrimSpace(string(data))
|
|
if newName == "" {
|
|
return nil
|
|
}
|
|
ag.SetName(newName)
|
|
session.PersistSession(s.Name())
|
|
return nil
|
|
}
|
|
|
|
func readAgentID(ag *agent.Agent) ([]byte, error) {
|
|
return []byte(ag.ID() + "\n"), nil
|
|
}
|
|
|
|
func streamChat(al *AgentLog, cctx context.Context, base string) ([]byte, string, error) {
|
|
ctx, cancel := context.WithCancel(cctx)
|
|
defer cancel()
|
|
context.AfterFunc(ctx, func() { al.chatCond.Broadcast() })
|
|
|
|
var offset int
|
|
if base != "" {
|
|
fmt.Sscanf(base, "%d", &offset)
|
|
} else {
|
|
al.mu.RLock()
|
|
offset = len(al.log)
|
|
al.mu.RUnlock()
|
|
}
|
|
|
|
al.mu.RLock()
|
|
for len(al.log) <= offset {
|
|
if ctx.Err() != nil {
|
|
al.mu.RUnlock()
|
|
return nil, "", nil
|
|
}
|
|
al.chatCond.Wait()
|
|
}
|
|
|
|
data := make([]byte, len(al.log)-offset)
|
|
copy(data, al.log[offset:])
|
|
newOffset := len(al.log)
|
|
al.mu.RUnlock()
|
|
|
|
return data, fmt.Sprintf("%d", newOffset), nil
|
|
}
|
|
|
|
func readProcess(s *SessionNode, pid int) ([]byte, error) {
|
|
conn := s.DialToolServer()
|
|
if conn == nil {
|
|
return nil, fmt.Errorf("no tool server")
|
|
}
|
|
defer conn.Close()
|
|
output, err := conn.GetDetachedOutput(pid)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return []byte(output), nil
|
|
}
|
|
|
|
func readProcessStatus(s *SessionNode, pid int) ([]byte, error) {
|
|
conn := s.DialToolServer()
|
|
if conn == nil {
|
|
return nil, fmt.Errorf("no tool server")
|
|
}
|
|
defer conn.Close()
|
|
raw := conn.ListDetachedRaw()
|
|
for _, r := range raw {
|
|
if m, ok := r.(map[string]any); ok {
|
|
if id, ok := m["pid"].(int); ok && id == pid {
|
|
if exited, ok := m["exited"].(bool); ok && exited {
|
|
exitCode := 0
|
|
if code, ok := m["exit_code"].(int); ok {
|
|
exitCode = code
|
|
}
|
|
return []byte(fmt.Sprintf("exited %d\n", exitCode)), nil
|
|
}
|
|
return []byte("running\n"), nil
|
|
}
|
|
}
|
|
}
|
|
return nil, fmt.Errorf("process %d not found", pid)
|
|
}
|
|
|
|
func writeProcessCtl(s *SessionNode, pid int, data []byte) error {
|
|
conn := s.DialToolServer()
|
|
if conn == nil {
|
|
return fmt.Errorf("no tool server")
|
|
}
|
|
defer conn.Close()
|
|
cmd := strings.TrimSpace(string(data))
|
|
switch {
|
|
case cmd == "term":
|
|
return conn.SignalDetached(pid, syscall.SIGTERM)
|
|
case cmd == "kill":
|
|
return conn.SignalDetached(pid, syscall.SIGKILL)
|
|
case cmd == "dismiss":
|
|
if !conn.DismissDetached(pid) {
|
|
return fmt.Errorf("process %d not found", pid)
|
|
}
|
|
return nil
|
|
case strings.HasPrefix(cmd, "signal "):
|
|
parts := strings.Fields(cmd)
|
|
if len(parts) < 2 {
|
|
return fmt.Errorf("signal requires signal number")
|
|
}
|
|
sig, err := strconv.Atoi(parts[1])
|
|
if err != nil {
|
|
return fmt.Errorf("invalid signal: %s", parts[1])
|
|
}
|
|
return conn.SignalDetached(pid, syscall.Signal(sig))
|
|
default:
|
|
return fmt.Errorf("unknown command: %s (valid: term, kill, signal <n>, dismiss)", cmd)
|
|
}
|
|
}
|