// 9P filesystem server for ollie sessions. // See doc comment below for filesystem layout. package server import ( "context" "encoding/json" "fmt" "hash/fnv" "io" "net" "os" "os/user" "strings" "sync" "time" "ollie/backend" "ollie/tools" "ollie/elevate" olog "ollie/log" "ollie/paths" "olliesrv/fs" "olliesrv/session" "9fans.net/go/plan9" ) // Re-export fs types for use in server.go. type ( File = fs.File FileTree = fs.FileTree syntheticFileInfo = fs.SyntheticFileInfo ) const ( QTDir = plan9.QTDIR QTFile = plan9.QTFILE ) // fid tracks per-descriptor state for a single 9P connection. type fid struct { path string qid plan9.Qid mode uint8 writeBuf []byte waitBase string // for *wait files: value snapshotted at open time dirCache []byte // cached serialized dir entries for readdir completeResult []byte // for /complete: result of last write (per-fid state) generateResult []byte // for /generate: result of last write (per-fid state) routeResult []byte // for /route: result of last write (per-fid state) } // connState tracks all open fids for a single 9P connection. type connState struct { mu sync.RWMutex fids map[uint32]*fid ctx context.Context cancel context.CancelFunc pending map[uint16]context.CancelFunc // in-flight request cancels, keyed by tag uname string // user principal from Tattach } // Server is the 9P server for ollie sessions. type Server struct { mu sync.RWMutex conns []*connState log *olog.Logger sink *olog.Sink sessionMgr *session.Manager rootStore *fs.Tree elevateTree *elevateTree groups map[string]map[string]bool // group → set of members invalidateModels func() toolPrompt func(string) string toolRegistry *tools.Registry } // Config holds the pre-built trees and manager for the server. type Config struct { Sink *olog.Sink SessionMgr *session.Manager RootStore *fs.Tree ElevateBroker *elevate.Broker InvalidateModels func() ToolPrompt func(string) string // ToolRegistry is the shared tool registry for lazy tool promotion. ToolRegistry *tools.Registry } // Option configures the 9P server. type Option func(*Server) // New creates a new Server from a pre-built Config. func New(cfg Config) *Server { s := &Server{ log: cfg.Sink.Logger("9p", olog.LevelDebug), sink: cfg.Sink, sessionMgr: cfg.SessionMgr, rootStore: cfg.RootStore, groups: make(map[string]map[string]bool), invalidateModels: cfg.InvalidateModels, toolPrompt: cfg.ToolPrompt, toolRegistry: cfg.ToolRegistry, } if cfg.ElevateBroker != nil { s.elevateTree = newElevateTree(cfg.ElevateBroker) } return s } // fsRoute maps a path prefix to its backing fs and directory permissions. type fsRoute struct { prefix string dirMode os.FileMode tree func() FileTree } // routes returns the route table for the file tree namespace. // Directory permissions come from the fs.Perms registry. func (s *Server) routes() []fsRoute { routes := []fsRoute{ {fs.PathSessions, fs.Perms[fs.PathSessions].DirMode, func() FileTree { return s.sessionMgr.Tree() }}, } if s.elevateTree != nil { routes = append(routes, fsRoute{fs.PathElevate, fs.Perms[fs.PathElevate].DirMode, func() FileTree { return s.elevateTree }}) } return routes } // route returns the tree and entry name for a path, or nil if no route matches. func (s *Server) route(path string) (FileTree, string) { for _, r := range s.routes() { if strings.HasPrefix(path, r.prefix) { return r.tree(), strings.TrimPrefix(path, r.prefix) } } // Root-level files (e.g. /backends, /help). name := strings.TrimPrefix(path, "/") if !strings.Contains(name, "/") { if _, err := s.rootStore.Stat(name); err == nil { return s.rootStore, name } } return nil, "" } // openEntry opens a File for the given path. func (s *Server) openEntry(path string) (File, error) { // Intercept s/{id}/elevate — session policy via broker if s.elevateTree != nil && strings.HasPrefix(path, fs.PathSessions) { rel := strings.TrimPrefix(path, fs.PathSessions) if parts := strings.SplitN(rel, "/", 2); len(parts) == 2 && parts[1] == "elevate" { sessionID := parts[0] return &sessionElevateFile{broker: s.elevateTree.broker, sessionID: sessionID}, nil } } st, name := s.route(path) if st == nil { return nil, fmt.Errorf("%s: not found", path) } return st.Open(name) } // routeDir returns the tree for a directory path (e.g. "/a" → agentStore). func (s *Server) routeDir(path string) FileTree { for _, r := range s.routes() { if path+"/" == r.prefix { return r.tree() } } return nil } // dirMode returns the directory permission for a path. // Top-level directories get their mode from the route table. func (s *Server) dirMode(path string) plan9.Perm { for _, r := range s.routes() { if path+"/" == r.prefix { return plan9.Perm(r.dirMode) } if strings.HasPrefix(path, r.prefix) { return plan9.Perm(r.dirMode) } } return plan9.Perm(fs.Perms[fs.PathRoot].DirMode) } // filePerm returns the file permission for a path from the permission registry. func (s *Server) filePerm(path string) os.FileMode { // Session files have per-file overrides. if strings.HasPrefix(path, fs.PathSessions) { // Peer/proc file permissions. name := strings.TrimPrefix(path, fs.PathSessions) parts := strings.SplitN(name, "/", 3) if len(parts) == 3 && parts[1] == "peer" { return 0666 } if len(parts) == 3 && parts[1] == "proc" { return 0444 } base := pathBase(path) if m, ok := fs.Perms[fs.PathSessionFile].Files[base]; ok { return m } if m, ok := fs.Perms[fs.PathSessions].Files[base]; ok { return m } return fs.Perms[fs.PathSessions].FileMode } for _, r := range s.routes() { if strings.HasPrefix(path, r.prefix) { if p, ok := fs.Perms[r.prefix]; ok { return p.FileMode } } } // Root-level files: query the rootStore for actual permission name := strings.TrimPrefix(path, "/") if !strings.Contains(name, "/") { if info, err := s.rootStore.Stat(name); err == nil { return info.Mode() } } return fs.PermIdx } // AddGroup adds a user to a group. func (s *Server) AddGroup(group, user string) { s.mu.Lock() if s.groups[group] == nil { s.groups[group] = make(map[string]bool) } s.groups[group][user] = true s.mu.Unlock() } // RemoveGroup removes a user from a group. func (s *Server) RemoveGroup(group, user string) { s.mu.Lock() if m := s.groups[group]; m != nil { delete(m, user) if len(m) == 0 { delete(s.groups, group) } } s.mu.Unlock() } // InGroup returns true if user is a member of group. func (s *Server) InGroup(group, user string) bool { s.mu.RLock() defer s.mu.RUnlock() return s.groups[group][user] } // fileOwnerGroup returns the uid and gid for a given path. // Session namespace (/session/{sid}/**) is owned by the agent principal with group "agent". // Everything else is owned by the current system user and their primary group. func (s *Server) fileOwnerGroup(path string) (uid, gid string) { if strings.HasPrefix(path, fs.PathSessions) { parts := strings.SplitN(strings.TrimPrefix(path, fs.PathSessions), "/", 2) if len(parts) >= 1 && parts[0] != "new" && !isSessionFile(path) { if sess := s.sessionMgr.Session(parts[0]); sess != nil { return sess.Uname(), "agent" } } } if u, err := user.Current(); err == nil { return u.Username, "agent" } return "ollie", "agent" } // checkPerm verifies that uname has the requested access (mode) to path. // mode is the 9P open mode: OREAD=0, OWRITE=1, ORDWR=2, OEXEC=3. // Enforced on all paths. func (s *Server) checkPerm(uname, path string, mode uint8) error { uid, gid := s.fileOwnerGroup(path) dir := s.makeStat(path) perm := uint32(dir.Mode) & 0777 // Determine which permission bits apply. var bits uint32 if uname == uid { bits = (perm >> 6) & 7 } else if s.InGroup(gid, uname) { bits = (perm >> 3) & 7 } else { bits = perm & 7 } // Map 9P open mode to required permission bit. omode := mode & 3 switch omode { case plan9.OREAD: if bits&4 == 0 { return fmt.Errorf("permission denied") } case plan9.OWRITE: if bits&2 == 0 { return fmt.Errorf("permission denied") } case plan9.ORDWR: if bits&6 != 6 { return fmt.Errorf("permission denied") } case plan9.OEXEC: if bits&1 == 0 { return fmt.Errorf("permission denied") } } return nil } // readFile opens an entry in a tree and reads it. func readFile(s fs.FileTree, name string) ([]byte, error) { e, err := s.Open(name) if err != nil { return nil, err } return e.Read() } // writeFile opens an entry in a tree and writes to it. func writeFile(s fs.FileTree, name string, data []byte) error { e, err := s.Open(name) if err != nil { return err } return e.Write(data) } // readTimeout is the server-side deadline for all non-blocking reads. const readTimeout = 10 * time.Second // Serve handles a single 9P connection. Each request is dispatched to its own // goroutine so blocking reads (e.g. *wait files) do not stall the serve loop. func (s *Server) Serve(conn net.Conn) { defer conn.Close() connCtx, connCancel := context.WithCancel(context.Background()) cs := &connState{ fids: make(map[uint32]*fid), ctx: connCtx, cancel: connCancel, pending: make(map[uint16]context.CancelFunc), } s.mu.Lock() s.conns = append(s.conns, cs) s.mu.Unlock() defer func() { connCancel() s.mu.Lock() for i, c := range s.conns { if c == cs { s.conns = append(s.conns[:i], s.conns[i+1:]...) break } } s.mu.Unlock() }() responses := make(chan *plan9.Fcall, 16) var wg sync.WaitGroup // writer: serialises responses back onto the connection. go func() { for resp := range responses { plan9.WriteFcall(conn, resp) //nolint:errcheck } }() for { fc, err := plan9.ReadFcall(conn) if err != nil { if err != io.EOF { s.log.Error("read: %v", err) } break } reqCtx, reqCancel := context.WithCancel(connCtx) cs.mu.Lock() cs.pending[fc.Tag] = reqCancel cs.mu.Unlock() wg.Add(1) go func(fc *plan9.Fcall, ctx context.Context) { defer func() { reqCancel() cs.mu.Lock() delete(cs.pending, fc.Tag) cs.mu.Unlock() wg.Done() }() responses <- s.handle(cs, fc, ctx) }(fc, reqCtx) } wg.Wait() close(responses) } func (s *Server) handle(cs *connState, fc *plan9.Fcall, ctx context.Context) *plan9.Fcall { start := time.Now() defer func() { s.log.Debug(" -> %s %dµs", fcallTypeName(fc.Type), time.Since(start).Microseconds()) }() switch fc.Type { case plan9.Tversion: msize := fc.Msize if msize > 65536 { msize = 65536 } s.log.Debug("Tversion msize=%d", msize) return &plan9.Fcall{Type: plan9.Rversion, Tag: fc.Tag, Msize: msize, Version: "9P2000"} case plan9.Tauth: return errFcall(fc, "no auth required") case plan9.Tattach: s.log.Debug("Tattach fid=%d", fc.Fid) return s.attach(cs, fc) case plan9.Twalk: return s.walk(cs, fc) case plan9.Topen: return s.open(cs, fc) case plan9.Tcreate: return s.create(cs, fc) case plan9.Tread: return s.read(cs, fc, ctx) case plan9.Twrite: return s.write(cs, fc) case plan9.Tstat: return s.stat(cs, fc) case plan9.Twstat: return s.wstat(cs, fc) case plan9.Tflush: cs.mu.Lock() if cancel, ok := cs.pending[fc.Oldtag]; ok { cancel() } cs.mu.Unlock() return &plan9.Fcall{Type: plan9.Rflush, Tag: fc.Tag} case plan9.Tclunk: return s.clunk(cs, fc) case plan9.Tremove: return s.remove(cs, fc) default: return errFcall(fc, "unsupported operation") } } func fcallTypeName(t uint8) string { switch t { case plan9.Tversion: return "Tversion" case plan9.Tauth: return "Tauth" case plan9.Tattach: return "Tattach" case plan9.Twalk: return "Twalk" case plan9.Topen: return "Topen" case plan9.Tcreate: return "Tcreate" case plan9.Tread: return "Tread" case plan9.Twrite: return "Twrite" case plan9.Tstat: return "Tstat" case plan9.Twstat: return "Twstat" case plan9.Tflush: return "Tflush" case plan9.Tclunk: return "Tclunk" case plan9.Tremove: return "Tremove" default: return fmt.Sprintf("T%d", t) } } // isSessionFile reports whether path is a fixed file directly under /session/ // (i.e. /session/ where name is in sessionStoreFiles). func isSessionFile(path string) bool { name, ok := strings.CutPrefix(path, fs.PathSessions) if !ok || strings.Contains(name, "/") { return false } _, ok = session.FileMode(name) return ok } func errFcall(fc *plan9.Fcall, msg string) *plan9.Fcall { return &plan9.Fcall{Type: plan9.Rerror, Tag: fc.Tag, Ename: msg} } // qidPath returns a stable numeric path for use in Qid structs. func qidPath(path string) uint64 { if path == "/" { return 0 } h := fnv.New64a() h.Write([]byte(path)) return h.Sum64() } // pathType returns "dir", "file", or "" (not found) for a logical path. func (s *Server) pathType(path string) string { if path == "/" { return "dir" } // Check route table: directory match (e.g. "/a") or file match (e.g. "/a/foo"). if st := s.routeDir(path); st != nil { return "dir" } // Session paths: determine type from path structure without calling Stat // (which triggers expensive content reads). if strings.HasPrefix(path, fs.PathSessions) { name := strings.TrimPrefix(path, fs.PathSessions) if !strings.Contains(name, "/") { // /session/{name} — could be a session dir or a fixed file (new, idx, ls, etc.) if isSessionFile(path) { return "file" } return "dir" } // /session/{id}/peer or /session/{id}/proc — directory parts := strings.SplitN(name, "/", 3) if len(parts) == 2 && (parts[1] == "peer" || parts[1] == "proc") { return "dir" } // /session/{id}/peer/{peer-id} — verify peer exists if len(parts) == 3 && parts[1] == "peer" { sess := s.sessionMgr.Session(parts[0]) if sess == nil { return "" } if sess.HasPeer(parts[2]) { return "file" } return "" } // /session/{id}/proc/{pid} — verify PID exists in detached list if len(parts) == 3 && parts[1] == "proc" { sess := s.sessionMgr.Session(parts[0]) if sess == nil { return "" } for _, p := range sess.Core.ListDetached() { if fmt.Sprintf("%d", p.PID) == parts[2] { return "file" } } return "" } // /session/{id}/{file} — file return "file" } if st, name := s.route(path); st != nil { if info, err := st.Stat(name); err == nil { if info.IsDir() { return "dir" } return "file" } return "" } return "" } func pathParent(path string) string { if path == "/" { return "/" } i := strings.LastIndex(path, "/") if i == 0 { return "/" } return path[:i] } func pathJoin(parent, name string) string { if parent == "/" { return "/" + name } return parent + "/" + name } func pathBase(path string) string { i := strings.LastIndex(path, "/") if i < 0 { return path } return path[i+1:] } func boolToDir(isDir bool) uint32 { if isDir { return uint32(plan9.DMDIR) } return 0 } func (s *Server) attach(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() defer cs.mu.Unlock() cs.uname = fc.Uname qid := plan9.Qid{Type: QTDir, Path: 0} cs.fids[fc.Fid] = &fid{path: "/", qid: qid} s.log.Debug("Tattach uname=%q", fc.Uname) // All clients get agent group membership so they can access shared // directories. Owner bits handle user-vs-agent distinction. if fc.Uname != "" { s.AddGroup("agent", fc.Uname) } return &plan9.Fcall{Type: plan9.Rattach, Tag: fc.Tag, Qid: qid} } func (s *Server) walk(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() defer cs.mu.Unlock() f, ok := cs.fids[fc.Fid] if !ok { return errFcall(fc, "bad fid") } // Clone: copy the fid to newfid (even when wnames is empty). newf := &fid{path: f.path, qid: f.qid} if len(fc.Wname) == 0 { cs.fids[fc.Newfid] = newf return &plan9.Fcall{Type: plan9.Rwalk, Tag: fc.Tag, Wqid: []plan9.Qid{}} } s.log.Debug("Twalk fid=%d newfid=%d from=%q wnames=%v", fc.Fid, fc.Newfid, f.path, fc.Wname) wqids := make([]plan9.Qid, 0, len(fc.Wname)) cur := f.path for _, name := range fc.Wname { var next string if name == ".." { next = pathParent(cur) } else { next = pathJoin(cur, name) } t := s.pathType(next) if t == "" { if len(wqids) == 0 { return errFcall(fc, name+": file not found") } break } q := plan9.Qid{Path: qidPath(next)} if t == "dir" { q.Type = QTDir } wqids = append(wqids, q) cur = next } if len(wqids) == len(fc.Wname) { newf.path = cur newf.qid = wqids[len(wqids)-1] cs.fids[fc.Newfid] = newf } return &plan9.Fcall{Type: plan9.Rwalk, Tag: fc.Tag, Wqid: wqids} } func (s *Server) open(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() defer cs.mu.Unlock() f, ok := cs.fids[fc.Fid] if !ok { return errFcall(fc, "bad fid") } if err := s.checkPerm(cs.uname, f.path, fc.Mode); err != nil { return errFcall(fc, err.Error()) } f.mode = fc.Mode s.log.Debug("Topen fid=%d path=%q mode=%d", fc.Fid, f.path, fc.Mode) return &plan9.Fcall{Type: plan9.Ropen, Tag: fc.Tag, Qid: f.qid} } func (s *Server) create(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() defer cs.mu.Unlock() f, ok := cs.fids[fc.Fid] if !ok { return errFcall(fc, "bad fid") } newPath := pathJoin(f.path, fc.Name) if fc.Perm&plan9.DMDIR != 0 { // mkdir: delegate to tree if it supports MkdirAll. type dirCreator interface { MkdirAll(string) error } st, name := s.route(newPath) if st == nil { return errFcall(fc, "mkdir not supported") } if dc, ok := st.(dirCreator); ok { if err := dc.MkdirAll(name); err != nil { return errFcall(fc, err.Error()) } } else { return errFcall(fc, "mkdir not supported") } qid := plan9.Qid{Type: QTDir, Path: qidPath(newPath)} f.path = newPath f.qid = qid f.mode = fc.Mode return &plan9.Fcall{Type: plan9.Rcreate, Tag: fc.Tag, Qid: qid} } s.log.Debug("Tcreate parent=%q name=%q", f.path, fc.Name) // File create: route to the parent directory's fs. if st := s.routeDir(f.path); st != nil { if err := st.Create(fc.Name); err != nil { return errFcall(fc, err.Error()) } } else if st, parentName := s.route(f.path); st != nil { // Parent is a subdirectory within a tree; create relative to it. type subCreator interface { Create(string) error } if sc, ok := st.(subCreator); ok { if err := sc.Create(parentName + "/" + fc.Name); err != nil { return errFcall(fc, err.Error()) } } } qid := plan9.Qid{Path: qidPath(newPath)} f.path = newPath f.qid = qid f.mode = fc.Mode return &plan9.Fcall{Type: plan9.Rcreate, Tag: fc.Tag, Qid: qid} } func (s *Server) read(cs *connState, fc *plan9.Fcall, ctx context.Context) *plan9.Fcall { cs.mu.RLock() f, ok := cs.fids[fc.Fid] if !ok { cs.mu.RUnlock() return errFcall(fc, "bad fid") } path := f.path isDir := f.qid.Type&QTDir != 0 cs.mu.RUnlock() if isDir { s.log.Debug("Tread dir path=%q offset=%d count=%d", path, fc.Offset, fc.Count) // Use cached dir data if available; rebuild on offset=0 (new listing). if fc.Offset == 0 || f.dirCache == nil { f.dirCache = s.buildDirData(path) } data := s.sliceDirData(f.dirCache, fc.Offset, fc.Count) return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(data)), Data: data} } // /complete: return per-fid result from previous write (offset 0-based) if path == "/complete" { s.log.Debug("Tread path=%q offset=%d count=%d", path, fc.Offset, fc.Count) return s.readDNS(fc, &f.completeResult) } // /generate: return per-fid result from previous write (offset 0-based) if path == "/generate" { s.log.Debug("Tread path=%q offset=%d count=%d", path, fc.Offset, fc.Count) return s.readDNS(fc, &f.generateResult) } // /route: return per-fid result from previous write (offset 0-based) if path == "/route" { s.log.Debug("Tread path=%q offset=%d count=%d", path, fc.Offset, fc.Count) return s.readDNS(fc, &f.routeResult) } s.log.Debug("Tread path=%q offset=%d count=%d", path, fc.Offset, fc.Count) entry, err := s.openEntry(path) if err != nil { return errFcall(fc, err.Error()) } // OneShot entries yield data once per open; offset>0 means EOF. if entry.OneShot() && fc.Offset > 0 { return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0} } // Blocking entries use BlockingRead with timeout and waitBase tracking. if entry.IsBlocking() { cs.mu.RLock() f, fidOK := cs.fids[fc.Fid] var base string if fidOK { base = f.waitBase } cs.mu.RUnlock() waitCtx, waitCancel := context.WithTimeout(ctx, 5*time.Second) defer waitCancel() content, nextBase, err := entry.BlockingRead(waitCtx, base) if err != nil { return errFcall(fc, err.Error()) } if nextBase != "" { cs.mu.Lock() if f, ok := cs.fids[fc.Fid]; ok { f.waitBase = nextBase } cs.mu.Unlock() } return s.readSlice(fc, content) } // Normal read with timeout. type result struct { data []byte err error } ch := make(chan result, 1) go func() { data, err := entry.Read() ch <- result{data, err} }() timer := time.NewTimer(readTimeout) defer timer.Stop() select { case r := <-ch: if r.err != nil { return errFcall(fc, r.err.Error()) } return s.readSlice(fc, r.data) case <-ctx.Done(): return errFcall(fc, ctx.Err().Error()) case <-timer.C: return errFcall(fc, "read timeout") } } // readSlice serves a byte slice at the requested offset/count. func (s *Server) readSlice(fc *plan9.Fcall, content []byte) *plan9.Fcall { var data []byte off := int(fc.Offset) if off < len(content) { end := off + int(fc.Count) if end > len(content) { end = len(content) } data = content[off:end] } return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(data)), Data: data} } // readDNS implements the Plan 9 /net/dns read pattern: the result is always // read from offset 0 regardless of the client's file position (which advances // after a write). Returns the content once, then returns EOF on subsequent reads. func (s *Server) readDNS(fc *plan9.Fcall, content *[]byte) *plan9.Fcall { var data []byte if len(*content) > 0 { end := int(fc.Count) if end > len(*content) { end = len(*content) } data = (*content)[:end] *content = nil // consumed; next read returns EOF } return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(data)), Data: data} } func (s *Server) helpPath() string { return paths.CfgDir() + "/help.md" } func (s *Server) write(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() f, ok := cs.fids[fc.Fid] if !ok { cs.mu.Unlock() return errFcall(fc, "bad fid") } s.log.Debug("Twrite fid=%d path=%q offset=%d len=%d", fc.Fid, f.path, fc.Offset, len(fc.Data)) // /complete: synchronous request-response (per-fid state, like /net/dns) if f.path == "/complete" { cs.mu.Unlock() var req struct { CWD string `json:"cwd"` FilePath string `json:"file"` Prefix string `json:"prefix"` Suffix string `json:"suffix"` ExtraContext string `json:"context"` } if err := json.Unmarshal(fc.Data, &req); err != nil { // If not JSON, treat entire write as prefix (simple mode) req.Prefix = strings.TrimSpace(string(fc.Data)) } result, err := s.sessionMgr.Complete(req.CWD, req.FilePath, req.Prefix, req.Suffix, req.ExtraContext) if err != nil { return errFcall(fc, err.Error()) } f.completeResult = []byte(result) return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))} } // /generate: stateless one-shot generation (per-fid state, like /net/dns) if f.path == "/generate" { cs.mu.Unlock() var req backend.GenerateRequest if err := json.Unmarshal(fc.Data, &req); err != nil { req.Prompt = strings.TrimSpace(string(fc.Data)) } result, err := backend.Generate(context.Background(), req) if err != nil { return errFcall(fc, err.Error()) } f.generateResult = []byte(result) return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))} } // /route: classify task and select appropriate backend+model (per-fid state, like /net/dns) if f.path == "/route" { cs.mu.Unlock() var req backend.RouteRequest if err := json.Unmarshal(fc.Data, &req); err != nil { req.Task = strings.TrimSpace(string(fc.Data)) } result, err := backend.Route(context.Background(), req) if err != nil { return errFcall(fc, err.Error()) } f.routeResult = []byte(fmt.Sprintf("backend=%s model=%s", result.Backend, result.Model)) return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))} } // Accumulate; the 9P client may split large writes across multiple Twrite messages. end := int(fc.Offset) + len(fc.Data) if end > len(f.writeBuf) { grown := make([]byte, end) copy(grown, f.writeBuf) f.writeBuf = grown } copy(f.writeBuf[fc.Offset:], fc.Data) cs.mu.Unlock() return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))} } func (s *Server) stat(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.RLock() f, ok := cs.fids[fc.Fid] cs.mu.RUnlock() if !ok { return errFcall(fc, "bad fid") } dir := s.makeStat(f.path) s.log.Debug("Tstat path=%q mode=%o len=%d", f.path, dir.Mode, dir.Length) stat, err := dir.Bytes() if err != nil { return errFcall(fc, err.Error()) } return &plan9.Fcall{Type: plan9.Rstat, Tag: fc.Tag, Stat: stat} } func (s *Server) wstat(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() f, ok := cs.fids[fc.Fid] cs.mu.Unlock() if !ok { return errFcall(fc, "bad fid") } // Parse the new Dir from the stat bytes. newDir, err := plan9.UnmarshalDir(fc.Stat) if err != nil { // Some clients send minimal wstat (e.g. truncate); accept silently. return &plan9.Fcall{Type: plan9.Rwstat, Tag: fc.Tag} } oldName := pathBase(f.path) s.log.Debug("Twstat path=%q oldName=%q newName=%q", f.path, oldName, newDir.Name) if newDir.Name == "" || newDir.Name == oldName { return &plan9.Fcall{Type: plan9.Rwstat, Tag: fc.Tag} } st, relPath := s.route(f.path) if st == nil { return &plan9.Fcall{Type: plan9.Rwstat, Tag: fc.Tag} } if err := st.Rename(relPath, newDir.Name); err != nil { return errFcall(fc, err.Error()) } // Rewrite all fids that share the old path prefix. for _, r := range s.routes() { if strings.HasPrefix(f.path, r.prefix) { oldPath := f.path newPath := r.prefix + relPath[:len(relPath)-len(oldName)] + newDir.Name s.mu.RLock() conns := s.conns s.mu.RUnlock() for _, c := range conns { c.mu.Lock() for _, fid := range c.fids { if fid.path == oldPath || strings.HasPrefix(fid.path, oldPath+"/") { fid.path = newPath + fid.path[len(oldPath):] fid.qid.Path = qidPath(fid.path) } } c.mu.Unlock() } // Update group membership for session renames. if r.prefix == fs.PathSessions && !strings.Contains(relPath, "/") { s.RemoveGroup("agent", oldName) s.AddGroup("agent", newDir.Name) } break } } return &plan9.Fcall{Type: plan9.Rwstat, Tag: fc.Tag} } func (s *Server) clunk(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() f, ok := cs.fids[fc.Fid] var path string var data []byte var writable bool if ok { if m := f.mode & 3; m == plan9.OWRITE || m == plan9.ORDWR { path = f.path writable = true data = make([]byte, len(f.writeBuf)) copy(data, f.writeBuf) } delete(cs.fids, fc.Fid) } uname := cs.uname cs.mu.Unlock() if writable { s.log.Debug("Tclunk flush path=%q writeBuf=%d uname=%q", path, len(data), uname) input := strings.TrimSpace(string(data)) async := false if entry, err := s.openEntry(path); err == nil { async = entry.Async() } if async { go s.handleWrite(path, input, uname) //nolint:errcheck } else if err := s.handleWrite(path, input, uname); err != nil { s.log.Debug("Tclunk handleWrite err=%v", err) return errFcall(fc, err.Error()) } } else { s.log.Debug("Tclunk fid=%d path=%q (no write)", fc.Fid, path) } return &plan9.Fcall{Type: plan9.Rclunk, Tag: fc.Tag} } func (s *Server) remove(cs *connState, fc *plan9.Fcall) *plan9.Fcall { cs.mu.Lock() f, ok := cs.fids[fc.Fid] if ok { delete(cs.fids, fc.Fid) } cs.mu.Unlock() if !ok { return errFcall(fc, "bad fid") } path := f.path s.log.Debug("Tremove path=%q", path) st, name := s.route(path) if st == nil { return errFcall(fc, "remove not supported") } if err := st.Delete(name); err != nil { return errFcall(fc, err.Error()) } return &plan9.Fcall{Type: plan9.Rremove, Tag: fc.Tag} } // handleWrite processes a fully-assembled write payload for the given path. // Called synchronously from clunk; prompt writes are the exception (spawned // as a goroutine because they block for the entire agent turn). func (s *Server) handleWrite(path, input, uname string) error { s.log.Debug("handleWrite path=%q input_len=%d uname=%q", path, len(input), uname) // Handle /tools/load - promote a tool for the current session st, name := s.route(path) if st == nil { return nil } return writeFile(st, name, []byte(input)) } // Shutdown kills all active sessions and batch jobs. func (s *Server) Shutdown() { s.sessionMgr.Shutdown() } // InterruptAll cancels any in-progress agent turn on every active session. func (s *Server) InterruptAll() { s.sessionMgr.InterruptAll() } // buildDirData serializes all directory entries for the given path into a byte slice. func (s *Server) buildDirData(path string) []byte { var dirs []plan9.Dir if path == "/" { rootEntries := []string{"agents", "backends", "complete", "ctl", "generate", "help", "models", "route", "session"} for _, name := range rootEntries { fpath := "/" + name st := s.makeStat(fpath) dirs = append(dirs, st) } } else if st := s.routeDir(path); st != nil { // Top-level tree directory (e.g. /a, /s, /m). entries, _ := st.List() for _, e := range entries { fpath := path + "/" + e.Name() d := s.makeStat(fpath) if info, err := e.Info(); err == nil { d.Atime = uint32(info.ModTime().Unix()) d.Mtime = uint32(info.ModTime().Unix()) } dirs = append(dirs, d) } } else if st, name := s.route(path); st != nil { // Subdirectory within a tree (e.g. /session/mysession, /session/mysession/t). type dirLister interface { Readdir(string) ([]os.DirEntry, error) } if dl, ok := st.(dirLister); ok { entries, _ := dl.Readdir(name) for _, e := range entries { fpath := path + "/" + e.Name() d := s.makeStat(fpath) dirs = append(dirs, d) } } } // Serialize all entries to a byte slice. var allData []byte for _, d := range dirs { b, err := d.Bytes() if err != nil { continue } allData = append(allData, b...) } return allData } // sliceDirData extracts complete directory entries from allData at the given offset, // up to count bytes. func (s *Server) sliceDirData(allData []byte, offset uint64, count uint32) []byte { if offset >= uint64(len(allData)) { return nil } remaining := allData[offset:] var result []byte for len(remaining) >= 2 { entrySize := int(remaining[0]) | int(remaining[1])<<8 total := entrySize + 2 if total > len(remaining) { break } if uint32(len(result)+total) > count { break } result = append(result, remaining[:total]...) remaining = remaining[total:] } return result } // makeStat builds a plan9.Dir for a logical path. func (s *Server) makeStat(path string) plan9.Dir { base := pathBase(path) if path == "/" { base = "." } t := s.pathType(path) isDir := t == "dir" qid := plan9.Qid{Path: qidPath(path)} var mode plan9.Perm if isDir { qid.Type = QTDir mode = plan9.DMDIR | s.dirMode(path) } else { // Use the permission registry for file mode rather than os.Stat // (which is affected by umask and may not reflect intended perms). mode = plan9.Perm(s.filePerm(path)) } uid, gid := s.fileOwnerGroup(path) dir := plan9.Dir{ Qid: qid, Mode: mode, Name: base, Uid: uid, Gid: gid, Muid: uid, } // For chat and tailable mutable files, report actual size and // Qid version so polling tools (tail -f) can detect changes via stat. // Path format: /session/{sessid}/{file} if strings.HasPrefix(path, fs.PathSessions) { parts := strings.SplitN(strings.TrimPrefix(path, "/"), "/", 3) if len(parts) == 3 && parts[0] == "session" { if sess := s.sessionMgr.Session(parts[1]); sess != nil { if base == "chat" { length, vers := sess.LogInfo() dir.Length = uint64(length) dir.Qid.Vers = vers } } } } // For all other readable files, compute content length so clients // that check stat before reading (cat, 9pfuse, etc.) see non-zero size. if dir.Length == 0 && !isDir { switch { case isSessionFile(path): if content, err := readFile(s.sessionMgr.Tree(), base); err == nil { dir.Length = uint64(len(content)) } case path == "/backends": dir.Length = uint64(len(strings.Join(backend.Backends(), "\n") + "\n")) case path == "/models": if content, err := readFile(s.rootStore, "models"); err == nil { dir.Length = uint64(len(content)) } case path == "/help": if info, err := os.Stat(s.helpPath()); err == nil { dir.Length = uint64(info.Size()) } case strings.HasPrefix(path, fs.PathSessions): // HACK: 9pfuse trusts stat Length — if we report 0, the kernel // never issues a read. Compute size by reading the file content. if entry, err := s.openEntry(path); err == nil { if data, err := entry.Read(); err == nil { dir.Length = uint64(len(data)) } } } } return dir }