diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index d8fa365..daeaa94 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -237,6 +237,14 @@ func (ag *Agent) notifyChange() { ag.signalMu.Unlock() } +// SignalCh returns the current signal channel (closed on any change). +func (ag *Agent) SignalCh() <-chan struct{} { + ag.signalMu.Lock() + ch := ag.signalCh + ag.signalMu.Unlock() + return ch +} + // WaitChange blocks until the agent's state differs from current. // Returns the new value and true, or ("", false) if ctx is cancelled. func (ag *Agent) WaitChange(ctx context.Context, field, current string) (string, bool) { diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index d3d6e38..5f36ae4 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -48,6 +48,9 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl { treeSpecUID = u.Username } + // Event subscription for eventwait (lives for server lifetime). + eventRead, eventSignal := session.EventValue(session.SubscribeEvents(context.Background(), "*")) + return virtfs.DirNode("/", virtfs.UID(treeSpecUID), virtfs.GID(treeSpecGID), @@ -131,11 +134,10 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl { virtfs.FileNode("eventwait", 0444, virtfs.Doc("Blocking read for server events"), virtfs.Read(func() ([]byte, error) { - return nil, fmt.Errorf("eventwait: use blocking read") - }), - virtfs.BlockOnce(func(cctx context.Context, _ string) ([]byte, string, error) { - return session.WaitEvent(cctx) + data, _, _ := eventRead() + return data, nil }), + virtfs.BlockOnce(eventRead, eventSignal), ), virtfs.FileNode("generate", 0666, virtfs.Doc("One-shot LLM generation"), @@ -405,20 +407,9 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl } return nil }), - virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) { - merged, cancel := mergeCtx(s.Ctx, cctx) - defer cancel() - if base == "" { - base = a.Feed.Hash() - } - _, ok := a.WaitChange(merged, agent.WatchFeed, base) - if !ok { - return nil, base, merged.Err() - } - data := a.Feed.Get() - hash := a.Feed.Hash() - return data, hash, nil - }), + virtfs.BlockOnce(func() ([]byte, string, error) { + return a.Feed.Get(), a.Feed.Hash(), nil + }, a.SignalCh), ), virtfs.FileNode("chat", 0444, virtfs.StatOverride(chatStat), @@ -463,19 +454,10 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl virtfs.Read(func() ([]byte, error) { return []byte(a.State() + "\n"), nil }), - virtfs.BlockOnce(func(cctx context.Context, base string) ([]byte, string, error) { - merged, cancel := mergeCtx(s.Ctx, cctx) - defer cancel() - if base == "" { - base = a.State() - } - v, ok := a.WaitChange(merged, agent.WatchState, base) - if !ok { - st := a.State() - return []byte(st + "\n"), st, nil - } - return []byte(v + "\n"), v, nil - }), + virtfs.BlockOnce(func() ([]byte, string, error) { + st := a.State() + return []byte(st + "\n"), st, nil + }, a.SignalCh), ), virtfs.FileNode("log", 0444, virtfs.Read(func() ([]byte, error) { diff --git a/cmd/olliesrv/internal/session/event.go b/cmd/olliesrv/internal/session/event.go index 31348f6..1a2d5f7 100644 --- a/cmd/olliesrv/internal/session/event.go +++ b/cmd/olliesrv/internal/session/event.go @@ -2,7 +2,10 @@ package session import ( "context" + "crypto/sha256" + "encoding/hex" "strings" + "sync" "github.com/simonfxr/pubsub" ) @@ -16,19 +19,6 @@ type Event struct { } // PublishEvent publishes an event to a specific topic. -// Topic hierarchy: -// -// session.{sid}.new/kill/pause/resume/rename -// session.{sid}.agent.{aid}.new/kill/state -// -// Wildcard routing publishes to all ancestor wildcards: -// -// session.{sid}.agent.{aid}.state -> also publishes to: -// session.{sid}.agent.{aid}.* -// session.{sid}.agent.* -// session.{sid}.* -// session.* -// * func PublishEvent(topic, payload string) { ev := Event{Topic: topic, Payload: payload} eventBus.Publish(topic, ev) @@ -43,7 +33,6 @@ func PublishEvent(topic, payload string) { } // SubscribeEvents returns a channel that receives all events matching the topic. -// Use "*" for all events, "session.*" for all session events, etc. func SubscribeEvents(ctx context.Context, topic string) <-chan Event { ch := make(chan Event, 64) sub := eventBus.SubscribeChan(topic, ch, pubsub.KeepOpen) @@ -55,20 +44,42 @@ func SubscribeEvents(ctx context.Context, topic string) <-chan Event { return ch } -// WaitEvent blocks until an event is available. Returns the event as "topic payload\n". -func WaitEvent(ctx context.Context) ([]byte, string, error) { - ch := SubscribeEvents(ctx, "*") - select { - case ev, ok := <-ch: - if !ok { - return nil, "", nil +// EventValue adapts an event channel into BlockOnce-compatible (readFn, signalFn). +// A goroutine drains the channel; each arrival updates the value and fires the signal. +func EventValue(ch <-chan Event) (readFn func() ([]byte, string, error), signalFn func() <-chan struct{}) { + var mu sync.Mutex + var data []byte + var hash string + sig := make(chan struct{}) + + go func() { + for ev := range ch { + line := ev.Topic + if ev.Payload != "" { + line += " " + ev.Payload + } + d := []byte(line + "\n") + h := sha256.Sum256(d) + mu.Lock() + data = d + hash = hex.EncodeToString(h[:8]) + close(sig) + sig = make(chan struct{}) + mu.Unlock() } - line := ev.Topic - if ev.Payload != "" { - line += " " + ev.Payload - } - return []byte(line + "\n"), "", nil - case <-ctx.Done(): - return nil, "", nil + }() + + readFn = func() ([]byte, string, error) { + mu.Lock() + defer mu.Unlock() + out := make([]byte, len(data)) + copy(out, data) + return out, hash, nil } + signalFn = func() <-chan struct{} { + mu.Lock() + defer mu.Unlock() + return sig + } + return } diff --git a/cmd/toolsrv/internal/server/server.go b/cmd/toolsrv/internal/server/server.go index ba8dd2a..c2f6081 100644 --- a/cmd/toolsrv/internal/server/server.go +++ b/cmd/toolsrv/internal/server/server.go @@ -134,7 +134,7 @@ func Spec(srv *Server) virtfs.FsNodeDecl { virtfs.DirNode("bypass", virtfs.FileNode("pending", 0444, virtfs.Doc("Blocks until bypass request; returns JSON {id, cmd, cwd, env}"), - virtfs.BlockOnce(func(ctx context.Context, _ string) ([]byte, string, error) { + virtfs.BlockOnceRaw(func(ctx context.Context, _ string) ([]byte, string, error) { req := bypass.NextPending(ctx) if req == nil { return nil, "", fmt.Errorf("context cancelled") @@ -211,7 +211,7 @@ func Spec(srv *Server) virtfs.FsNodeDecl { ), virtfs.FileNode("wait", 0444, virtfs.Doc("Block until process exits, returns exit code"), - virtfs.BlockOnce(func(_ context.Context, _ string) ([]byte, string, error) { + virtfs.BlockOnceRaw(func(_ context.Context, _ string) ([]byte, string, error) { exitCode := p.Wait() return []byte(fmt.Sprintf("%d\n", exitCode)), "", nil }), diff --git a/virtfs/decl.go b/virtfs/decl.go index eb5d9fe..4aa03a1 100644 --- a/virtfs/decl.go +++ b/virtfs/decl.go @@ -127,8 +127,37 @@ func Stream(fn func(context.Context, string) ([]byte, string, error)) NodeOption return func(d *FsNodeDecl) { d.Stream = fn } } -// BlockOnce sets a blocking read handler that returns once. -func BlockOnce(fn func(context.Context, string) ([]byte, string, error)) NodeOption { +// BlockOnce sets a blocking read handler that returns once per open. +// readFn returns (data, hash, error) — the current value. +// signal returns a channel that is closed when the value may have changed. +// The framework blocks until readFn returns a hash different from base. +func BlockOnce(readFn func() ([]byte, string, error), signal func() <-chan struct{}) NodeOption { + return func(d *FsNodeDecl) { + d.BlockOnce = func(ctx context.Context, base string) ([]byte, string, error) { + for { + data, hash, err := readFn() + if err != nil { + return nil, base, err + } + if base == "" { + base = hash + } + if hash != base { + return data, hash, nil + } + select { + case <-signal(): + case <-ctx.Done(): + return nil, "", nil // timeout: empty response, client re-opens + } + } + } + } +} + +// BlockOnceRaw sets a blocking read handler with custom blocking logic. +// Use for queue-style consumers where there is no "current value" to compare. +func BlockOnceRaw(fn func(context.Context, string) ([]byte, string, error)) NodeOption { return func(d *FsNodeDecl) { d.BlockOnce = fn } } diff --git a/virtfs/decl_test.go b/virtfs/decl_test.go index 71647f5..1b4f121 100644 --- a/virtfs/decl_test.go +++ b/virtfs/decl_test.go @@ -23,7 +23,7 @@ func TestNodeOptions(t *testing.T) { }) t.Run("BlockOnce", func(t *testing.T) { - spec := FileNode("test", 0444, BlockOnce(func(ctx context.Context, base string) ([]byte, string, error) { + spec := FileNode("test", 0444, BlockOnceRaw(func(ctx context.Context, base string) ([]byte, string, error) { return nil, "", nil })) if spec.BlockOnce == nil { @@ -249,7 +249,7 @@ func TestRenameDirNotFound(t *testing.T) { func TestOpenLeafBlockOnce(t *testing.T) { spec := DirNode("/", - FileNode("blocking", 0444, BlockOnce(func(ctx context.Context, base string) ([]byte, string, error) { + FileNode("blocking", 0444, BlockOnceRaw(func(ctx context.Context, base string) ([]byte, string, error) { return []byte("data"), "next", nil })), ) @@ -483,7 +483,7 @@ func TestValidateMultipleBlockingModes(t *testing.T) { spec := DirNode("/", FileNode("bad", 0444, - BlockOnce(func(context.Context, string) ([]byte, string, error) { return nil, "", nil }), + BlockOnceRaw(func(context.Context, string) ([]byte, string, error) { return nil, "", nil }), Stream(func(context.Context, string) ([]byte, string, error) { return nil, "", nil }), ), )