olliesrv/fs: inline all handlers into spec.go
The namespace spec is now fully self-contained. Every handler is a closure right where it's declared — no jumping between files. Deleted: handlers_root.go, handlers_session.go, handlers_agent.go, ctl.go Added: support.go (shared utilities: mergeCtx, streamChat, stripMarkers, dispatch, wireAgentEvents, help memoization) spec.go grew to ~920 lines but is now the single source of truth for the entire 9P namespace. Know where to look.
This commit is contained in:
Ollie Agent
parent
a40e1dbed0
commit
066659b8b6
|
|
@ -1,21 +0,0 @@
|
||||||
package fs
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
"strings"
|
|
||||||
)
|
|
||||||
|
|
||||||
// dispatch routes a command to the matching handler based on the first word.
|
|
||||||
func dispatch(handlers map[string]func([]string) ([]byte, error), data []byte) ([]byte, error) {
|
|
||||||
input := strings.TrimSpace(string(data))
|
|
||||||
if input == "" {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
parts := strings.Fields(input)
|
|
||||||
cmd := parts[0]
|
|
||||||
h, ok := handlers[cmd]
|
|
||||||
if !ok {
|
|
||||||
return nil, fmt.Errorf("unknown ctl command: %s", cmd)
|
|
||||||
}
|
|
||||||
return h(parts[1:])
|
|
||||||
}
|
|
||||||
|
|
@ -1,524 +0,0 @@
|
||||||
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) {
|
|
||||||
s.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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,205 +0,0 @@
|
||||||
package fs
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
"strings"
|
|
||||||
"sync"
|
|
||||||
|
|
||||||
"ollie/cmd/olliesrv/internal/agent"
|
|
||||||
"ollie/cmd/olliesrv/internal/backend"
|
|
||||||
"ollie/cmd/olliesrv/internal/bypass"
|
|
||||||
"ollie/virtfs"
|
|
||||||
"ollie/cmd/olliesrv/internal/session"
|
|
||||||
"ollie/toolsrv"
|
|
||||||
)
|
|
||||||
|
|
||||||
// mergeCtx returns a context that cancels when either parent or child cancels.
|
|
||||||
// parent is typically the session context; child is the per-read timeout from the server.
|
|
||||||
func mergeCtx(parent, child context.Context) (context.Context, context.CancelFunc) {
|
|
||||||
merged, cancel := context.WithCancel(parent)
|
|
||||||
stop := context.AfterFunc(child, func() { cancel() })
|
|
||||||
return merged, func() { stop(); cancel() }
|
|
||||||
}
|
|
||||||
|
|
||||||
// wireAgentEvents connects an agent's output and state change events.
|
|
||||||
// Called once per agent: at creation (handlers_agent.go) and on restore (newroot.go).
|
|
||||||
//
|
|
||||||
// Contract:
|
|
||||||
// - SetOutput → events flow to the AgentLog (chat stream, tool blocks, etc.)
|
|
||||||
// - SetOnStateChange → state transitions publish to the event bus
|
|
||||||
func wireAgentEvents(sessID string, ag *agent.Agent, al *AgentLog) {
|
|
||||||
ag.SetOutput(NewEventHandler(al))
|
|
||||||
ag.SetOnStateChange(func(agentID, state string) {
|
|
||||||
session.PublishEvent("session."+sessID+".agent."+agentID+".state", state)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
func readBackends() ([]byte, error) {
|
|
||||||
return []byte(strings.Join(backend.Backends(), "\n") + "\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
var helpFn func() string
|
|
||||||
|
|
||||||
// helpSpec stores the spec for help generation. Set during tree build.
|
|
||||||
var helpSpec *virtfs.FsNodeDecl
|
|
||||||
|
|
||||||
func readHelp() ([]byte, error) {
|
|
||||||
if helpFn == nil {
|
|
||||||
return []byte("help not initialized\n"), nil
|
|
||||||
}
|
|
||||||
return []byte(helpFn()), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func init() {
|
|
||||||
helpFn = func() string {
|
|
||||||
helpOnce.Do(func() {
|
|
||||||
if helpSpec != nil {
|
|
||||||
helpText = virtfs.GenerateHelp(*helpSpec)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
return helpText
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
var (
|
|
||||||
helpOnce sync.Once
|
|
||||||
helpText string
|
|
||||||
)
|
|
||||||
|
|
||||||
func readModels(mc *ModelCache) ([]byte, error) {
|
|
||||||
if mc != nil {
|
|
||||||
return mc.Get(), nil
|
|
||||||
}
|
|
||||||
return []byte("(no model cache)\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readAgents() ([]byte, error) {
|
|
||||||
var sb strings.Builder
|
|
||||||
for _, dir := range agent.AgentsDirs() {
|
|
||||||
entries, err := os.ReadDir(dir)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
for _, e := range entries {
|
|
||||||
if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") {
|
|
||||||
sb.WriteString(strings.TrimSuffix(e.Name(), ".json"))
|
|
||||||
sb.WriteByte('\n')
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return []byte(sb.String()), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readTools() ([]byte, error) {
|
|
||||||
tools := toolsrv.DiscoverTools()
|
|
||||||
if len(tools) == 0 {
|
|
||||||
return []byte("(no tools found)\n"), nil
|
|
||||||
}
|
|
||||||
var sb strings.Builder
|
|
||||||
for _, ti := range tools {
|
|
||||||
fmt.Fprintf(&sb, "%-24s %s\n", ti.Name, ti.Description)
|
|
||||||
}
|
|
||||||
return []byte(sb.String()), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func rootCtlHandler(mc *ModelCache, invalidate func(), shutdown func()) func([]byte) ([]byte, error) {
|
|
||||||
handlers := map[string]func([]string) ([]byte, error){
|
|
||||||
"invalidate": func(_ []string) ([]byte, error) {
|
|
||||||
if mc != nil {
|
|
||||||
mc.Invalidate()
|
|
||||||
}
|
|
||||||
if invalidate != nil {
|
|
||||||
invalidate()
|
|
||||||
}
|
|
||||||
return []byte("ok\n"), nil
|
|
||||||
},
|
|
||||||
"kill": func(_ []string) ([]byte, error) {
|
|
||||||
if shutdown != nil {
|
|
||||||
shutdown()
|
|
||||||
}
|
|
||||||
return []byte("ok\n"), nil
|
|
||||||
},
|
|
||||||
}
|
|
||||||
return func(data []byte) ([]byte, error) {
|
|
||||||
return dispatch(handlers, data)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func readEventwait() ([]byte, error) {
|
|
||||||
return nil, fmt.Errorf("eventwait: use blocking read")
|
|
||||||
}
|
|
||||||
|
|
||||||
func blockEventwait(cctx context.Context, _ string) ([]byte, string, error) {
|
|
||||||
return session.WaitEvent(cctx)
|
|
||||||
}
|
|
||||||
|
|
||||||
func requestGenerate(ctx context.Context, data []byte) ([]byte, error) {
|
|
||||||
var req backend.GenerateRequest
|
|
||||||
if err := json.Unmarshal(data, &req); err != nil {
|
|
||||||
req.Prompt = strings.TrimSpace(string(data))
|
|
||||||
}
|
|
||||||
result, err := backend.Generate(ctx, req)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return []byte(result + "\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readAliases() ([]byte, error) {
|
|
||||||
var sb strings.Builder
|
|
||||||
for name, sess := range session.Sessions() {
|
|
||||||
sid := sess.ID
|
|
||||||
fmt.Fprintf(&sb, "%s\tsession/%s\n", sid, name)
|
|
||||||
for _, ag := range sess.Agents() {
|
|
||||||
fmt.Fprintf(&sb, "%s\tsession/%s/agent/%s\n", ag.ID(), name, ag.Name())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return []byte(sb.String()), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
var errNoBypass = fmt.Errorf("bypass broker not available")
|
|
||||||
|
|
||||||
func readBypassPolicy(b *bypass.Broker) ([]byte, error) {
|
|
||||||
if b == nil {
|
|
||||||
return nil, errNoBypass
|
|
||||||
}
|
|
||||||
p := b.GlobalPolicy().Global()
|
|
||||||
return p.Marshal()
|
|
||||||
}
|
|
||||||
|
|
||||||
func writeBypassPolicy(b *bypass.Broker, data []byte) error {
|
|
||||||
if b == nil {
|
|
||||||
return errNoBypass
|
|
||||||
}
|
|
||||||
var p bypass.Policy
|
|
||||||
if err := bypass.ParsePolicy(data, &p); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return b.GlobalPolicy().SetGlobal(p)
|
|
||||||
}
|
|
||||||
|
|
||||||
func readBypassRequest(req *bypass.Request) ([]byte, error) {
|
|
||||||
return []byte(req.Summary() + "\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func writeBypassRequest(b *bypass.Broker, req *bypass.Request, data []byte) error {
|
|
||||||
cmd := strings.TrimSpace(string(data))
|
|
||||||
var res bypass.Resolution
|
|
||||||
switch cmd {
|
|
||||||
case "approve":
|
|
||||||
res = bypass.ResolveApprove
|
|
||||||
case "deny":
|
|
||||||
res = bypass.ResolveDeny
|
|
||||||
case "persist":
|
|
||||||
res = bypass.ResolvePersist
|
|
||||||
default:
|
|
||||||
return fmt.Errorf("unknown resolution: %s (use approve/deny/persist)", cmd)
|
|
||||||
}
|
|
||||||
if !b.Resolve(req.ID, res) {
|
|
||||||
return fmt.Errorf("request already resolved or timed out")
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,139 +0,0 @@
|
||||||
package fs
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
"strings"
|
|
||||||
|
|
||||||
"ollie/cmd/olliesrv/internal/bypass"
|
|
||||||
"ollie/cmd/olliesrv/internal/session"
|
|
||||||
)
|
|
||||||
|
|
||||||
func readSessionIdx() ([]byte, error) {
|
|
||||||
return session.BuildIndex(), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readAgentIdx(sess *session.Session) ([]byte, error) {
|
|
||||||
return session.BuildAgentIndex(sess), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func requestSessionNew(data []byte) ([]byte, error) {
|
|
||||||
args := strings.Fields(string(data))
|
|
||||||
name := ""
|
|
||||||
remote := ""
|
|
||||||
for _, arg := range args {
|
|
||||||
if k, v, ok := strings.Cut(arg, "="); ok {
|
|
||||||
switch k {
|
|
||||||
case "name":
|
|
||||||
name = v
|
|
||||||
case "remote":
|
|
||||||
remote = v
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
sess, err := session.CreateEmpty(name, remote)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return []byte(sess.Name() + "\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readSessionEnv(s *SessionNode) ([]byte, error) {
|
|
||||||
var sb strings.Builder
|
|
||||||
fmt.Fprintf(&sb, "OLLIE_SESSION_ID=%s\n", s.ID)
|
|
||||||
for _, e := range os.Environ() {
|
|
||||||
if strings.HasPrefix(e, "OLLIE_") && !strings.HasPrefix(e, "OLLIE_SESSION_ID=") {
|
|
||||||
sb.WriteString(e)
|
|
||||||
sb.WriteByte('\n')
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return []byte(sb.String()), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readSessionPaused(s *SessionNode) ([]byte, error) {
|
|
||||||
if s.IsPaused() {
|
|
||||||
return []byte("true\n"), nil
|
|
||||||
}
|
|
||||||
return []byte("false\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readSessionConnected(s *SessionNode) ([]byte, error) {
|
|
||||||
if s.IsConnected() {
|
|
||||||
return []byte("true\n"), nil
|
|
||||||
}
|
|
||||||
return []byte("false\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func sessionCtlHandler(s *SessionNode, removeFn func() error, renameFn func(string) error) func([]byte) ([]byte, error) {
|
|
||||||
handlers := map[string]func([]string) ([]byte, error){
|
|
||||||
"kill": func(_ []string) ([]byte, error) {
|
|
||||||
return []byte("ok\n"), removeFn()
|
|
||||||
},
|
|
||||||
".": func(_ []string) ([]byte, error) {
|
|
||||||
return []byte("ok\n"), removeFn()
|
|
||||||
},
|
|
||||||
"save": func(_ []string) ([]byte, error) {
|
|
||||||
s.Save()
|
|
||||||
return []byte("ok\n"), nil
|
|
||||||
},
|
|
||||||
"invalidate": func(_ []string) ([]byte, error) {
|
|
||||||
s.InvalidateModelsCache()
|
|
||||||
return []byte("ok\n"), nil
|
|
||||||
},
|
|
||||||
"pause": func(_ []string) ([]byte, error) {
|
|
||||||
if err := s.Pause(); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return []byte("ok\n"), nil
|
|
||||||
},
|
|
||||||
"resume": func(_ []string) ([]byte, error) {
|
|
||||||
if err := s.Resume(); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return []byte("ok\n"), nil
|
|
||||||
},
|
|
||||||
}
|
|
||||||
return func(data []byte) ([]byte, error) {
|
|
||||||
return dispatch(handlers, data)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func readSessionName(s *SessionNode) ([]byte, error) {
|
|
||||||
return []byte(s.Name() + "\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func writeSessionName(s *SessionNode, renameFn func(string) error, data []byte) error {
|
|
||||||
newName := strings.TrimSpace(string(data))
|
|
||||||
if newName == "" {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
oldName := s.Name()
|
|
||||||
if newName != oldName {
|
|
||||||
return renameFn(newName)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readSessionID(s *SessionNode) ([]byte, error) {
|
|
||||||
return []byte(s.ID + "\n"), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func readSessionBypass(b *bypass.Broker, s *SessionNode) ([]byte, error) {
|
|
||||||
if b == nil {
|
|
||||||
return nil, fmt.Errorf("bypass broker not available")
|
|
||||||
}
|
|
||||||
p := b.SessionPolicy(s.ID)
|
|
||||||
return p.Marshal()
|
|
||||||
}
|
|
||||||
|
|
||||||
func writeSessionBypass(b *bypass.Broker, s *SessionNode, data []byte) error {
|
|
||||||
if b == nil {
|
|
||||||
return fmt.Errorf("bypass broker not available")
|
|
||||||
}
|
|
||||||
var p bypass.Policy
|
|
||||||
if err := bypass.ParsePolicy(data, &p); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
b.SetSessionPolicy(s.ID, p)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,16 +1,24 @@
|
||||||
package fs
|
package fs
|
||||||
|
|
||||||
// Filesystem specification — the single source of truth for the 9P namespace.
|
// spec.go — the single source of truth for the olliesrv 9P namespace.
|
||||||
// Constructed via buildTreeSpec with closure capture.
|
// Every handler is inline. No archaeological expeditions needed.
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
"os/user"
|
"os/user"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"syscall"
|
||||||
|
|
||||||
"ollie/cmd/olliesrv/internal/agent"
|
"ollie/cmd/olliesrv/internal/agent"
|
||||||
|
"ollie/cmd/olliesrv/internal/backend"
|
||||||
"ollie/cmd/olliesrv/internal/bypass"
|
"ollie/cmd/olliesrv/internal/bypass"
|
||||||
"ollie/cmd/olliesrv/internal/session"
|
"ollie/cmd/olliesrv/internal/session"
|
||||||
|
"ollie/paths"
|
||||||
|
"ollie/toolsrv"
|
||||||
"ollie/virtfs"
|
"ollie/virtfs"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -31,7 +39,7 @@ const (
|
||||||
|
|
||||||
var treeSpecUID, treeSpecGID string
|
var treeSpecUID, treeSpecGID string
|
||||||
|
|
||||||
// buildTreeSpec constructs the full 9P namespace as an virtfs.FsNodeDecl.
|
// buildTreeSpec constructs the full 9P namespace.
|
||||||
// All handlers are closures that capture their dependencies at binding time.
|
// All handlers are closures that capture their dependencies at binding time.
|
||||||
func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl {
|
func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl {
|
||||||
cfg := &rs.Cfg
|
cfg := &rs.Cfg
|
||||||
|
|
@ -51,80 +59,228 @@ func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl {
|
||||||
|
|
||||||
virtfs.FileNode("backends", 0444,
|
virtfs.FileNode("backends", 0444,
|
||||||
virtfs.Doc("Available backend names, one per line"),
|
virtfs.Doc("Available backend names, one per line"),
|
||||||
virtfs.Read(readBackends),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
return []byte(strings.Join(backend.Backends(), "\n") + "\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("help", 0444,
|
virtfs.FileNode("help", 0444,
|
||||||
virtfs.Doc("Filesystem reference"),
|
virtfs.Doc("Filesystem reference"),
|
||||||
virtfs.Read(readHelp),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
return []byte(helpFn()), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("models", 0444,
|
virtfs.FileNode("models", 0444,
|
||||||
virtfs.Doc("Available models. Format: backend<tab>model per line."),
|
virtfs.Doc("Available models. Format: backend<tab>model per line."),
|
||||||
virtfs.Read(func() ([]byte, error) { return readModels(mc) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
if mc != nil {
|
||||||
|
return mc.Get(), nil
|
||||||
|
}
|
||||||
|
return []byte("(no model cache)\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("agents", 0444,
|
virtfs.FileNode("agents", 0444,
|
||||||
virtfs.Doc("Available agent profiles, one per line"),
|
virtfs.Doc("Available agent profiles, one per line"),
|
||||||
virtfs.Read(readAgents),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
var sb strings.Builder
|
||||||
|
for _, dir := range agent.AgentsDirs() {
|
||||||
|
entries, err := os.ReadDir(dir)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for _, e := range entries {
|
||||||
|
if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") {
|
||||||
|
sb.WriteString(strings.TrimSuffix(e.Name(), ".json"))
|
||||||
|
sb.WriteByte('\n')
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return []byte(sb.String()), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("tools", 0444,
|
virtfs.FileNode("tools", 0444,
|
||||||
virtfs.Doc("Available tools. Format: name<spaces>description per line."),
|
virtfs.Doc("Available tools. Format: name<spaces>description per line."),
|
||||||
virtfs.Read(readTools),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
tools := toolsrv.DiscoverTools()
|
||||||
|
if len(tools) == 0 {
|
||||||
|
return []byte("(no tools found)\n"), nil
|
||||||
|
}
|
||||||
|
var sb strings.Builder
|
||||||
|
for _, ti := range tools {
|
||||||
|
fmt.Fprintf(&sb, "%-24s %s\n", ti.Name, ti.Description)
|
||||||
|
}
|
||||||
|
return []byte(sb.String()), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("ctl", 0666,
|
virtfs.FileNode("ctl", 0666,
|
||||||
virtfs.Doc("Server control. Write: 'invalidate', 'kill'"),
|
virtfs.Doc("Server control. Write: 'invalidate', 'kill'"),
|
||||||
virtfs.Request(rootCtlHandler(mc, cfg.Invalidate, cfg.Shutdown)),
|
virtfs.Request(func(data []byte) ([]byte, error) {
|
||||||
|
return dispatch(map[string]func([]string) ([]byte, error){
|
||||||
|
"invalidate": func(_ []string) ([]byte, error) {
|
||||||
|
if mc != nil {
|
||||||
|
mc.Invalidate()
|
||||||
|
}
|
||||||
|
if cfg.Invalidate != nil {
|
||||||
|
cfg.Invalidate()
|
||||||
|
}
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"kill": func(_ []string) ([]byte, error) {
|
||||||
|
if cfg.Shutdown != nil {
|
||||||
|
cfg.Shutdown()
|
||||||
|
}
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
}, data)
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("eventwait", 0444,
|
virtfs.FileNode("eventwait", 0444,
|
||||||
virtfs.Doc("Blocking read for server events"),
|
virtfs.Doc("Blocking read for server events"),
|
||||||
virtfs.Read(readEventwait),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
virtfs.BlockOnce(blockEventwait),
|
return nil, fmt.Errorf("eventwait: use blocking read")
|
||||||
|
}),
|
||||||
|
virtfs.BlockOnce(func(cctx context.Context, _ string) ([]byte, string, error) {
|
||||||
|
return session.WaitEvent(cctx)
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("generate", 0666,
|
virtfs.FileNode("generate", 0666,
|
||||||
virtfs.Doc("One-shot LLM generation"),
|
virtfs.Doc("One-shot LLM generation"),
|
||||||
virtfs.Request(func(data []byte) ([]byte, error) { return requestGenerate(cfg.Ctx, data) }),
|
virtfs.Request(func(data []byte) ([]byte, error) {
|
||||||
|
var req backend.GenerateRequest
|
||||||
|
if err := json.Unmarshal(data, &req); err != nil {
|
||||||
|
req.Prompt = strings.TrimSpace(string(data))
|
||||||
|
}
|
||||||
|
result, err := backend.Generate(cfg.Ctx, req)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return []byte(result + "\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("aliases", 0444,
|
virtfs.FileNode("aliases", 0444,
|
||||||
virtfs.Doc("Alias table. Format: id<tab>path per line."),
|
virtfs.Doc("Alias table. Format: id<tab>path per line."),
|
||||||
virtfs.Read(readAliases),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
var sb strings.Builder
|
||||||
|
for name, sess := range session.Sessions() {
|
||||||
|
fmt.Fprintf(&sb, "%s\tsession/%s\n", sess.ID, name)
|
||||||
|
for _, ag := range sess.Agents() {
|
||||||
|
fmt.Fprintf(&sb, "%s\tsession/%s/agent/%s\n", ag.ID(), name, ag.Name())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return []byte(sb.String()), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
|
|
||||||
|
// bypass/
|
||||||
virtfs.DirNode("bypass",
|
virtfs.DirNode("bypass",
|
||||||
virtfs.Doc("Bypass broker for sandboxed command approval"),
|
virtfs.Doc("Bypass broker for sandboxed command approval"),
|
||||||
virtfs.FileNode("policy", 0666,
|
virtfs.FileNode("policy", 0666,
|
||||||
virtfs.Doc("Global bypass policy"),
|
virtfs.Doc("Global bypass policy"),
|
||||||
virtfs.Read(func() ([]byte, error) { return readBypassPolicy(broker) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
virtfs.Write(func(data []byte) error { return writeBypassPolicy(broker, data) }),
|
if broker == nil {
|
||||||
|
return nil, errNoBypass
|
||||||
|
}
|
||||||
|
p := broker.GlobalPolicy().Global()
|
||||||
|
return p.Marshal()
|
||||||
|
}),
|
||||||
|
virtfs.Write(func(data []byte) error {
|
||||||
|
if broker == nil {
|
||||||
|
return errNoBypass
|
||||||
|
}
|
||||||
|
var p bypass.Policy
|
||||||
|
if err := bypass.ParsePolicy(data, &p); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return broker.GlobalPolicy().SetGlobal(p)
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.Each("{reqid}", func() ([]virtfs.FsNodeDecl, error) {
|
virtfs.Each("{reqid}", func() ([]virtfs.FsNodeDecl, error) {
|
||||||
return adaptBypassBindings(broker)
|
if broker == nil {
|
||||||
|
return nil, errNoBypass
|
||||||
|
}
|
||||||
|
pending := broker.Pending()
|
||||||
|
var out []virtfs.FsNodeDecl
|
||||||
|
for _, r := range pending {
|
||||||
|
req := r
|
||||||
|
out = append(out, virtfs.FsNodeDecl{
|
||||||
|
Name: req.ID,
|
||||||
|
Children: []virtfs.FsNodeDecl{
|
||||||
|
virtfs.FileNode("request", 0666,
|
||||||
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
return []byte(req.Summary() + "\n"), nil
|
||||||
|
}),
|
||||||
|
virtfs.Write(func(data []byte) error {
|
||||||
|
cmd := strings.TrimSpace(string(data))
|
||||||
|
var res bypass.Resolution
|
||||||
|
switch cmd {
|
||||||
|
case "approve":
|
||||||
|
res = bypass.ResolveApprove
|
||||||
|
case "deny":
|
||||||
|
res = bypass.ResolveDeny
|
||||||
|
case "persist":
|
||||||
|
res = bypass.ResolvePersist
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unknown resolution: %s (use approve/deny/persist)", cmd)
|
||||||
|
}
|
||||||
|
if !broker.Resolve(req.ID, res) {
|
||||||
|
return fmt.Errorf("request already resolved or timed out")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
|
|
||||||
|
// session/
|
||||||
virtfs.DirNode("session",
|
virtfs.DirNode("session",
|
||||||
virtfs.Doc("Session management"),
|
virtfs.Doc("Session management"),
|
||||||
virtfs.FileNode("new", 0666,
|
virtfs.FileNode("new", 0666,
|
||||||
virtfs.Doc("Create session"),
|
virtfs.Doc("Create session"),
|
||||||
virtfs.Request(requestSessionNew),
|
virtfs.Request(func(data []byte) ([]byte, error) {
|
||||||
|
args := strings.Fields(string(data))
|
||||||
|
name, remote := "", ""
|
||||||
|
for _, arg := range args {
|
||||||
|
if k, v, ok := strings.Cut(arg, "="); ok {
|
||||||
|
switch k {
|
||||||
|
case "name":
|
||||||
|
name = v
|
||||||
|
case "remote":
|
||||||
|
remote = v
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sess, err := session.CreateEmpty(name, remote)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return []byte(sess.Name() + "\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("idx", 0444,
|
virtfs.FileNode("idx", 0444,
|
||||||
virtfs.Doc("Session index"),
|
virtfs.Doc("Session index"),
|
||||||
virtfs.Read(readSessionIdx),
|
virtfs.Read(func() ([]byte, error) { return session.BuildIndex(), nil }),
|
||||||
),
|
),
|
||||||
virtfs.Each("{sname}", func() ([]virtfs.FsNodeDecl, error) {
|
virtfs.Each("{sname}", func() ([]virtfs.FsNodeDecl, error) {
|
||||||
return buildSessionBindings(rs, broker)
|
return buildSessionEntries(rs, broker)
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
// buildSessionBindings produces bindings for each session.
|
var errNoBypass = fmt.Errorf("bypass broker not available")
|
||||||
func buildSessionBindings(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) {
|
|
||||||
|
// buildSessionEntries produces one FsNodeDecl per session.
|
||||||
|
func buildSessionEntries(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) {
|
||||||
sessions := session.Sessions()
|
sessions := session.Sessions()
|
||||||
var out []virtfs.FsNodeDecl
|
var out []virtfs.FsNodeDecl
|
||||||
|
|
||||||
for name, sess := range sessions {
|
for name, sess := range sessions {
|
||||||
node := rs.GetOrCreateNode(sess)
|
node := rs.GetOrCreateNode(sess)
|
||||||
s := node // capture
|
s := node
|
||||||
n := name // capture
|
n := name
|
||||||
|
|
||||||
removeFn := func() error {
|
removeFn := func() error {
|
||||||
session.Kill(n)
|
session.Kill(n)
|
||||||
|
|
@ -144,7 +300,7 @@ func buildSessionBindings(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNode
|
||||||
Aliases: []string{sess.ID},
|
Aliases: []string{sess.ID},
|
||||||
Remove: removeFn,
|
Remove: removeFn,
|
||||||
Rename: renameFn,
|
Rename: renameFn,
|
||||||
Children: buildSessionChildren(s, rs, broker, removeFn, renameFn),
|
Children: buildSessionChildren(s, broker, removeFn, renameFn),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
return out, nil
|
return out, nil
|
||||||
|
|
@ -153,51 +309,126 @@ func buildSessionBindings(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNode
|
||||||
// buildSessionChildren returns the file nodes for a single session.
|
// buildSessionChildren returns the file nodes for a single session.
|
||||||
func buildSessionChildren(
|
func buildSessionChildren(
|
||||||
s *SessionNode,
|
s *SessionNode,
|
||||||
rs *RootState,
|
|
||||||
broker *bypass.Broker,
|
broker *bypass.Broker,
|
||||||
removeFn func() error,
|
removeFn func() error,
|
||||||
renameFn func(string) error,
|
renameFn func(string) error,
|
||||||
) []virtfs.FsNodeDecl {
|
) []virtfs.FsNodeDecl {
|
||||||
return []virtfs.FsNodeDecl{
|
return []virtfs.FsNodeDecl{
|
||||||
virtfs.FileNode("env", 0444,
|
virtfs.FileNode("env", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readSessionEnv(s) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
var sb strings.Builder
|
||||||
|
fmt.Fprintf(&sb, "OLLIE_SESSION_ID=%s\n", s.ID)
|
||||||
|
for _, e := range os.Environ() {
|
||||||
|
if strings.HasPrefix(e, "OLLIE_") && !strings.HasPrefix(e, "OLLIE_SESSION_ID=") {
|
||||||
|
sb.WriteString(e)
|
||||||
|
sb.WriteByte('\n')
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return []byte(sb.String()), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("paused", 0444,
|
virtfs.FileNode("paused", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readSessionPaused(s) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
if s.IsPaused() {
|
||||||
|
return []byte("true\n"), nil
|
||||||
|
}
|
||||||
|
return []byte("false\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("connected", 0444,
|
virtfs.FileNode("connected", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readSessionConnected(s) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
if s.IsConnected() {
|
||||||
|
return []byte("true\n"), nil
|
||||||
|
}
|
||||||
|
return []byte("false\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("ctl", 0666,
|
virtfs.FileNode("ctl", 0666,
|
||||||
virtfs.Request(sessionCtlHandler(s, removeFn, renameFn)),
|
virtfs.Request(func(data []byte) ([]byte, error) {
|
||||||
|
return dispatch(map[string]func([]string) ([]byte, error){
|
||||||
|
"kill": func(_ []string) ([]byte, error) { return []byte("ok\n"), removeFn() },
|
||||||
|
".": func(_ []string) ([]byte, error) { return []byte("ok\n"), removeFn() },
|
||||||
|
"save": func(_ []string) ([]byte, error) {
|
||||||
|
s.Save()
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"invalidate": func(_ []string) ([]byte, error) {
|
||||||
|
s.InvalidateModelsCache()
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"pause": func(_ []string) ([]byte, error) {
|
||||||
|
if err := s.Pause(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"resume": func(_ []string) ([]byte, error) {
|
||||||
|
if err := s.Resume(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
}, data)
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("name", 0666,
|
virtfs.FileNode("name", 0666,
|
||||||
virtfs.Read(func() ([]byte, error) { return readSessionName(s) }),
|
virtfs.Read(func() ([]byte, error) { return []byte(s.Name() + "\n"), nil }),
|
||||||
virtfs.Write(func(data []byte) error { return writeSessionName(s, renameFn, data) }),
|
virtfs.Write(func(data []byte) error {
|
||||||
|
newName := strings.TrimSpace(string(data))
|
||||||
|
if newName == "" || newName == s.Name() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return renameFn(newName)
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("id", 0444,
|
virtfs.FileNode("id", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readSessionID(s) }),
|
virtfs.Read(func() ([]byte, error) { return []byte(s.ID + "\n"), nil }),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("bypass", 0666,
|
virtfs.FileNode("bypass", 0666,
|
||||||
virtfs.Read(func() ([]byte, error) { return readSessionBypass(broker, s) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
virtfs.Write(func(data []byte) error { return writeSessionBypass(broker, s, data) }),
|
if broker == nil {
|
||||||
|
return nil, errNoBypass
|
||||||
|
}
|
||||||
|
p := broker.SessionPolicy(s.ID)
|
||||||
|
return p.Marshal()
|
||||||
|
}),
|
||||||
|
virtfs.Write(func(data []byte) error {
|
||||||
|
if broker == nil {
|
||||||
|
return errNoBypass
|
||||||
|
}
|
||||||
|
var p bypass.Policy
|
||||||
|
if err := bypass.ParsePolicy(data, &p); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
broker.SetSessionPolicy(s.ID, p)
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.DirNode("agent",
|
virtfs.DirNode("agent",
|
||||||
virtfs.FileNode("new", 0666,
|
virtfs.FileNode("new", 0666,
|
||||||
virtfs.Request(func(data []byte) ([]byte, error) { return requestAgentNew(s, data) }),
|
virtfs.Request(func(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)
|
||||||
|
}
|
||||||
|
al := NewAgentLog("")
|
||||||
|
s.SetAgentLog(ag.ID(), al)
|
||||||
|
wireAgentEvents(s.ID, ag, al)
|
||||||
|
return []byte(ag.ID() + "\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("idx", 0444,
|
virtfs.FileNode("idx", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentIdx(s.Session) }),
|
virtfs.Read(func() ([]byte, error) { return session.BuildAgentIndex(s.Session), nil }),
|
||||||
),
|
),
|
||||||
virtfs.Each("{aname}", func() ([]virtfs.FsNodeDecl, error) {
|
virtfs.Each("{aname}", func() ([]virtfs.FsNodeDecl, error) {
|
||||||
return buildAgentBindings(s, rs)
|
return buildAgentEntries(s)
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// buildAgentBindings produces bindings for each agent in a session.
|
// buildAgentEntries produces one FsNodeDecl per agent in a session.
|
||||||
func buildAgentBindings(s *SessionNode, rs *RootState) ([]virtfs.FsNodeDecl, error) {
|
func buildAgentEntries(s *SessionNode) ([]virtfs.FsNodeDecl, error) {
|
||||||
agents := s.Agents()
|
agents := s.Agents()
|
||||||
var out []virtfs.FsNodeDecl
|
var out []virtfs.FsNodeDecl
|
||||||
|
|
||||||
|
|
@ -226,26 +457,81 @@ func buildAgentBindings(s *SessionNode, rs *RootState) ([]virtfs.FsNodeDecl, err
|
||||||
|
|
||||||
// buildAgentChildren returns the file nodes for a single agent.
|
// buildAgentChildren returns the file nodes for a single agent.
|
||||||
func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.FsNodeDecl {
|
func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.FsNodeDecl {
|
||||||
|
chatStat := func() os.FileInfo {
|
||||||
|
return &SyntheticFileInfo{Name_: "chat", Mode_: 0444, Size_: 64 * 1024}
|
||||||
|
}
|
||||||
|
|
||||||
return []virtfs.FsNodeDecl{
|
return []virtfs.FsNodeDecl{
|
||||||
virtfs.FileNode("prompt", 0666,
|
virtfs.FileNode("prompt", 0666,
|
||||||
virtfs.Write(func(data []byte) error { return writeAgentPrompt(s, a, al, data) }),
|
virtfs.Write(func(data []byte) error {
|
||||||
|
input := strings.TrimSpace(string(data))
|
||||||
|
if input == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if input == "/invalidate" {
|
||||||
|
s.InvalidateModelsCache()
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
go func() {
|
||||||
|
a.Submit(s.Ctx, input)
|
||||||
|
al.EnsureTrailingNewline()
|
||||||
|
}()
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("fifo", 0666,
|
virtfs.FileNode("fifo", 0666,
|
||||||
virtfs.Write(func(data []byte) error { return writeAgentFifoIn(a, data) }),
|
virtfs.Write(func(data []byte) error {
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentFifoOut(a) }),
|
input := strings.TrimSpace(string(data))
|
||||||
|
if input == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
a.Queue(input)
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
item, ok := a.PopQueue()
|
||||||
|
if !ok {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
return []byte(item), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("chat", 0444,
|
virtfs.FileNode("chat", 0444,
|
||||||
virtfs.StatOverride(statAgentChat),
|
virtfs.StatOverride(chatStat),
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentChatText(al) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
al.mu.RLock()
|
||||||
|
defer al.mu.RUnlock()
|
||||||
|
return stripMarkers(al.log), nil
|
||||||
|
}),
|
||||||
virtfs.Stream(func(cctx context.Context, base string) ([]byte, string, error) {
|
virtfs.Stream(func(cctx context.Context, base string) ([]byte, string, error) {
|
||||||
return streamAgentChatText(s, al, cctx, base)
|
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
|
||||||
|
}
|
||||||
|
base = nextBase
|
||||||
|
}
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("chat.raw", 0444,
|
virtfs.FileNode("chat.raw", 0444,
|
||||||
virtfs.StatOverride(statAgentChat),
|
virtfs.StatOverride(chatStat),
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentChat(al) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
al.mu.RLock()
|
||||||
|
defer al.mu.RUnlock()
|
||||||
|
data := make([]byte, len(al.log))
|
||||||
|
copy(data, al.log)
|
||||||
|
return data, nil
|
||||||
|
}),
|
||||||
virtfs.Stream(func(cctx context.Context, base string) ([]byte, string, error) {
|
virtfs.Stream(func(cctx context.Context, base string) ([]byte, string, error) {
|
||||||
return streamAgentChat(s, al, cctx, base)
|
merged, cancel := mergeCtx(s.Ctx, cctx)
|
||||||
|
defer cancel()
|
||||||
|
return streamChat(al, merged, base)
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("statewait", 0444,
|
virtfs.FileNode("statewait", 0444,
|
||||||
|
|
@ -253,109 +539,386 @@ func buildAgentChildren(a *agent.Agent, al *AgentLog, s *SessionNode) []virtfs.F
|
||||||
return []byte(a.State() + "\n"), nil
|
return []byte(a.State() + "\n"), nil
|
||||||
}),
|
}),
|
||||||
virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) {
|
virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) {
|
||||||
return blockAgentStateWait(s, a, cctx, base)
|
merged, cancel := mergeCtx(s.Ctx, cctx)
|
||||||
|
defer cancel()
|
||||||
|
if base == "" {
|
||||||
|
base = a.State()
|
||||||
|
}
|
||||||
|
v, ok := a.WaitChange(merged, agent.WatchState, base)
|
||||||
|
if !ok {
|
||||||
|
st := a.State()
|
||||||
|
return []byte(st + "\n"), st, nil
|
||||||
|
}
|
||||||
|
return []byte(v + "\n"), v, nil
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("log", 0444,
|
virtfs.FileNode("log", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentLog(al) }),
|
virtfs.Read(func() ([]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
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("plan", 0666,
|
virtfs.FileNode("plan", 0666,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentPlan(al) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
virtfs.Write(func(data []byte) error { return writeAgentPlan(al, data) }),
|
al.mu.RLock()
|
||||||
|
data := make([]byte, len(al.plan))
|
||||||
|
copy(data, al.plan)
|
||||||
|
al.mu.RUnlock()
|
||||||
|
return data, nil
|
||||||
|
}),
|
||||||
|
virtfs.Write(func(data []byte) error {
|
||||||
|
al.mu.Lock()
|
||||||
|
al.plan = make([]byte, len(data))
|
||||||
|
copy(al.plan, data)
|
||||||
|
al.mu.Unlock()
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("cfg", 0666,
|
virtfs.FileNode("cfg", 0666,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentCfg(s, a, al) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
virtfs.Write(func(data []byte) error { return writeAgentCfg(a, data) }),
|
al.mu.RLock()
|
||||||
|
defer al.mu.RUnlock()
|
||||||
|
p := a.GenParams()
|
||||||
|
var sb strings.Builder
|
||||||
|
fmt.Fprintf(&sb, "name=%s\n", s.ID)
|
||||||
|
fmt.Fprintf(&sb, "backend=%s\n", a.BackendName())
|
||||||
|
fmt.Fprintf(&sb, "model=%s\n", a.ModelName())
|
||||||
|
fmt.Fprintf(&sb, "id=%s\n", a.ID())
|
||||||
|
fmt.Fprintf(&sb, "profile=%s\n", a.Profile())
|
||||||
|
fmt.Fprintf(&sb, "displayName=%s\n", a.Name())
|
||||||
|
fmt.Fprintf(&sb, "cwd=%s\n", a.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
|
||||||
|
}),
|
||||||
|
virtfs.Write(func(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]) != "" {
|
||||||
|
a.SetName(strings.TrimSpace(parts[1]))
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unknown cfg key: %s", parts[0])
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("ctl", 0666,
|
virtfs.FileNode("ctl", 0666,
|
||||||
virtfs.Request(agentCtlHandler(s, a)),
|
virtfs.Request(func(data []byte) ([]byte, error) {
|
||||||
|
return dispatch(map[string]func([]string) ([]byte, error){
|
||||||
|
"kill": func(_ []string) ([]byte, error) {
|
||||||
|
s.RemoveAgent(a.ID())
|
||||||
|
session.PublishEvent("session."+s.ID+".agent."+a.ID()+".kill", "")
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"stop": func(_ []string) ([]byte, error) {
|
||||||
|
a.Interrupt(agent.ErrInterrupted)
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"compact": func(_ []string) ([]byte, error) {
|
||||||
|
if err := a.Compact(s.Ctx); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"clear": func(_ []string) ([]byte, error) {
|
||||||
|
if err := a.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")
|
||||||
|
}
|
||||||
|
a.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")
|
||||||
|
}
|
||||||
|
a.InjectRewrite(text)
|
||||||
|
return []byte("ok\n"), nil
|
||||||
|
},
|
||||||
|
"agent": func(args []string) ([]byte, error) {
|
||||||
|
if len(args) == 0 {
|
||||||
|
return []byte(a.Profile() + "\n"), nil
|
||||||
|
}
|
||||||
|
if err := a.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 := a.Backend(); be != nil {
|
||||||
|
return []byte(be.Model() + "\n"), nil
|
||||||
|
}
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if be := a.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) {
|
||||||
|
ts := a.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
|
||||||
|
},
|
||||||
|
"tool_load": func(args []string) ([]byte, error) {
|
||||||
|
if len(args) == 0 {
|
||||||
|
return nil, fmt.Errorf("tool_load requires a tool name")
|
||||||
|
}
|
||||||
|
ts := a.ToolServer()
|
||||||
|
if ts == nil {
|
||||||
|
return nil, fmt.Errorf("no tool server")
|
||||||
|
}
|
||||||
|
if err := ts.LoadTool(args[0]); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
a.RefreshTools()
|
||||||
|
return []byte(args[0] + "\n"), nil
|
||||||
|
},
|
||||||
|
"backend": func(args []string) ([]byte, error) {
|
||||||
|
if len(args) == 0 {
|
||||||
|
if be := a.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(a.Name() + "\n"), nil
|
||||||
|
}
|
||||||
|
a.SetName(strings.Join(args, " "))
|
||||||
|
return []byte(strings.Join(args, " ") + "\n"), nil
|
||||||
|
},
|
||||||
|
"cwd": func(args []string) ([]byte, error) {
|
||||||
|
if len(args) == 0 {
|
||||||
|
return []byte(a.Cwd() + "\n"), nil
|
||||||
|
}
|
||||||
|
dir := paths.ExpandHome(strings.Join(args, " "))
|
||||||
|
if _, err := os.Stat(dir); err != nil {
|
||||||
|
return nil, fmt.Errorf("cwd: %w", err)
|
||||||
|
}
|
||||||
|
a.SetCWD(dir)
|
||||||
|
return []byte(dir + "\n"), nil
|
||||||
|
},
|
||||||
|
"systemprompt": func(_ []string) ([]byte, error) {
|
||||||
|
return []byte(a.SystemPrompt() + "\n"), nil
|
||||||
|
},
|
||||||
|
"help": func(_ []string) ([]byte, error) {
|
||||||
|
return []byte("stop compact clear inject agent model models tools tool_load cwd name backend systemprompt help\n"), nil
|
||||||
|
},
|
||||||
|
}, data)
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("stats", 0444,
|
virtfs.FileNode("stats", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentStats(a) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
return []byte("usage=" + a.UsageStr() + "\ncost=" + a.CostStr() + "\nctxsz=" + a.CtxSz() + "\n"), nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("tools", 0666,
|
virtfs.FileNode("tools", 0666,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentTools(a) }),
|
virtfs.Read(func() ([]byte, error) {
|
||||||
virtfs.Write(func(data []byte) error { return writeAgentTools(a, data) }),
|
ts := a.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
|
||||||
|
}),
|
||||||
|
virtfs.Write(func(data []byte) error {
|
||||||
|
name := strings.TrimSpace(string(data))
|
||||||
|
if name == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
ts := a.ToolServer()
|
||||||
|
if ts == nil {
|
||||||
|
return fmt.Errorf("no tool server")
|
||||||
|
}
|
||||||
|
if err := ts.LoadTool(name); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
a.RefreshTools()
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("name", 0666,
|
virtfs.FileNode("name", 0666,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentName(a) }),
|
virtfs.Read(func() ([]byte, error) { return []byte(a.Name() + "\n"), nil }),
|
||||||
virtfs.Write(func(data []byte) error { return writeAgentName(s, a, data) }),
|
virtfs.Write(func(data []byte) error {
|
||||||
|
newName := strings.TrimSpace(string(data))
|
||||||
|
if newName == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
a.SetName(newName)
|
||||||
|
session.PersistSession(s.Name())
|
||||||
|
return nil
|
||||||
|
}),
|
||||||
),
|
),
|
||||||
virtfs.FileNode("id", 0444,
|
virtfs.FileNode("id", 0444,
|
||||||
virtfs.Read(func() ([]byte, error) { return readAgentID(a) }),
|
virtfs.Read(func() ([]byte, error) { return []byte(a.ID() + "\n"), nil }),
|
||||||
),
|
),
|
||||||
virtfs.DirNode("proc",
|
virtfs.DirNode("proc",
|
||||||
virtfs.Each("{pid}", func() ([]virtfs.FsNodeDecl, error) {
|
virtfs.Each("{pid}", func() ([]virtfs.FsNodeDecl, error) {
|
||||||
return adaptProcessBindings(s)
|
conn := s.DialToolServer()
|
||||||
|
if conn == nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
defer conn.Close()
|
||||||
|
raw := conn.ListDetachedRaw()
|
||||||
|
var out []virtfs.FsNodeDecl
|
||||||
|
for _, r := range raw {
|
||||||
|
m, ok := r.(map[string]any)
|
||||||
|
if !ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
id, ok := m["pid"].(int)
|
||||||
|
if !ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
pid := id
|
||||||
|
out = append(out, virtfs.FsNodeDecl{
|
||||||
|
Name: strconv.Itoa(pid),
|
||||||
|
Children: []virtfs.FsNodeDecl{
|
||||||
|
virtfs.FileNode("out", 0444,
|
||||||
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
c := s.DialToolServer()
|
||||||
|
if c == nil {
|
||||||
|
return nil, fmt.Errorf("no tool server")
|
||||||
|
}
|
||||||
|
defer c.Close()
|
||||||
|
output, err := c.GetDetachedOutput(pid)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return []byte(output), nil
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
virtfs.FileNode("ctl", 0222,
|
||||||
|
virtfs.Write(func(data []byte) error {
|
||||||
|
c := s.DialToolServer()
|
||||||
|
if c == nil {
|
||||||
|
return fmt.Errorf("no tool server")
|
||||||
|
}
|
||||||
|
defer c.Close()
|
||||||
|
cmd := strings.TrimSpace(string(data))
|
||||||
|
switch {
|
||||||
|
case cmd == "term":
|
||||||
|
return c.SignalDetached(pid, syscall.SIGTERM)
|
||||||
|
case cmd == "kill":
|
||||||
|
return c.SignalDetached(pid, syscall.SIGKILL)
|
||||||
|
case cmd == "dismiss":
|
||||||
|
if !c.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 c.SignalDetached(pid, syscall.Signal(sig))
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unknown command: %s (valid: term, kill, signal <n>, dismiss)", cmd)
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
virtfs.FileNode("status", 0444,
|
||||||
|
virtfs.Read(func() ([]byte, error) {
|
||||||
|
c := s.DialToolServer()
|
||||||
|
if c == nil {
|
||||||
|
return nil, fmt.Errorf("no tool server")
|
||||||
|
}
|
||||||
|
defer c.Close()
|
||||||
|
for _, r := range c.ListDetachedRaw() {
|
||||||
|
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)
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// adaptBypassBindings returns bindings for pending bypass requests.
|
|
||||||
func adaptBypassBindings(broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) {
|
|
||||||
if broker == nil {
|
|
||||||
return nil, errNoBypass
|
|
||||||
}
|
|
||||||
pending := broker.Pending()
|
|
||||||
var out []virtfs.FsNodeDecl
|
|
||||||
for _, r := range pending {
|
|
||||||
req := r
|
|
||||||
out = append(out, virtfs.FsNodeDecl{
|
|
||||||
Name: req.ID,
|
|
||||||
Children: []virtfs.FsNodeDecl{
|
|
||||||
virtfs.FileNode("request", 0666,
|
|
||||||
virtfs.Read(func() ([]byte, error) {
|
|
||||||
return readBypassRequest(req)
|
|
||||||
}),
|
|
||||||
virtfs.Write(func(data []byte) error {
|
|
||||||
return writeBypassRequest(broker, req, data)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
},
|
|
||||||
})
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// adaptProcessBindings returns bindings for detached processes.
|
|
||||||
func adaptProcessBindings(s *SessionNode) ([]virtfs.FsNodeDecl, error) {
|
|
||||||
if s == nil {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
conn := s.DialToolServer()
|
|
||||||
if conn == nil {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
defer conn.Close()
|
|
||||||
raw := conn.ListDetachedRaw()
|
|
||||||
var out []virtfs.FsNodeDecl
|
|
||||||
for _, r := range raw {
|
|
||||||
if m, ok := r.(map[string]any); ok {
|
|
||||||
if id, ok := m["pid"].(int); ok {
|
|
||||||
pid := id
|
|
||||||
out = append(out, virtfs.FsNodeDecl{
|
|
||||||
Name: strconv.Itoa(pid),
|
|
||||||
Children: []virtfs.FsNodeDecl{
|
|
||||||
virtfs.FileNode("out", 0444,
|
|
||||||
virtfs.Read(func() ([]byte, error) {
|
|
||||||
return readProcess(s, pid)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
virtfs.FileNode("ctl", 0222,
|
|
||||||
virtfs.Write(func(data []byte) error {
|
|
||||||
return writeProcessCtl(s, pid, data)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
virtfs.FileNode("status", 0444,
|
|
||||||
virtfs.Read(func() ([]byte, error) {
|
|
||||||
return readProcessStatus(s, pid)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
},
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,120 @@
|
||||||
|
package fs
|
||||||
|
|
||||||
|
// Shared utilities for the 9P namespace handlers.
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
"ollie/cmd/olliesrv/internal/agent"
|
||||||
|
"ollie/cmd/olliesrv/internal/session"
|
||||||
|
"ollie/virtfs"
|
||||||
|
)
|
||||||
|
|
||||||
|
// mergeCtx returns a context that cancels when either parent or child cancels.
|
||||||
|
func mergeCtx(parent, child context.Context) (context.Context, context.CancelFunc) {
|
||||||
|
merged, cancel := context.WithCancel(parent)
|
||||||
|
stop := context.AfterFunc(child, func() { cancel() })
|
||||||
|
return merged, func() { stop(); cancel() }
|
||||||
|
}
|
||||||
|
|
||||||
|
// wireAgentEvents connects an agent's output and state change events.
|
||||||
|
func wireAgentEvents(sessID string, ag *agent.Agent, al *AgentLog) {
|
||||||
|
ag.SetOutput(NewEventHandler(al))
|
||||||
|
ag.SetOnStateChange(func(agentID, state string) {
|
||||||
|
session.PublishEvent("session."+sessID+".agent."+agentID+".state", state)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// dispatch routes a command to the matching handler based on the first word.
|
||||||
|
func dispatch(handlers map[string]func([]string) ([]byte, error), data []byte) ([]byte, error) {
|
||||||
|
input := strings.TrimSpace(string(data))
|
||||||
|
if input == "" {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
parts := strings.Fields(input)
|
||||||
|
h, ok := handlers[parts[0]]
|
||||||
|
if !ok {
|
||||||
|
return nil, fmt.Errorf("unknown ctl command: %s", parts[0])
|
||||||
|
}
|
||||||
|
return h(parts[1:])
|
||||||
|
}
|
||||||
|
|
||||||
|
// streamChat blocks until new data is available in the agent log.
|
||||||
|
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
|
||||||
|
}
|
||||||
|
|
||||||
|
// stripMarkers removes [[[...]]] marker lines and source fence delimiter lines
|
||||||
|
// from raw chat data. Content inside fences is preserved — only the ``` lines
|
||||||
|
// themselves are stripped.
|
||||||
|
func stripMarkers(data []byte) []byte {
|
||||||
|
var out []byte
|
||||||
|
for len(data) > 0 {
|
||||||
|
nl := bytes.IndexByte(data, '\n')
|
||||||
|
if nl < 0 {
|
||||||
|
out = append(out, data...)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
line := data[:nl+1]
|
||||||
|
data = data[nl+1:]
|
||||||
|
|
||||||
|
trimmed := bytes.TrimRight(line, "\n")
|
||||||
|
if len(trimmed) >= 7 && bytes.HasPrefix(trimmed, []byte("[[[")) && bytes.HasSuffix(trimmed, []byte("]]]")) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
stripped := bytes.TrimSpace(trimmed)
|
||||||
|
if bytes.HasPrefix(stripped, []byte("```")) && !bytes.HasPrefix(stripped, []byte("````")) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
out = append(out, line...)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// helpSpec stores the spec for help generation. Set during tree build.
|
||||||
|
var helpSpec *virtfs.FsNodeDecl
|
||||||
|
|
||||||
|
var (
|
||||||
|
helpOnce sync.Once
|
||||||
|
helpText string
|
||||||
|
helpFn = func() string {
|
||||||
|
helpOnce.Do(func() {
|
||||||
|
if helpSpec != nil {
|
||||||
|
helpText = virtfs.GenerateHelp(*helpSpec)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
return helpText
|
||||||
|
}
|
||||||
|
)
|
||||||
Loading…
Reference in New Issue