9P streaming: add chat file, rename chat→log, rewrite acme frontend
- Rename 'chat' to 'log': returns last 64KB sliding window (static read). Tail still works via Qid.Vers. Full history persisted to disk per turn. - Add 'chat' streaming file: blocking read that delivers new output as the agent produces it. Per-fid offset tracked via waitBase. Blocks indefinitely between turns (no EOF). EOF only on kill/session death. - Add LongBlock interface to File: streaming files bypass the 5s read timeout. Server returns content directly (ignores Tread offset for streaming files since waitBase tracks position). - Add 'kill'/'.'' ctl aliases with FIXME: currently kills entire session. When multi-agent-per-session lands, kill should kill agent only. - Rewrite acme frontend (cmd/Ollie) to use 9P client directly instead of FUSE mount. streamChat() is a blocking read loop — no polling. All file operations (read, write, ls) go through plan9/client. - Add 9P streaming prototype in experiments/9p-stream demonstrating the blocking-read pattern for token delivery. The key insight: 9P's request-response model gives natural streaming. Server holds the Tread until data arrives, client blocks on Read(). No polling, no signals, no offset tracking needed.
This commit is contained in:
Levi Neely
parent
7eb022faba
commit
c39149ca3e
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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/<id> 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 }
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,5 @@
|
|||
module ollie/experiments/9p-stream
|
||||
|
||||
go 1.25
|
||||
|
||||
require 9fans.net/go v0.0.7
|
||||
|
|
@ -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=
|
||||
|
|
@ -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 <text>" 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}
|
||||
}
|
||||
|
|
@ -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 != "" {
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Reference in New Issue