diff --git a/cmd/Ollie/main.go b/cmd/Ollie/main.go index eff7536..8e4d112 100644 --- a/cmd/Ollie/main.go +++ b/cmd/Ollie/main.go @@ -1,24 +1,22 @@ -// ollie-acme: acme interface for ollie AI sessions +// ollie-acme: acme interface for ollie AI sessions. +// All file operations use 9P directly (not FUSE mount) for proper +// blocking read support on streaming files like chat. package main import ( + "io" "log" "os" "path/filepath" "sort" "strings" "sync" - "time" "9fans.net/go/acme" + "9fans.net/go/plan9" + "9fans.net/go/plan9/client" ) -var ollie string - -func init() { - ollie = filepath.Join(os.Getenv("HOME"), "mnt", "ollie") -} - func main() { acme.AutoExit(true) if err := newSessionList(); err != nil { @@ -27,20 +25,90 @@ func main() { select {} } -// agentBase returns the filesystem path to the active agent within a session. -// Layout: s/{sid}/agent/{aid}/ -func agentBase(sid string) string { - agentDir := filepath.Join(ollie, "session", sid, "agent") - entries, err := os.ReadDir(agentDir) - if err != nil { - return filepath.Join(agentDir, "0") // fallback +// --- 9P helpers --- + +func dial9p() *client.Fsys { + ns := os.Getenv("NAMESPACE") + if ns == "" { + ns = client.Namespace() } - for _, e := range entries { - if e.IsDir() { - return filepath.Join(agentDir, e.Name()) + if ns == "" { + log.Fatal("cannot determine namespace; set $NAMESPACE") + } + address := ns + "/ollie" + + conn, err := client.Dial("unix", address) + if err != nil { + log.Fatalf("dial %s: %v", address, err) + } + fsys, err := conn.Attach(nil, getuser(), "") + if err != nil { + log.Fatalf("attach: %v", err) + } + return fsys +} + +func getuser() string { + if u := os.Getenv("USER"); u != "" { + return u + } + return "none" +} + +func read9p(path string) ([]byte, error) { + fsys := dial9p() + fid, err := fsys.Open(path, plan9.OREAD) + if err != nil { + return nil, err + } + defer fid.Close() + var buf []byte + tmp := make([]byte, 8192) + for { + n, err := fid.Read(tmp) + if n > 0 { + buf = append(buf, tmp[:n]...) + } + if err != nil { + if err == io.EOF { + break + } + return buf, err } } - return filepath.Join(agentDir, "0") + return buf, nil +} + +func write9p(path string, data []byte) error { + fsys := dial9p() + fid, err := fsys.Open(path, plan9.OWRITE) + if err != nil { + return err + } + defer fid.Close() + _, err = fid.Write(data) + return err +} + +// agentPath returns the 9P path to the active agent within a session. +func agentPath(sid string) string { + fsys := dial9p() + dirPath := "session/" + sid + "/agent" + fid, err := fsys.Open(dirPath, plan9.OREAD) + if err != nil { + return dirPath + "/0" + } + dirs, err := fid.Dirreadall() + fid.Close() + if err != nil || len(dirs) == 0 { + return dirPath + "/0" + } + for _, d := range dirs { + if d.Mode&plan9.DMDIR != 0 { + return dirPath + "/" + d.Name + } + } + return dirPath + "/" + dirs[0].Name } // --- session list window --- @@ -66,7 +134,7 @@ func newSessionList() error { } func (sl *sessionList) refresh() { - data, err := os.ReadFile(filepath.Join(ollie, "session", "idx")) + data, err := read9p("session/idx") if err != nil { sl.win.Errf("read idx: %v", err) return @@ -119,15 +187,15 @@ func (sl *sessionList) eventLoop() { case 'x', 'X': cmd := strings.TrimSpace(string(e.Text)) if cmd == "New" { - openFile(filepath.Join(ollie, "session", "new")) + openNewSession() continue } if cmd == "Kill" { arg := strings.TrimSpace(string(e.Arg)) sid := sl.resolveFullSID(arg) if sid != "" { - ab := agentBase(sid) - os.WriteFile(filepath.Join(ab, "ctl"), []byte("kill"), 0644) //nolint:errcheck + ab := agentPath(sid) + write9p(ab+"/ctl", []byte("kill")) sl.refresh() } continue @@ -137,7 +205,7 @@ func (sl *sessionList) eventLoop() { continue } sid := sl.resolveFullSID(cmd) - if sid != "" && sessionExists(sid) { + if sid != "" { openChat(sid) } else { sl.win.WriteEvent(e) @@ -149,18 +217,12 @@ func (sl *sessionList) eventLoop() { os.Exit(0) } -func sessionExists(sid string) bool { - _, err := os.Stat(filepath.Join(ollie, "session", sid)) - return err == nil -} - // --- chat window --- type chatWin struct { - win *acme.Win - sid string - offset int - mu sync.Mutex + win *acme.Win + sid string + mu sync.Mutex } var chatWindows sync.Map @@ -177,7 +239,7 @@ func openChat(sid string) { return } winName := "ollie/sessions/" + sid - if data, err := os.ReadFile(filepath.Join(ollie, "session", "idx")); err == nil { + if data, err := read9p("session/idx"); err == nil { for _, line := range strings.Split(string(data), "\n") { fields := strings.Split(line, "\t") if len(fields) >= 3 && fields[0] == sid { @@ -193,39 +255,38 @@ func openChat(sid string) { cw := &chatWin{win: w, sid: sid} chatWindows.Store(sid, cw) - // Load existing chat. - ab := agentBase(sid) - chatPath := filepath.Join(ab, "chat") - if data, err := os.ReadFile(chatPath); err == nil && len(data) > 0 { + // Load existing log (last 64KB) for initial context + ab := agentPath(sid) + if data, err := read9p(ab + "/log"); err == nil && len(data) > 0 { w.Write("body", data) - cw.offset = len(data) } w.Ctl("clean") - go cw.tail() - go cw.eventLoop() + go cw.streamChat(ab) + go cw.eventLoop(ab) } -func (cw *chatWin) eventLoop() { +func (cw *chatWin) eventLoop(ab string) { defer chatWindows.Delete(cw.sid) - ab := agentBase(cw.sid) for e := range cw.win.EventChan() { switch e.C2 { case 'x', 'X': cmd := strings.TrimSpace(string(e.Text)) switch cmd { case "Prompt": - openFile(filepath.Join(ab, "prompt")) + openPrompt(ab) case "Stop": - os.WriteFile(filepath.Join(ab, "ctl"), []byte("stop"), 0644) //nolint:errcheck + write9p(ab+"/ctl", []byte("stop")) case "Ctl": - openFile(filepath.Join(ab, "ctl")) + openFile9p(ab + "/ctl") case "Cfg": - openFile(filepath.Join(ab, "cfg")) + openFile9p(ab + "/cfg") case "Nudge": - os.WriteFile(filepath.Join(ab, "prompt"), []byte("."), 0644) //nolint:errcheck + write9p(ab+"/prompt", []byte(".")) case "Usage": - openFile(filepath.Join(ab, "usage")) + if data, err := read9p(ab + "/usage"); err == nil { + cw.win.Errf("%s", string(data)) + } default: cw.win.WriteEvent(e) } @@ -235,51 +296,117 @@ func (cw *chatWin) eventLoop() { } } -// tail polls the chat file for new content and streams it to the window. -func (cw *chatWin) tail() { - ab := agentBase(cw.sid) - chatPath := filepath.Join(ab, "chat") +// streamChat reads the chat file via 9P (blocking reads) and appends to the window. +// The read blocks until new tokens arrive, then returns them. Loops until EOF +// (session killed) or error. +func (cw *chatWin) streamChat(ab string) { + fsys := dial9p() + fid, err := fsys.Open(ab+"/chat", plan9.OREAD) + if err != nil { + log.Printf("open chat stream: %v", err) + return + } + defer fid.Close() + buf := make([]byte, 8192) for { - fi, err := os.Stat(chatPath) - if err != nil { - if os.IsNotExist(err) { - return - } - time.Sleep(50 * time.Millisecond) - continue + n, err := fid.Read(buf) + if n > 0 { + cw.mu.Lock() + cw.win.Addr("$") + cw.win.Write("data", buf[:n]) + cw.win.Ctl("clean") + cw.mu.Unlock() } - size := int(fi.Size()) - cw.mu.Lock() - off := cw.offset - cw.mu.Unlock() - if size > off { - f, err := os.Open(chatPath) - if err == nil { - buf := make([]byte, size-off) - n, _ := f.ReadAt(buf, int64(off)) - f.Close() - if n > 0 { - cw.mu.Lock() - cw.win.Addr("$") - cw.win.Write("data", buf[:n]) - cw.offset += n - cw.win.Ctl("clean") - cw.mu.Unlock() + if err != nil { + // EOF = session killed; any other error = done + return + } + } +} + +// --- helper windows --- + +func openNewSession() { + w, err := acme.New() + if err != nil { + return + } + w.Name("ollie/new") + w.Write("body", []byte("cwd="+os.Getenv("HOME")+"\n")) + w.Ctl("cleartag") + w.Fprintf("tag", " Put") + go func() { + for e := range w.EventChan() { + if e.C2 == 'x' || e.C2 == 'X' { + cmd := strings.TrimSpace(string(e.Text)) + if cmd == "Put" { + w.Addr(",") + data, _ := w.ReadAll("body") + write9p("session/new", data) + w.Del(true) + return } } } - time.Sleep(50 * time.Millisecond) - } + }() } -// openFile opens a 9P/filesystem path as a plain acme editor window. -func openFile(path string) { +func openPrompt(ab string) { w, err := acme.New() if err != nil { - log.Printf("open %s: %v", path, err) return } - w.Name("%s", path) - w.Ctl("get") + w.Name("%s", filepath.Base(ab)+"/prompt") + w.Ctl("cleartag") + w.Fprintf("tag", " Put") + go func() { + for e := range w.EventChan() { + if e.C2 == 'x' || e.C2 == 'X' { + cmd := strings.TrimSpace(string(e.Text)) + if cmd == "Put" { + w.Addr(",") + data, _ := w.ReadAll("body") + write9p(ab+"/prompt", data) + w.Addr(",") + w.Write("data", nil) + w.Ctl("clean") + } + } + } + }() +} + +func openFile9p(path string) { + w, err := acme.New() + if err != nil { + return + } + parts := strings.Split(path, "/") + w.Name("%s", parts[len(parts)-1]) + if data, err := read9p(path); err == nil { + w.Write("body", data) + } + w.Ctl("cleartag") + w.Fprintf("tag", " Put Get") + go func() { + for e := range w.EventChan() { + if e.C2 == 'x' || e.C2 == 'X' { + cmd := strings.TrimSpace(string(e.Text)) + switch cmd { + case "Put": + w.Addr(",") + data, _ := w.ReadAll("body") + write9p(path, data) + w.Ctl("clean") + case "Get": + if data, err := read9p(path); err == nil { + w.Addr(",") + w.Write("data", data) + w.Ctl("clean") + } + } + } + } + }() } diff --git a/cmd/olliesrv/elevate_tree.go b/cmd/olliesrv/elevate_tree.go index 956e0b4..81b4c86 100644 --- a/cmd/olliesrv/elevate_tree.go +++ b/cmd/olliesrv/elevate_tree.go @@ -91,6 +91,7 @@ func (f *elevatePolicyFile) BlockingRead(_ context.Context, _ string) ([]byte, s } func (f *elevatePolicyFile) OneShot() bool { return false } func (f *elevatePolicyFile) IsBlocking() bool { return false } +func (f *elevatePolicyFile) LongBlock() bool { return false } func (f *elevatePolicyFile) Async() bool { return false } // --- pending dir --- @@ -121,6 +122,7 @@ func (f *elevatePendingDir) BlockingRead(_ context.Context, _ string) ([]byte, s } func (f *elevatePendingDir) OneShot() bool { return false } func (f *elevatePendingDir) IsBlocking() bool { return false } +func (f *elevatePendingDir) LongBlock() bool { return false } func (f *elevatePendingDir) Async() bool { return false } // --- pending/ file --- @@ -162,6 +164,7 @@ func (f *elevatePendingFile) BlockingRead(_ context.Context, _ string) ([]byte, } func (f *elevatePendingFile) OneShot() bool { return false } func (f *elevatePendingFile) IsBlocking() bool { return false } +func (f *elevatePendingFile) LongBlock() bool { return false } func (f *elevatePendingFile) Async() bool { return false } // --- s/{id}/elevate file (session policy) --- @@ -194,4 +197,5 @@ func (f *sessionElevateFile) BlockingRead(_ context.Context, _ string) ([]byte, } func (f *sessionElevateFile) OneShot() bool { return false } func (f *sessionElevateFile) IsBlocking() bool { return false } +func (f *sessionElevateFile) LongBlock() bool { return false } func (f *sessionElevateFile) Async() bool { return false } diff --git a/cmd/olliesrv/server.go b/cmd/olliesrv/server.go index 3ed2355..834c142 100644 --- a/cmd/olliesrv/server.go +++ b/cmd/olliesrv/server.go @@ -832,7 +832,14 @@ func (s *Server) read(cs *connState, fc *plan9.Fcall, ctx context.Context) *plan base = f.waitBase } cs.mu.RUnlock() - waitCtx, waitCancel := context.WithTimeout(ctx, 5*time.Second) + var waitCtx context.Context + var waitCancel context.CancelFunc + if entry.LongBlock() { + // Streaming files block until data arrives or connection closes. + waitCtx, waitCancel = context.WithCancel(ctx) + } else { + waitCtx, waitCancel = context.WithTimeout(ctx, 5*time.Second) + } defer waitCancel() content, nextBase, err := entry.BlockingRead(waitCtx, base) if err != nil { @@ -845,6 +852,15 @@ func (s *Server) read(cs *connState, fc *plan9.Fcall, ctx context.Context) *plan } cs.mu.Unlock() } + if entry.LongBlock() { + // Streaming files: return content directly, ignoring fc.Offset. + // The per-fid waitBase tracks position; the Tread offset is irrelevant. + count := int(fc.Count) + if count > len(content) { + count = len(content) + } + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(count), Data: content[:count]} + } return s.readSlice(fc, content) } @@ -1248,8 +1264,11 @@ func (s *Server) makeStat(path string) plan9.Dir { parts := strings.SplitN(strings.TrimPrefix(path, "/"), "/", 3) if len(parts) == 3 && parts[0] == "session" { if sess := fs.Lookup(s.sessionTree, parts[1]); sess != nil { - if base == "chat" { + if base == "log" { length, vers := sess.LogInfo() + if length > 64*1024 { + length = 64 * 1024 + } dir.Length = uint64(length) dir.Qid.Vers = vers } diff --git a/experiments/9p-stream/go.mod b/experiments/9p-stream/go.mod new file mode 100644 index 0000000..55e4816 --- /dev/null +++ b/experiments/9p-stream/go.mod @@ -0,0 +1,5 @@ +module ollie/experiments/9p-stream + +go 1.25 + +require 9fans.net/go v0.0.7 diff --git a/experiments/9p-stream/go.sum b/experiments/9p-stream/go.sum new file mode 100644 index 0000000..d81a876 --- /dev/null +++ b/experiments/9p-stream/go.sum @@ -0,0 +1,34 @@ +9fans.net/go v0.0.7 h1:H5CsYJTf99C8EYAQr+uSoEJnLP/iZU8RmDuhyk30iSM= +9fans.net/go v0.0.7/go.mod h1:Rxvbbc1e+1TyGMjAvLthGTyO97t+6JMQ6ly+Lcs9Uf0= +dmitri.shuralyov.com/gpu/mtl v0.0.0-20201218220906-28db891af037/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= +github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= +github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56/go.mod h1:JhuoJpWY28nO4Vef9tZUw9qufEGTyX1+7lmHxV5q5G4= +golang.org/x/exp v0.0.0-20210405174845-4513512abef3/go.mod h1:I6l2HNBLBZEcrOoCpyKLdY2lHoRZ8lI4x60KMCQDft4= +golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= +golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= +golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE= +golang.org/x/mobile v0.0.0-20201217150744-e6ae53a27f4f/go.mod h1:skQtrUTUwhdJvXM/2KKJzY8pDgNr9I/FOMqDVRPBUS4= +golang.org/x/mobile v0.0.0-20210220033013-bdb1ca9a1e08/go.mod h1:skQtrUTUwhdJvXM/2KKJzY8pDgNr9I/FOMqDVRPBUS4= +golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY= +golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg= +golang.org/x/mod v0.1.1-0.20191209134235-331c550502dd/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.1-0.20200828183125-ce943fd02449/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210415045647-66c3f260301c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200117012304-6edc0a871e69/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= +golang.org/x/tools v0.0.0-20200207183749-b753a1ba74fa/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/experiments/9p-stream/main.go b/experiments/9p-stream/main.go new file mode 100644 index 0000000..e7aa3d8 --- /dev/null +++ b/experiments/9p-stream/main.go @@ -0,0 +1,448 @@ +// 9P streaming chat prototype. +// +// Demonstrates native 9P streaming: reads on `stream` block until +// new tokens arrive from the simulated LLM, then return them. +// No polling needed — the protocol handles the wait naturally. +// +// Files: +// ctl - write "prompt " to start generation +// stream - blocking read: delivers tokens as they arrive, EOF when done +// chat - full conversation log (non-blocking) +// +// Usage: +// go run . +// echo "prompt hello" | 9p -a localhost:5640 write ctl +// 9p -a localhost:5640 read stream +// 9p -a localhost:5640 read chat +package main + +import ( + "context" + "fmt" + "log" + "math/rand" + "net" + "os" + "strings" + "sync" + "time" + + "9fans.net/go/plan9" +) + +const listenAddr = "localhost:5640" + +// Simulated response — what the "LLM" produces +var cannedResponse = `I'll explain how 9P streaming works. + +In Plan 9, the file server controls when it responds to read requests. +This gives us natural streaming semantics without polling: + +1. Client opens the stream file +2. Client sends Tread +3. Server holds the request until tokens arrive +4. Server responds with new tokens +5. Repeat until response is complete (EOF) + +The transport handles the wait — no timers, no busy loops. + +` + "```go" + ` +// The key: server delays Rread until data exists +func handleRead(req) { + wait until tokens available + respond with tokens +} +` + "```" + ` + +This is how /dev/cons, /dev/mouse, and other Plan 9 devices work. +Streaming is not a feature — it's the natural mode of operation. +` + +func main() { + srv := &chatServer{} + srv.streamCond = sync.NewCond(&srv.mu) + + ln, err := net.Listen("tcp", listenAddr) + if err != nil { + log.Fatal(err) + } + log.Printf("9P stream prototype on %s", listenAddr) + log.Printf(" echo 'prompt hello' | 9p -a %s write ctl", listenAddr) + log.Printf(" 9p -a %s read stream", listenAddr) + + for { + conn, err := ln.Accept() + if err != nil { + log.Printf("accept: %v", err) + continue + } + go srv.serve(conn) + } +} + +type chatServer struct { + mu sync.Mutex + streamCond *sync.Cond + + // Current streaming response + streamBuf []byte // tokens produced so far + streamDone bool // generation complete + + // Full conversation log + chat []byte +} + +type fidState struct { + path string + offset int64 +} + +func (s *chatServer) serve(conn net.Conn) { + defer conn.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var fidsMu sync.Mutex + fids := make(map[uint32]*fidState) + var msize uint32 = 8192 + + responses := make(chan *plan9.Fcall, 16) + + // Writer + go func() { + for resp := range responses { + if err := plan9.WriteFcall(conn, resp); err != nil { + cancel() + return + } + } + }() + + var wg sync.WaitGroup + for { + fc, err := plan9.ReadFcall(conn) + if err != nil { + break + } + + wg.Add(1) + go func(fc *plan9.Fcall) { + defer wg.Done() + resp := s.handle(ctx, fc, fids, &fidsMu, &msize) + select { + case responses <- resp: + case <-ctx.Done(): + } + }(fc) + } + + cancel() + // Wake any blocked stream readers + s.streamCond.Broadcast() + wg.Wait() + close(responses) +} + +func (s *chatServer) handle(ctx context.Context, fc *plan9.Fcall, fids map[uint32]*fidState, fidsMu *sync.Mutex, msize *uint32) *plan9.Fcall { + switch fc.Type { + case plan9.Tversion: + if fc.Msize < *msize { + *msize = fc.Msize + } + return &plan9.Fcall{Type: plan9.Rversion, Tag: fc.Tag, Msize: *msize, Version: "9P2000"} + + case plan9.Tauth: + return &plan9.Fcall{Type: plan9.Rerror, Tag: fc.Tag, Ename: "no auth required"} + + case plan9.Tattach: + fidsMu.Lock() + fids[fc.Fid] = &fidState{path: "/"} + fidsMu.Unlock() + return &plan9.Fcall{Type: plan9.Rattach, Tag: fc.Tag, Qid: dirQid("/")} + + case plan9.Twalk: + fidsMu.Lock() + src, ok := fids[fc.Fid] + fidsMu.Unlock() + if !ok { + return errResp(fc, "bad fid") + } + if len(fc.Wname) == 0 { + fidsMu.Lock() + fids[fc.Newfid] = &fidState{path: src.path} + fidsMu.Unlock() + return &plan9.Fcall{Type: plan9.Rwalk, Tag: fc.Tag} + } + path := src.path + var qids []plan9.Qid + for _, name := range fc.Wname { + switch name { + case "ctl", "stream", "chat": + path = "/" + name + qids = append(qids, fileQid(name)) + default: + return errResp(fc, "not found: "+name) + } + } + fidsMu.Lock() + fids[fc.Newfid] = &fidState{path: path} + fidsMu.Unlock() + return &plan9.Fcall{Type: plan9.Rwalk, Tag: fc.Tag, Wqid: qids} + + case plan9.Topen: + fidsMu.Lock() + f, ok := fids[fc.Fid] + fidsMu.Unlock() + if !ok { + return errResp(fc, "bad fid") + } + f.offset = 0 + if f.path == "/stream" { + s.mu.Lock() + f.offset = int64(len(s.streamBuf)) + s.mu.Unlock() + } + return &plan9.Fcall{Type: plan9.Ropen, Tag: fc.Tag, Qid: fileQid(strings.TrimPrefix(f.path, "/"))} + + case plan9.Tread: + fidsMu.Lock() + f, ok := fids[fc.Fid] + fidsMu.Unlock() + if !ok { + return errResp(fc, "bad fid") + } + count := int(fc.Count) + if uint32(count) > *msize-plan9.IOHDRSZ { + count = int(*msize - plan9.IOHDRSZ) + } + + switch f.path { + case "/": + return s.readDir(fc, f, count) + case "/chat": + return s.readChat(fc, f, count) + case "/stream": + return s.readStream(ctx, fc, f, count) + default: + return errResp(fc, "cannot read "+f.path) + } + + case plan9.Twrite: + fidsMu.Lock() + f, ok := fids[fc.Fid] + fidsMu.Unlock() + if !ok { + return errResp(fc, "bad fid") + } + if f.path == "/ctl" { + cmd := strings.TrimSpace(string(fc.Data)) + if strings.HasPrefix(cmd, "prompt ") { + prompt := strings.TrimPrefix(cmd, "prompt ") + go s.generateResponse(prompt) + return &plan9.Fcall{Type: plan9.Rwrite, Tag: fc.Tag, Count: uint32(len(fc.Data))} + } + return errResp(fc, "unknown command: "+cmd) + } + return errResp(fc, "cannot write "+f.path) + + case plan9.Tstat: + fidsMu.Lock() + f, ok := fids[fc.Fid] + fidsMu.Unlock() + if !ok { + return errResp(fc, "bad fid") + } + dir := s.stat(f.path) + b, _ := dir.Bytes() + return &plan9.Fcall{Type: plan9.Rstat, Tag: fc.Tag, Stat: b} + + case plan9.Tclunk: + fidsMu.Lock() + delete(fids, fc.Fid) + fidsMu.Unlock() + return &plan9.Fcall{Type: plan9.Rclunk, Tag: fc.Tag} + + default: + return errResp(fc, fmt.Sprintf("unhandled %d", fc.Type)) + } +} + +// readStream blocks until new data is available or response is done. +func (s *chatServer) readStream(ctx context.Context, fc *plan9.Fcall, f *fidState, count int) *plan9.Fcall { + s.mu.Lock() + // Wait until there's data beyond our read position, or generation is done + for int64(len(s.streamBuf)) <= f.offset && !s.streamDone { + // Release lock and wait for signal + s.streamCond.Wait() + // Check if context cancelled (connection closed) + if ctx.Err() != nil { + s.mu.Unlock() + return errResp(fc, "interrupted") + } + } + + // EOF: no more data and generation is done + if f.offset >= int64(len(s.streamBuf)) { + s.mu.Unlock() + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0} + } + + // Return available data + end := int64(len(s.streamBuf)) + if end-f.offset > int64(count) { + end = f.offset + int64(count) + } + data := make([]byte, end-f.offset) + copy(data, s.streamBuf[f.offset:end]) + f.offset = end + s.mu.Unlock() + + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(data)), Data: data} +} + +// readChat returns the full log (non-blocking). +func (s *chatServer) readChat(fc *plan9.Fcall, f *fidState, count int) *plan9.Fcall { + s.mu.Lock() + data := s.chat + s.mu.Unlock() + + offset := int64(fc.Offset) + if offset >= int64(len(data)) { + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0} + } + end := offset + int64(count) + if end > int64(len(data)) { + end = int64(len(data)) + } + chunk := make([]byte, end-offset) + copy(chunk, data[offset:end]) + f.offset = end + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(chunk)), Data: chunk} +} + +func (s *chatServer) readDir(fc *plan9.Fcall, f *fidState, count int) *plan9.Fcall { + entries := []plan9.Dir{ + {Name: "ctl", Qid: fileQid("ctl"), Mode: 0222, Atime: uint32(time.Now().Unix()), Mtime: uint32(time.Now().Unix())}, + {Name: "stream", Qid: fileQid("stream"), Mode: 0444, Atime: uint32(time.Now().Unix()), Mtime: uint32(time.Now().Unix())}, + {Name: "chat", Qid: fileQid("chat"), Mode: 0444, Atime: uint32(time.Now().Unix()), Mtime: uint32(time.Now().Unix())}, + } + var buf []byte + for _, d := range entries { + b, _ := d.Bytes() + buf = append(buf, b...) + } + offset := int64(fc.Offset) + if offset >= int64(len(buf)) { + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0} + } + end := offset + int64(count) + if end > int64(len(buf)) { + end = int64(len(buf)) + } + chunk := buf[offset:end] + return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(chunk)), Data: chunk} +} + +func (s *chatServer) stat(path string) plan9.Dir { + now := uint32(time.Now().Unix()) + u := os.Getenv("USER") + switch path { + case "/": + return plan9.Dir{Name: "/", Qid: dirQid("/"), Mode: plan9.DMDIR | 0555, Atime: now, Mtime: now, Uid: u, Gid: u} + case "/ctl": + return plan9.Dir{Name: "ctl", Qid: fileQid("ctl"), Mode: 0222, Atime: now, Mtime: now, Uid: u, Gid: u} + case "/stream": + s.mu.Lock() + l := uint64(len(s.streamBuf)) + s.mu.Unlock() + return plan9.Dir{Name: "stream", Qid: fileQid("stream"), Mode: 0444, Length: l, Atime: now, Mtime: now, Uid: u, Gid: u} + case "/chat": + s.mu.Lock() + l := uint64(len(s.chat)) + s.mu.Unlock() + return plan9.Dir{Name: "chat", Qid: fileQid("chat"), Mode: 0444, Length: l, Atime: now, Mtime: now, Uid: u, Gid: u} + } + return plan9.Dir{} +} + +// generateResponse simulates streaming LLM token generation with network jitter. +func (s *chatServer) generateResponse(prompt string) { + // Reset stream state + s.mu.Lock() + s.streamBuf = nil + s.streamDone = false + s.chat = append(s.chat, []byte("\n[user]\n"+prompt+"\n\n[assistant]\n")...) + s.mu.Unlock() + s.streamCond.Broadcast() + + // Tokenize into small chunks (simulates token batching) + tokens := tokenize(cannedResponse) + + for _, tok := range tokens { + // Simulate network latency: 30-150ms per batch + delay := time.Duration(30+rand.Intn(120)) * time.Millisecond + // 10% chance of longer pause (network jitter) + if rand.Intn(10) == 0 { + delay += time.Duration(200+rand.Intn(400)) * time.Millisecond + } + time.Sleep(delay) + + s.mu.Lock() + s.streamBuf = append(s.streamBuf, []byte(tok)...) + s.chat = append(s.chat, []byte(tok)...) + s.mu.Unlock() + s.streamCond.Broadcast() + } + + // Mark complete + s.mu.Lock() + s.streamDone = true + s.mu.Unlock() + s.streamCond.Broadcast() + + log.Printf("response complete (%d bytes)", len(s.streamBuf)) +} + +func tokenize(text string) []string { + // Split into chunks of ~20-60 bytes each (simulates token batching from LLM) + var tokens []string + for len(text) > 0 { + n := 20 + rand.Intn(40) + if n > len(text) { + n = len(text) + } + // Try to break at whitespace + if n < len(text) { + for i := n; i > n-10 && i > 0; i-- { + if text[i] == ' ' || text[i] == '\n' { + n = i + 1 + break + } + } + } + tokens = append(tokens, text[:n]) + text = text[n:] + } + return tokens +} + +func dirQid(name string) plan9.Qid { + return plan9.Qid{Type: plan9.QTDIR, Path: hash(name)} +} + +func fileQid(name string) plan9.Qid { + return plan9.Qid{Type: plan9.QTFILE, Path: hash(name)} +} + +func hash(s string) uint64 { + h := uint64(0xcbf29ce484222325) + for _, c := range s { + h ^= uint64(c) + h *= 0x100000001b3 + } + return h +} + +func errResp(fc *plan9.Fcall, msg string) *plan9.Fcall { + return &plan9.Fcall{Type: plan9.Rerror, Tag: fc.Tag, Ename: msg} +} diff --git a/fs/session/files.go b/fs/session/files.go index 9d9ad68..1f1e660 100644 --- a/fs/session/files.go +++ b/fs/session/files.go @@ -34,6 +34,7 @@ var AgentFileList = []struct { {"prompt", false, true}, {"fifo.in", false, true}, {"fifo.out", true, false}, + {"log", false, false}, {"chat", false, false}, {"offset", false, false}, {"cfg", false, false}, @@ -123,18 +124,28 @@ func (h *sessionHelper) fileSpec(name string, mode os.FileMode) FileSpec { h.sess.mu.RUnlock() return data, nil } - case "chat": + case "log": fs.Read = func() ([]byte, error) { + const maxWindow = 64 * 1024 h.sess.mu.RLock() - data := make([]byte, len(h.sess.log)) - copy(data, h.sess.log) + log := h.sess.log + start := 0 + if len(log) > maxWindow { + start = len(log) - maxWindow + } + data := make([]byte, len(log)-start) + copy(data, log[start:]) h.sess.mu.RUnlock() return data, nil } fs.Size = func() int64 { + const maxWindow = 64 * 1024 h.sess.mu.RLock() n := len(h.sess.log) h.sess.mu.RUnlock() + if n > maxWindow { + return int64(maxWindow) + } return int64(n) } case "fifo.out": @@ -161,7 +172,7 @@ func (h *sessionHelper) fileSpec(name string, mode os.FileMode) FileSpec { h.sess.mu.Unlock() return nil } - case "chat": + case "log": fs.Write = func(data []byte) error { input := strings.TrimSpace(string(data)) if input == "" { @@ -248,6 +259,46 @@ func (h *sessionHelper) fileSpec(name string, mode os.FileMode) FileSpec { } } + // Chat stream: blocking read that delivers new log data as it arrives. + // Per-fid offset tracked via waitBase (string-encoded int64). + if name == "chat" { + fs.LongBlock = true + fs.Wait = func(connCtx context.Context, base string) ([]byte, string, error) { + ctx, cancel := context.WithCancel(connCtx) + defer cancel() + context.AfterFunc(h.sess.SessionCtx, cancel) + + // Parse offset from base; empty = start from current end ("now") + var offset int + if base != "" { + fmt.Sscanf(base, "%d", &offset) + } else { + h.sess.mu.RLock() + offset = len(h.sess.log) + h.sess.mu.RUnlock() + } + + // Block until new data is available or context cancelled + h.sess.mu.RLock() + for len(h.sess.log) <= offset { + if ctx.Err() != nil { + h.sess.mu.RUnlock() + // Context cancelled (session killed or connection closed) → EOF + return nil, "", nil + } + h.sess.chatCond.Wait() + } + + // Copy new data + data := make([]byte, len(h.sess.log)-offset) + copy(data, h.sess.log[offset:]) + newOffset := len(h.sess.log) + h.sess.mu.RUnlock() + + return data, fmt.Sprintf("%d", newOffset), nil + } + } + return fs } @@ -600,7 +651,10 @@ func (h *sessionHelper) handleCtl(input string) error { switch cmd[0] { case "stop": h.sess.Core.Agent().Interrupt(agent.ErrInterrupted) - case "kill": + case "kill", ".": + // FIXME: kill currently kills the entire session because we only support + // one agent per session. When multi-agent-per-session lands, kill should + // kill the agent only; session and chat stream persist. h.kill() case "rn": if name := strings.TrimSpace(input[3:]); name != "" { diff --git a/fs/session/fs.go b/fs/session/fs.go index 82438ef..71bac45 100644 --- a/fs/session/fs.go +++ b/fs/session/fs.go @@ -18,6 +18,9 @@ type File interface { OneShot() bool // IsBlocking returns true if Read blocks until data is available (use BlockingRead). IsBlocking() bool + // LongBlock returns true if blocking reads should not have a short timeout + // (e.g., streaming files that block indefinitely until data arrives). + LongBlock() bool // Async returns true if writes should be dispatched asynchronously. Async() bool } @@ -49,6 +52,7 @@ type FileConfig struct { BlockingReadFn func(context.Context, string) ([]byte, string, error) OneShot_ bool IsBlocking_ bool + LongBlock_ bool Async_ bool } @@ -60,6 +64,7 @@ func (e *FileConfig) BlockingRead(ctx context.Context, base string) ([]byte, str } func (e *FileConfig) OneShot() bool { return e.OneShot_ } func (e *FileConfig) IsBlocking() bool { return e.IsBlocking_ } +func (e *FileConfig) LongBlock() bool { return e.LongBlock_ } func (e *FileConfig) Async() bool { return e.Async_ } // SyntheticFileInfo implements os.FileInfo for entries with no backing file. diff --git a/fs/session/perm.go b/fs/session/perm.go index 63d53f8..1e1a983 100644 --- a/fs/session/perm.go +++ b/fs/session/perm.go @@ -71,6 +71,7 @@ var Perms = map[string]Perm{ "prompt": 0666, "fifo.in": 0222, "fifo.out": 0444, + "log": 0444, "chat": 0444, "offset": 0444, "cfg": 0666, diff --git a/fs/session/synth.go b/fs/session/synth.go index 5e2b4a3..aa167dd 100644 --- a/fs/session/synth.go +++ b/fs/session/synth.go @@ -8,14 +8,15 @@ import ( // FileSpec describes a single synthetic file exposed by a Tree. type FileSpec struct { - Name string - Mode os.FileMode - Read func() ([]byte, error) - Write func([]byte) error // nil = read-only - Wait func(ctx context.Context, base string) (content []byte, nextBase string, err error) // nil = not waitable - Size func() int64 // optional; if nil, len(Read()) - OneShot bool // true = yields data once per open - Async bool // true = writes dispatched asynchronously + Name string + Mode os.FileMode + Read func() ([]byte, error) + Write func([]byte) error // nil = read-only + Wait func(ctx context.Context, base string) (content []byte, nextBase string, err error) // nil = not waitable + Size func() int64 // optional; if nil, len(Read()) + OneShot bool // true = yields data once per open + Async bool // true = writes dispatched asynchronously + LongBlock bool // true = blocking reads have no short timeout } // NewSynth creates a Tree backed by a table of FileSpecs (synthetic files). @@ -71,8 +72,9 @@ func NewSynth(specs []FileSpec) *Tree { ReadFn: spec.Read, WriteFn: writeFn, BlockingReadFn: waitFn, - OneShot_: spec.OneShot || spec.Wait != nil, + OneShot_: spec.OneShot || (spec.Wait != nil && !spec.LongBlock), IsBlocking_: spec.Wait != nil, + LongBlock_: spec.LongBlock, Async_: spec.Async, }, nil }), diff --git a/fs/session/types.go b/fs/session/types.go index 94039a4..caaf009 100644 --- a/fs/session/types.go +++ b/fs/session/types.go @@ -27,6 +27,10 @@ type Session struct { remote string // SSH target for remote execution (empty = local) proc *toolsrv.Process // toolsrv subprocess (session owns lifecycle) + // Chat stream: condition variable broadcast on every log append. + // Readers of the `chat` file block on this until new data arrives. + chatCond *sync.Cond + // Cached model list (expensive API call; refreshed every 24h). modelsMu sync.Mutex modelsCache string @@ -35,6 +39,7 @@ type Session struct { func NewSession(id string, core *session.Session, ctx context.Context, cancel context.CancelFunc) *Session { sess := &Session{id: id, Core: core, SessionCtx: ctx, cancel: cancel} + sess.chatCond = sync.NewCond(sess.mu.RLocker()) sess.startEventLog() return sess } @@ -92,6 +97,7 @@ func (sess *Session) AppendLog(data []byte) { sess.log = append(sess.log, data...) sess.logVers++ sess.mu.Unlock() + sess.chatCond.Broadcast() } // EnsureTrailingNewline appends a newline if the log doesn't already end with one.