diff --git a/cmd/olliesrv/internal/fs/ctl.go b/cmd/olliesrv/internal/fs/ctl.go deleted file mode 100644 index 462921d..0000000 --- a/cmd/olliesrv/internal/fs/ctl.go +++ /dev/null @@ -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:]) -} diff --git a/cmd/olliesrv/internal/fs/handlers_agent.go b/cmd/olliesrv/internal/fs/handlers_agent.go deleted file mode 100644 index 0e3a672..0000000 --- a/cmd/olliesrv/internal/fs/handlers_agent.go +++ /dev/null @@ -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 — 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 — 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 , dismiss)", cmd) - } -} diff --git a/cmd/olliesrv/internal/fs/handlers_root.go b/cmd/olliesrv/internal/fs/handlers_root.go deleted file mode 100644 index a4ccc73..0000000 --- a/cmd/olliesrv/internal/fs/handlers_root.go +++ /dev/null @@ -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 -} diff --git a/cmd/olliesrv/internal/fs/handlers_session.go b/cmd/olliesrv/internal/fs/handlers_session.go deleted file mode 100644 index 3ee199c..0000000 --- a/cmd/olliesrv/internal/fs/handlers_session.go +++ /dev/null @@ -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 -} diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index 6c4610d..a7c37cb 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -1,16 +1,24 @@ package fs -// Filesystem specification — the single source of truth for the 9P namespace. -// Constructed via buildTreeSpec with closure capture. +// spec.go — the single source of truth for the olliesrv 9P namespace. +// Every handler is inline. No archaeological expeditions needed. import ( "context" + "encoding/json" + "fmt" + "os" "os/user" "strconv" + "strings" + "syscall" "ollie/cmd/olliesrv/internal/agent" + "ollie/cmd/olliesrv/internal/backend" "ollie/cmd/olliesrv/internal/bypass" "ollie/cmd/olliesrv/internal/session" + "ollie/paths" + "ollie/toolsrv" "ollie/virtfs" ) @@ -31,7 +39,7 @@ const ( 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. func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl { cfg := &rs.Cfg @@ -51,80 +59,228 @@ func buildTreeSpec(rs *RootState) virtfs.FsNodeDecl { virtfs.FileNode("backends", 0444, 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.Doc("Filesystem reference"), - virtfs.Read(readHelp), + virtfs.Read(func() ([]byte, error) { + return []byte(helpFn()), nil + }), ), virtfs.FileNode("models", 0444, virtfs.Doc("Available models. Format: backendmodel 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.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.Doc("Available tools. Format: namedescription 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.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.Doc("Blocking read for server events"), - virtfs.Read(readEventwait), - virtfs.BlockOnce(blockEventwait), + virtfs.Read(func() ([]byte, error) { + 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.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.Doc("Alias table. Format: idpath 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.Doc("Bypass broker for sandboxed command approval"), virtfs.FileNode("policy", 0666, virtfs.Doc("Global bypass policy"), - virtfs.Read(func() ([]byte, error) { return readBypassPolicy(broker) }), - virtfs.Write(func(data []byte) error { return writeBypassPolicy(broker, data) }), + virtfs.Read(func() ([]byte, error) { + 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) { - 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.Doc("Session management"), virtfs.FileNode("new", 0666, 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.Doc("Session index"), - virtfs.Read(readSessionIdx), + virtfs.Read(func() ([]byte, error) { return session.BuildIndex(), nil }), ), virtfs.Each("{sname}", func() ([]virtfs.FsNodeDecl, error) { - return buildSessionBindings(rs, broker) + return buildSessionEntries(rs, broker) }), ), ) } -// buildSessionBindings produces bindings for each session. -func buildSessionBindings(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) { +var errNoBypass = fmt.Errorf("bypass broker not available") + +// buildSessionEntries produces one FsNodeDecl per session. +func buildSessionEntries(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNodeDecl, error) { sessions := session.Sessions() var out []virtfs.FsNodeDecl for name, sess := range sessions { node := rs.GetOrCreateNode(sess) - s := node // capture - n := name // capture + s := node + n := name removeFn := func() error { session.Kill(n) @@ -144,7 +300,7 @@ func buildSessionBindings(rs *RootState, broker *bypass.Broker) ([]virtfs.FsNode Aliases: []string{sess.ID}, Remove: removeFn, Rename: renameFn, - Children: buildSessionChildren(s, rs, broker, removeFn, renameFn), + Children: buildSessionChildren(s, broker, removeFn, renameFn), }) } 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. func buildSessionChildren( s *SessionNode, - rs *RootState, broker *bypass.Broker, removeFn func() error, renameFn func(string) error, ) []virtfs.FsNodeDecl { return []virtfs.FsNodeDecl{ 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.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.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.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.Read(func() ([]byte, error) { return readSessionName(s) }), - virtfs.Write(func(data []byte) error { return writeSessionName(s, renameFn, data) }), + virtfs.Read(func() ([]byte, error) { return []byte(s.Name() + "\n"), nil }), + 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.Read(func() ([]byte, error) { return readSessionID(s) }), + virtfs.Read(func() ([]byte, error) { return []byte(s.ID + "\n"), nil }), ), virtfs.FileNode("bypass", 0666, - virtfs.Read(func() ([]byte, error) { return readSessionBypass(broker, s) }), - virtfs.Write(func(data []byte) error { return writeSessionBypass(broker, s, data) }), + virtfs.Read(func() ([]byte, error) { + 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.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.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) { - return buildAgentBindings(s, rs) + return buildAgentEntries(s) }), ), } } -// buildAgentBindings produces bindings for each agent in a session. -func buildAgentBindings(s *SessionNode, rs *RootState) ([]virtfs.FsNodeDecl, error) { +// buildAgentEntries produces one FsNodeDecl per agent in a session. +func buildAgentEntries(s *SessionNode) ([]virtfs.FsNodeDecl, error) { agents := s.Agents() 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. 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{ 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.Write(func(data []byte) error { return writeAgentFifoIn(a, data) }), - virtfs.Read(func() ([]byte, error) { return readAgentFifoOut(a) }), + virtfs.Write(func(data []byte) error { + 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.StatOverride(statAgentChat), - virtfs.Read(func() ([]byte, error) { return readAgentChatText(al) }), + virtfs.StatOverride(chatStat), + 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) { - 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.StatOverride(statAgentChat), - virtfs.Read(func() ([]byte, error) { return readAgentChat(al) }), + virtfs.StatOverride(chatStat), + 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) { - return streamAgentChat(s, al, cctx, base) + merged, cancel := mergeCtx(s.Ctx, cctx) + defer cancel() + return streamChat(al, merged, base) }), ), 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 }), 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.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.Read(func() ([]byte, error) { return readAgentPlan(al) }), - virtfs.Write(func(data []byte) error { return writeAgentPlan(al, data) }), + virtfs.Read(func() ([]byte, error) { + 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.Read(func() ([]byte, error) { return readAgentCfg(s, a, al) }), - virtfs.Write(func(data []byte) error { return writeAgentCfg(a, data) }), + virtfs.Read(func() ([]byte, error) { + 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.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.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.Read(func() ([]byte, error) { return readAgentTools(a) }), - virtfs.Write(func(data []byte) error { return writeAgentTools(a, data) }), + virtfs.Read(func() ([]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 + }), + 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.Read(func() ([]byte, error) { return readAgentName(a) }), - virtfs.Write(func(data []byte) error { return writeAgentName(s, a, data) }), + virtfs.Read(func() ([]byte, error) { return []byte(a.Name() + "\n"), nil }), + 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.Read(func() ([]byte, error) { return readAgentID(a) }), + virtfs.Read(func() ([]byte, error) { return []byte(a.ID() + "\n"), nil }), ), virtfs.DirNode("proc", 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 , 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 -} diff --git a/cmd/olliesrv/internal/fs/support.go b/cmd/olliesrv/internal/fs/support.go new file mode 100644 index 0000000..c4457c1 --- /dev/null +++ b/cmd/olliesrv/internal/fs/support.go @@ -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 + } +)