package fs // 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/exec" "os/user" "sort" "strconv" "strings" "syscall" "time" "ollie/cmd/olliesrv/internal/agent" "ollie/cmd/olliesrv/internal/backend" "ollie/cmd/olliesrv/internal/metrics" "ollie/cmd/olliesrv/internal/session" "ollie/util" "ollie/virtfs" ) // Type aliases for virtfs types used throughout the package. type ( Tree = virtfs.Tree File = virtfs.File SyntheticFileInfo = virtfs.SyntheticFileInfo ) var treeSpecUID, treeSpecGID string const asyncWorkLimit = 64 var asyncWorkSlots = make(chan struct{}, asyncWorkLimit) func startAsync(ctx context.Context, fn func()) bool { select { case asyncWorkSlots <- struct{}{}: case <-ctx.Done(): return false default: return false } go func() { defer func() { <-asyncWorkSlots }() fn() }() return true } // runWorkflow executes a workflow script in the background. // If variant is non-empty and not "default", the matching conf sidecar is // sourced as environment variables before the script runs. // If workflow is empty or "none", nothing happens. func runWorkflow(s *session.Session, workflow, variant string) { if workflow == "" || workflow == "none" { return } startAsync(s.Ctx, func() { workflowDir := util.CfgDir() + "/workflows/" scriptPath := workflowDir + workflow if _, err := os.Stat(scriptPath); err != nil { s.SetGoalStatus("error: workflow not found: " + workflow) return } cwd := s.Cwd() cmd := exec.CommandContext(s.Ctx, scriptPath) cmd.Env = append(os.Environ(), "OLLIE_SESSION_ID="+s.ID, "OLLIE_SESSION_NAME="+s.Name(), "OLLIE_CWD="+cwd, ) // Source variant conf if specified. if variant != "" && variant != "default" { confPath := workflowDir + workflow + "-" + variant + ".conf" confData, err := os.ReadFile(confPath) if err != nil { s.SetGoalStatus("error: variant conf not found: " + workflow + "-" + variant + ".conf") return } for _, line := range strings.Split(string(confData), "\n") { line = strings.TrimSpace(line) if line == "" || line[0] == '#' { continue } if _, _, ok := strings.Cut(line, "="); ok { cmd.Env = append(cmd.Env, line) } } } cmd.Dir = cwd out, err := cmd.CombinedOutput() if err != nil { msg := strings.TrimSpace(string(out)) if msg == "" { msg = err.Error() } s.SetGoalStatus("error: " + msg) } // Script exited — don't auto-set complete; the conductor agent handles that. }) } // buildTreeSpec constructs the full 9P namespace. func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl { mc := cfg.ModelCache u, _ := user.Current() treeSpecUID = "ollie" treeSpecGID = "agent" if u != nil { treeSpecUID = u.Username } // Event subscription for event (lives for server lifetime). eventRead, eventSignal := session.EventStream(session.SubscribeEvents(cfg.Ctx, "*")) return virtfs.DirNode("/", virtfs.UID(treeSpecUID), virtfs.GID(treeSpecGID), virtfs.FileNode("backends", 0444, virtfs.Doc("Available backend names, one per line"), virtfs.Read(func() ([]byte, error) { return []byte(strings.Join(backend.Backends(), "\n") + "\n"), nil }), ), virtfs.FileNode("help", 0444, virtfs.Doc("Filesystem reference"), virtfs.Read(func() ([]byte, error) { return []byte(helpFn()), nil }), ), virtfs.FileNode("models", 0444, virtfs.Doc("Available models. Format: backendmodel[inoutcache_readcache_write]. Pricing per 1M tokens USD; (e) suffix marks estimates."), 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(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("workflows", 0444, virtfs.Doc("Available workflows with variants. Format: name\\tdefault,variant1,variant2"), virtfs.Read(func() ([]byte, error) { dir := util.CfgDir() + "/workflows" entries, err := os.ReadDir(dir) if err != nil { return nil, nil } // Collect executable workflow scripts (no extension). workflows := make(map[string][]string) for _, e := range entries { if e.IsDir() { continue } name := e.Name() if strings.Contains(name, ".") { continue // skip .conf files } workflows[name] = []string{"default"} } // Find variant confs: {name}-{variant}.conf for _, e := range entries { name := e.Name() if !strings.HasSuffix(name, ".conf") { continue } base := strings.TrimSuffix(name, ".conf") // Find the longest workflow name that is a prefix. for wf := range workflows { if strings.HasPrefix(base, wf+"-") { variant := strings.TrimPrefix(base, wf+"-") if variant != "" { workflows[wf] = append(workflows[wf], variant) } } } } var sb strings.Builder // "none" is always the first entry — it means no workflow runs. sb.WriteString("none\tdefault\n") names := make([]string, 0, len(workflows)) for wf := range workflows { names = append(names, wf) } sort.Strings(names) for _, wf := range names { sb.WriteString(wf) sb.WriteByte('\t') sb.WriteString(strings.Join(workflows[wf], ",")) sb.WriteByte('\n') } return []byte(sb.String()), nil }), ), virtfs.FileNode("ctl", 0666, virtfs.Doc("Server control. Write: 'invalidate', 'kill'"), virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) { return dispatch([]ctlCmd{ {"invalidate", "clear the models cache", func(_ []string) ([]byte, error) { if mc != nil { mc.Invalidate() } if cfg.Invalidate != nil { cfg.Invalidate() } return []byte("ok\n"), nil }}, {"kill", "shut down the server", func(_ []string) ([]byte, error) { if cfg.Shutdown != nil { cfg.Shutdown() } return []byte("ok\n"), nil }}, }, data) }), ), virtfs.FileNode("event", 0666, virtfs.Doc("Event stream. Read: all events. Write filter then read: filtered events."), virtfs.Stream(eventRead, eventSignal), ), virtfs.FileNode("generate", 0666, virtfs.Doc("One-shot LLM generation"), virtfs.Rdwr(func(_ 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(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(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 }), ), virtfs.FileNode("stats", 0666, virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) { result, err := metrics.QueryFormat(string(data)) if err != nil { return nil, err } return []byte(result), nil }), ), virtfs.FileNode("metrics", 0444, virtfs.Read(func() ([]byte, error) { a, err := metrics.AggregateAll() if err != nil { return nil, err } return []byte(metrics.Format(a)), nil }), ), virtfs.FileNode("metrics.by-backend-model", 0444, virtfs.Read(func() ([]byte, error) { groups, err := metrics.GroupedAggregate("", "") if err != nil { return nil, err } return []byte(metrics.FormatGroups(groups)), nil }), ), // session/ virtfs.DirNode("session", virtfs.Doc("Session management"), virtfs.FileNode("new", 0666, virtfs.Doc("Create session"), virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) { args := strings.Fields(string(data)) name, remote, workflow, variant, cwd := "", "", "", "", "" yolo := false for _, arg := range args { if k, v, ok := strings.Cut(arg, "="); ok { switch k { case "name": name = v case "remote": remote = v case "workflow": workflow = v case "variant": variant = v case "cwd": cwd = v case "yolo": yolo = v == "true" } } } if cwd == "" { return nil, fmt.Errorf("cwd is required") } sess, created, err := session.CreateEmpty(name, remote, yolo) if err != nil { return nil, err } // Get-or-create: only a freshly created session takes the // provided configuration. An existing session is returned // untouched. if created { sess.SetCwd(cwd) if workflow != "" { sess.SetWorkflow(workflow) } if variant != "" { sess.SetVariant(variant) } } return []byte(sess.Name() + "\n"), nil }), ), virtfs.FileNode("idx", 0444, virtfs.Doc("Session index"), virtfs.Read(func() ([]byte, error) { return session.BuildIndex(), nil }), ), virtfs.Each("{sname}", func() ([]virtfs.FsNodeDecl, error) { return buildSessionEntries() }), ), ) } // buildSessionEntries produces one FsNodeDecl per session. func buildSessionEntries() ([]virtfs.FsNodeDecl, error) { sessions := session.Sessions() var out []virtfs.FsNodeDecl for name, sess := range sessions { s := sess n := name removeFn := func() error { session.Kill(n) return nil } renameFn := func(newName string) error { return session.Rename(n, newName) } out = append(out, virtfs.FsNodeDecl{ Name: s.Name(), Aliases: []string{s.ID}, Remove: removeFn, Rename: renameFn, Children: buildSessionChildren(s, removeFn, renameFn), }) } return out, nil } // buildSessionChildren returns the file nodes for a single session. func buildSessionChildren( s *session.Session, removeFn func() error, renameFn func(string) error, ) []virtfs.FsNodeDecl { return []virtfs.FsNodeDecl{ virtfs.FileNode("env", 0444, 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) { if s.IsPaused() { return []byte("true\n"), nil } return []byte("false\n"), nil }), ), virtfs.FileNode("goal", 0666, virtfs.Doc("Session goal text. Writing triggers the workflow if not already running."), virtfs.GID("agent"), virtfs.Read(func() ([]byte, error) { text, _ := s.Goal() if text == "" { return nil, nil } return []byte(text + "\n"), nil }), virtfs.Write(func(data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { s.ClearGoal() return nil } _, status := s.Goal() s.SetGoal(input) // Trigger workflow only if not already running. if status == "" || status == "complete" || status == "blocked" || strings.HasPrefix(status, "error") { runWorkflow(s, s.Workflow(), s.Variant()) } return nil }), ), virtfs.FileNode("goalstatus", 0666, virtfs.Doc("Goal status. Read/write."), virtfs.GID("agent"), virtfs.Read(func() ([]byte, error) { _, status := s.Goal() return []byte(status + "\n"), nil }), virtfs.Write(func(data []byte) error { status := strings.TrimSpace(string(data)) s.SetGoalStatus(status) return nil }), ), virtfs.FileNode("goalwait", 0444, virtfs.Doc("Blocks until goal status changes."), virtfs.BlockOnce(func() ([]byte, string, error) { _, status := s.Goal() return []byte(status + "\n"), status, nil }, s.GoalSignal), ), virtfs.FileNode("bypass", 0666, virtfs.Doc("Bypass request handling. Read: pending request JSON or empty. Write: 'id approve' or 'id deny'."), virtfs.Read(func() ([]byte, error) { req := s.BypassPending() if req == nil { return nil, nil } data, err := json.Marshal(req) if err != nil { return nil, err } return append(data, '\n'), nil }), virtfs.Write(func(data []byte) error { input := strings.TrimSpace(string(data)) parts := strings.SplitN(input, " ", 2) if len(parts) != 2 { return fmt.Errorf("expected 'id approve' or 'id deny'") } id, action := parts[0], parts[1] approved := action == "approve" return s.ResolveBypass(id, approved) }), ), virtfs.FileNode("bypasswait", 0444, virtfs.Doc("Blocks until a bypass request arrives."), virtfs.Read(func() ([]byte, error) { req := s.BypassPending() if req == nil { return nil, nil } data, err := json.Marshal(req) if err != nil { return nil, err } return append(data, '\n'), nil }), virtfs.BlockOnce(func() ([]byte, string, error) { req := s.BypassPending() if req == nil { return nil, "", nil } data, err := json.Marshal(req) if err != nil { return nil, "", err } return append(data, '\n'), req.ID, nil }, s.BypassSignal), ), virtfs.FileNode("stats", 0444, virtfs.Read(func() ([]byte, error) { a, err := metrics.AggregateSession(s.ID) if err != nil { return nil, err } return []byte(metrics.Format(a)), nil }), ), virtfs.FileNode("metrics.by-backend-model", 0444, virtfs.Read(func() ([]byte, error) { groups, err := metrics.GroupedAggregate(s.ID, "") if err != nil { return nil, err } return []byte(metrics.FormatGroups(groups)), nil }), ), virtfs.FileNode("ctl", 0666, virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) { return dispatch([]ctlCmd{ {"kill", "destroy this session", func(_ []string) ([]byte, error) { return []byte("ok\n"), removeFn() }}, {"save", "persist session state to disk", func(_ []string) ([]byte, error) { s.Save() return []byte("ok\n"), nil }}, {"invalidate", "clear the models cache", func(_ []string) ([]byte, error) { s.InvalidateModelsCache() return []byte("ok\n"), nil }}, {"pause", "pause all agents in this session", func(_ []string) ([]byte, error) { if err := s.Pause(); err != nil { return nil, err } return []byte("ok\n"), nil }}, {"resume", "resume all agents in this session", func(_ []string) ([]byte, error) { if err := s.Resume(); err != nil { return nil, err } return []byte("ok\n"), nil }}, {"run", "run a workflow: run [workflow] [variant]", func(args []string) ([]byte, error) { workflow := s.Workflow() variant := s.Variant() if len(args) > 0 { workflow = args[0] } if len(args) > 1 { variant = args[1] } runWorkflow(s, workflow, variant) return []byte("ok\n"), nil }}, }, data) }), ), virtfs.FileNode("name", 0666, 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("cfg", 0666, virtfs.Doc("Session configuration. Read: key=value lines. Write: single key=value to update."), virtfs.Read(func() ([]byte, error) { var sb strings.Builder fmt.Fprintf(&sb, "name=%s\n", s.Name()) fmt.Fprintf(&sb, "cwd=%s\n", s.Cwd()) fmt.Fprintf(&sb, "remote=%s\n", s.Remote) fmt.Fprintf(&sb, "workflow=%s\n", s.Workflow()) fmt.Fprintf(&sb, "variant=%s\n", s.Variant()) fmt.Fprintf(&sb, "yolo=%t\n", s.Yolo) return []byte(sb.String()), nil }), virtfs.Write(func(data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } for _, line := range strings.Split(input, "\n") { line = strings.TrimSpace(line) if line == "" { continue } k, v, ok := strings.Cut(line, "=") if !ok { return fmt.Errorf("invalid cfg format (expected key=value)") } switch k { case "workflow": s.SetWorkflow(v) case "variant": s.SetVariant(v) case "cwd": if v != "" { s.SetCwd(v) } default: return fmt.Errorf("unknown session cfg key: %s", k) } } return nil }), ), virtfs.FileNode("id", 0444, virtfs.Read(func() ([]byte, error) { return []byte(s.ID + "\n"), nil }), ), virtfs.DirNode("agent", virtfs.FileNode("new", 0666, virtfs.Doc("Create agent. With prompt=, runs as sub-agent: blocks until done, returns reply."), virtfs.Rdwr(func(ctx context.Context, data []byte) ([]byte, error) { req := parseAgentNewRequest(data) req.Params.ParentID = req.ParentID if req.Prompt == "" { // Persistent agent creation (no sub-agent mode). ag, err := session.CreateAgentWithParams(s.Name(), req.Params) if err != nil { return nil, fmt.Errorf("create agent: %w", err) } wireAgentEvents(s.ID, ag) return []byte(ag.ID() + "\n"), nil } // Sub-agent mode: enforce limits, create, run, destroy. maxDepth := req.MaxDepth if maxDepth <= 0 { maxDepth = 1 } maxParallel := 5 // default if req.MaxParallel != nil { maxParallel = *req.MaxParallel } if maxParallel == 0 { return nil, fmt.Errorf("sub-agent spawning is disabled (max_parallel=0)") } timeout := req.Timeout if timeout <= 0 { timeout = 600 } var parent *agent.Agent if req.ParentID != "" { parent = s.FindAgent(req.ParentID) if parent == nil { return nil, fmt.Errorf("parent agent %q not found", req.ParentID) } } if parent != nil { // Depth check. if parent.Depth()+1 > maxDepth { return nil, fmt.Errorf("sub-agent depth limit exceeded (max %d)", maxDepth) } // Parallelism check. if maxParallel > 0 && parent.ActiveChildren() >= int32(maxParallel) { return nil, fmt.Errorf("sub-agent parallelism limit exceeded (max %d)", maxParallel) } // Inherit context (optionally truncated at fork_at). msgs := parent.Messages() if req.ForkAt > 0 { msgs = messagesUpToTurn(msgs, req.ForkAt) } // Sanitize to remove dangling tool calls msgs = backend.SanitizeMessages(msgs) req.Params.History = agent.RestoreHistoryFromMessages(msgs) parent.IncChildren() defer parent.DecChildren() } ag, err := session.CreateAgentWithParams(s.Name(), req.Params) if err != nil { return nil, fmt.Errorf("create agent: %w", err) } wireAgentEvents(s.ID, ag) if parent != nil { ag.SetDepth(parent.Depth() + 1) } ag.SetEnv("OLLIE_SUBAGENT_DEPTH", fmt.Sprintf("%d", ag.Depth())) // Run with timeout. subCtx, cancel := context.WithTimeout(ctx, time.Duration(timeout)*time.Second) defer cancel() session.PublishEvent("session."+s.ID+".agent."+ag.ID()+".new", "") subPrompt := "You are a sub-agent. Your task is below. " + "Execute it completely using tools — investigate, implement, verify. " + "Do NOT respond with a plan or intentions. Do NOT say what you will do. " + "Call tools. Do the work. Your final text response must summarize what you ACCOMPLISHED.\n\n" + req.Prompt ag.Submit(subCtx, subPrompt) ag.EnsureTrailingNewline() reply := ag.Reply() s.RemoveAgent(ag.ID()) session.PublishEvent("session."+s.ID+".agent."+ag.ID()+".kill", "") if reply == "" { return []byte("(no reply)\n"), nil } return []byte(reply), nil }), ), virtfs.FileNode("idx", 0444, virtfs.Read(func() ([]byte, error) { return session.BuildAgentIndex(s), nil }), ), virtfs.Each("{aname}", func() ([]virtfs.FsNodeDecl, error) { return buildAgentEntries(s) }), ), } } // buildAgentEntries produces one FsNodeDecl per agent in a session. func buildAgentEntries(s *session.Session) ([]virtfs.FsNodeDecl, error) { agents := s.Agents() var out []virtfs.FsNodeDecl for _, ag := range agents { a := ag wireAgentEvents(s.ID, a) out = append(out, virtfs.FsNodeDecl{ Name: a.Name(), Aliases: []string{a.ID()}, Remove: func() error { s.RemoveAgent(a.ID()) session.PublishEvent("session."+s.ID+".agent."+a.ID()+".kill", "") return nil }, Children: buildAgentChildren(a, s), }) } return out, nil } // buildAgentChildren returns the file nodes for a single agent. func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl { chatStat := func() os.FileInfo { return &SyntheticFileInfo{Name_: "chat", Mode_: 0444, Size_: 64 * 1024} } stats := func(_ []string) ([]byte, error) { u := a.Usage() var sb strings.Builder fmt.Fprintf(&sb, "usage=%s\n", a.UsageStr()) fmt.Fprintf(&sb, "cost=%s\n", a.CostStr()) fmt.Fprintf(&sb, "ctxsz=%s\n", a.CtxSz()) if u != nil { fmt.Fprintf(&sb, "cachedInputTokens=%d\n", u.TotalCachedInputTokens) fmt.Fprintf(&sb, "cacheCreationTokens=%d\n", u.TotalCacheCreationTokens) fmt.Fprintf(&sb, "cacheHitRatio=%g\n", u.CacheHitRatio) } return []byte(sb.String()), nil } inject := func(args []string) ([]byte, error) { text := strings.Join(args, " ") if text == "" { return nil, fmt.Errorf("inject requires text") } if a.IsRunning() { a.InjectRewrite(text) } else { startAsync(s.Ctx, func() { a.Submit(s.Ctx, text) a.EnsureTrailingNewline() }) } return []byte("ok\n"), nil } return []virtfs.FsNodeDecl{ virtfs.FileNode("prompt", 0666, virtfs.Write(func(data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } if input == "/invalidate" { s.InvalidateModelsCache() return nil } startAsync(s.Ctx, func() { a.Submit(s.Ctx, input) a.EnsureTrailingNewline() }) return nil }), ), virtfs.FileNode("fifo", 0666, virtfs.Write(func(data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { return nil } if err := a.Queue(input); err != nil { return err } // If the agent is idle, trigger processing. if !a.IsRunning() { if !startAsync(s.Ctx, func() { if next, ok := a.PopQueue(); ok { a.Submit(s.Ctx, next) a.EnsureTrailingNewline() } }) { return fmt.Errorf("async work limit reached") } } return nil }), virtfs.Read(func() ([]byte, error) { item, ok := a.PopQueue() if !ok { return nil, nil } return []byte(item), nil }), ), virtfs.FileNode("feed", 0666, virtfs.Write(func(data []byte) error { if len(data) == 0 { return nil } for _, b := range data { if b != 0 { a.FeedWrite(data) return nil } } return nil }), virtfs.BlockOnce(func() ([]byte, string, error) { return a.Feed.Get(), a.Feed.Hash(), nil }, a.SignalCh), ), virtfs.FileNode("chat", 0444, virtfs.StatOverride(chatStat), virtfs.Read(func() ([]byte, error) { a.ChatMu().RLock() defer a.ChatMu().RUnlock() return stripMarkers(a.ChatLog()), nil }), virtfs.Stream(func(base string) ([]byte, string, error) { data, nextBase, err := a.ChatRead(base) if err != nil || len(data) == 0 { return data, nextBase, err } stripped := stripMarkers(data) if len(stripped) > 0 { return stripped, nextBase, nil } return nil, nextBase, nil }, a.ChatSignal), ), virtfs.FileNode("chat.raw", 0444, virtfs.StatOverride(chatStat), virtfs.Read(func() ([]byte, error) { a.ChatMu().RLock() defer a.ChatMu().RUnlock() log := a.ChatLog() data := make([]byte, len(log)) copy(data, log) return data, nil }), virtfs.Stream(a.ChatRead, a.ChatSignal), ), virtfs.FileNode("state", 0444, virtfs.Doc("Current agent state (idle, calling, thinking, paused)"), virtfs.Read(func() ([]byte, error) { return []byte(a.State() + "\n"), nil }), ), virtfs.FileNode("status", 0444, virtfs.Doc("Human-readable status: current activity and elapsed time"), virtfs.Read(func() ([]byte, error) { return []byte(a.Status() + "\n"), nil }), ), virtfs.FileNode("log", 0444, virtfs.Read(func() ([]byte, error) { const maxWindow = 64 * 1024 a.ChatMu().RLock() log := a.ChatLog() start := 0 if len(log) > maxWindow { start = len(log) - maxWindow } data := make([]byte, len(log)-start) copy(data, log[start:]) a.ChatMu().RUnlock() return data, nil }), ), virtfs.FileNode("plan", 0666, virtfs.Read(func() ([]byte, error) { return a.Plan(), nil }), virtfs.Write(func(data []byte) error { a.SetPlan(data) return nil }), ), virtfs.FileNode("cfg", 0666, virtfs.Read(func() ([]byte, error) { 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", s.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.Each("peer", func() ([]virtfs.FsNodeDecl, error) { var out []virtfs.FsNodeDecl for _, peerName := range a.Peers() { name := peerName out = append(out, virtfs.FsNodeDecl{ Name: name, Mode: 0222, Write: func(data []byte) error { target := s.FindAgent(name) if target == nil { return fmt.Errorf("peer %q not found", name) } input := strings.TrimSpace(string(data)) if input == "" { return nil } startAsync(s.Ctx, func() { target.Submit(s.Ctx, input) target.EnsureTrailingNewline() }) return nil }, }) } return out, nil }), virtfs.FileNode("ctl", 0666, virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) { return dispatch([]ctlCmd{ {"kill", "destroy this agent", func(_ []string) ([]byte, error) { s.RemoveAgent(a.ID()) session.PublishEvent("session."+s.ID+".agent."+a.ID()+".kill", "") return []byte("ok\n"), nil }}, {"stop", "interrupt the current turn", func(_ []string) ([]byte, error) { a.Interrupt(agent.ErrInterrupted) return []byte("ok\n"), nil }}, {"detach", "detach the foreground process to background", func(_ []string) ([]byte, error) { ts := a.ToolServer() if ts == nil { return nil, fmt.Errorf("no tool server") } if !ts.Detach() { return nil, fmt.Errorf("no foreground process to detach") } return []byte("ok\n"), nil }}, {"compact", "compact context now", func(_ []string) ([]byte, error) { if err := a.Compact(s.Ctx); err != nil { return nil, err } return []byte("ok\n"), nil }}, {"compactionmodel", "get/set the compaction model", func(args []string) ([]byte, error) { model := strings.TrimSpace(strings.Join(args, " ")) if model == "" { return []byte(a.CompactionModel() + "\n"), nil } a.SetCompactionModel(model) return []byte(model + "\n"), nil }}, {"clear", "clear chat history", func(_ []string) ([]byte, error) { if err := a.Clear(); err != nil { return nil, err } return []byte("ok\n"), nil }}, {"i", "inject text into the running turn", inject}, {"agent", "get/switch agent profile: agent [name]", func(args []string) ([]byte, error) { if len(args) == 0 { return []byte(a.Profile() + "\n"), nil } cfg, err := a.SwitchProfile(args[0]) if err != nil { return nil, err } // Reload tools for the new profile if ts := a.ToolServer(); ts != nil { ts.ClearTools() session.LoadTools(cfg, ts, s.ID, a.ID(), nil) } return []byte(args[0] + "\n"), nil }}, {"model", "get/set model: model [name]", func(args []string) ([]byte, error) { if len(args) == 0 { if be := a.Backend(); be != nil { return []byte(be.Model() + "\n"), nil } return nil, nil } name := strings.Join(args, " ") if be := a.Backend(); be != nil { be.SetModel(name) return []byte(modelPricingBlock(be, name)), nil } return []byte(name + "\n"), nil }}, {"models", "list available models", func(_ []string) ([]byte, error) { return []byte(s.CachedListModels() + "\n"), nil }}, {"tools", "list loaded 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", "load a tool: 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 } if infos, err := ts.ListTools(); err == nil { a.SetToolsPreamble(agent.RenderTools(infos)) } return []byte(args[0] + "\n"), nil }}, {"tools_all", "list all available tools", func(args []string) ([]byte, error) { ts := a.ToolServer() if ts == nil { return nil, fmt.Errorf("no tool server") } return ts.ListAllTools() }}, {"tool_unload", "unload a tool: tool_unload ", func(args []string) ([]byte, error) { if len(args) == 0 { return nil, fmt.Errorf("tool_unload requires a tool name") } ts := a.ToolServer() if ts == nil { return nil, fmt.Errorf("no tool server") } if err := ts.UnloadTool(args[0]); err != nil { return nil, err } if infos, err := ts.ListTools(); err == nil { a.SetToolsPreamble(agent.RenderTools(infos)) } return []byte(args[0] + "\n"), nil }}, {"backend", "get/set backend: backend [name]", func(args []string) ([]byte, error) { if len(args) == 0 { if be := a.Backend(); be != nil { return []byte(be.Name() + "\n"), nil } return nil, nil } name := strings.Join(args, " ") if err := a.SwitchBackend(name); err != nil { return nil, err } return []byte(name + "\n"), nil }}, {"name", "get/set agent name: name [value]", 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", "print the working directory", func(args []string) ([]byte, error) { return []byte(a.Cwd() + "\n"), nil }}, {"proc", "manage background procs: proc [top|term|kill|out|dismiss ]", func(args []string) ([]byte, error) { ts := a.ToolServer() if ts == nil { return nil, fmt.Errorf("no tool server") } if len(args) == 0 || args[0] == "top" { data, err := ts.ListProcs() if err != nil { return nil, err } if len(data) == 0 { return []byte("(no background processes)\n"), nil } return data, nil } subcmd := args[0] switch subcmd { case "term": if len(args) < 2 { return nil, fmt.Errorf("proc term requires pid") } pid, err := strconv.Atoi(args[1]) if err != nil { return nil, fmt.Errorf("invalid pid: %s", args[1]) } if err := ts.SignalDetached(pid, syscall.SIGTERM); err != nil { return nil, err } return []byte("ok\n"), nil case "kill": if len(args) < 2 { return nil, fmt.Errorf("proc kill requires pid") } pid, err := strconv.Atoi(args[1]) if err != nil { return nil, fmt.Errorf("invalid pid: %s", args[1]) } if err := ts.SignalDetached(pid, syscall.SIGKILL); err != nil { return nil, err } return []byte("ok\n"), nil case "out": if len(args) < 2 { return nil, fmt.Errorf("proc out requires pid") } pid, err := strconv.Atoi(args[1]) if err != nil { return nil, fmt.Errorf("invalid pid: %s", args[1]) } out, err := ts.GetDetachedOutput(pid) if err != nil { return nil, err } return []byte(out), nil case "dismiss": if len(args) < 2 { return nil, fmt.Errorf("proc dismiss requires pid") } pid, err := strconv.Atoi(args[1]) if err != nil { return nil, fmt.Errorf("invalid pid: %s", args[1]) } if !ts.DismissDetached(pid) { return nil, fmt.Errorf("process %d not found", pid) } return []byte("ok\n"), nil default: return nil, fmt.Errorf("unknown proc subcommand: %s (use: top, term , kill , out , dismiss )", subcmd) } }}, {"systemprompt", "print the rendered system prompt", func(_ []string) ([]byte, error) { return []byte(a.SystemPrompt() + "\n"), nil }}, {"peeradd", "link a peer agent: peeradd ", func(args []string) ([]byte, error) { if len(args) == 0 { return nil, fmt.Errorf("peeradd requires agent name") } target := s.FindAgent(args[0]) if target == nil { return nil, fmt.Errorf("agent %q not found in session", args[0]) } a.AddPeer(target.Name()) target.AddPeer(a.Name()) go s.Save() return []byte("ok\n"), nil }}, {"peerdel", "unlink a peer agent: peerdel ", func(args []string) ([]byte, error) { if len(args) == 0 { return nil, fmt.Errorf("peerdel requires agent name") } a.RemovePeer(args[0]) if target := s.FindAgent(args[0]); target != nil { target.RemovePeer(a.Name()) } go s.Save() return []byte("ok\n"), nil }}, {"peers", "list linked peer agents", func(_ []string) ([]byte, error) { peers := a.Peers() if len(peers) == 0 { return []byte("(no peers)\n"), nil } return []byte(strings.Join(peers, "\n") + "\n"), nil }}, {"stats", "print usage and cost statistics", stats}, }, data) }), ), virtfs.FileNode("stats", 0444, virtfs.Read(func() ([]byte, error) { a, err := metrics.AggregateAgent(s.ID, a.ID()) if err != nil { return nil, err } return []byte(metrics.Format(a)), nil }), ), virtfs.FileNode("metrics.by-backend-model", 0444, virtfs.Read(func() ([]byte, error) { groups, err := metrics.GroupedAggregate(s.ID, a.ID()) if err != nil { return nil, err } return []byte(metrics.FormatGroups(groups)), nil }), ), virtfs.FileNode("name", 0666, 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 []byte(a.ID() + "\n"), nil }), ), } }