diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index e7ad5a1..d8fa365 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -11,6 +11,7 @@ import ( "syscall" "ollie/cmd/olliesrv/internal/backend" + lib9p "ollie/lib9p/client" olog "ollie/log" "ollie/paths" "ollie/toolsrv" @@ -680,25 +681,31 @@ func (ag *Agent) FeedWrite(data []byte) { ag.notifyChange() } -// StartFeedConsumer spawns a goroutine that blocks on feed changes and -// submits new data as prompts. Exits when ctx is cancelled. -func (ag *Agent) StartFeedConsumer(ctx context.Context) { +// ConsumeFeed reads from the agent's feed file in a loop via 9P. +// Blocks until feed changes, submits the new data as a prompt. +// Exits when ctx is cancelled or on read error. +func ConsumeFeed(ctx context.Context, ag *Agent) { + path := fmt.Sprintf("session/%s/agent/%s/feed", ag.sessionID, ag.id) + client, err := lib9p.DialDefault() + if err != nil { + ag.log.Error("feed consumer: dial: %v", err) + return + } go func() { - var base string - for { - _, ok := ag.WaitChange(ctx, WatchFeed, base) - if !ok { - return - } - data := ag.Feed.Get() - base = ag.Feed.Hash() - if len(data) == 0 { - continue - } - ag.Submit(ctx, string(data)) - ag.EnsureTrailingNewline() - } + <-ctx.Done() + client.Close() }() + for ctx.Err() == nil { + data, err := client.Read(path) + if err != nil { + return + } + if len(data) == 0 { + continue + } + ag.Submit(ctx, string(data)) + ag.EnsureTrailingNewline() + } } diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index 20ed712..d3d6e38 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -408,11 +408,16 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl 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() } - return a.Feed.Get(), a.Feed.Hash(), nil + data := a.Feed.Get() + hash := a.Feed.Hash() + return data, hash, nil }), ), virtfs.FileNode("chat", 0444, diff --git a/cmd/olliesrv/internal/session/session.go b/cmd/olliesrv/internal/session/session.go index 40cb894..47ef5e9 100644 --- a/cmd/olliesrv/internal/session/session.go +++ b/cmd/olliesrv/internal/session/session.go @@ -179,7 +179,7 @@ func (s *Session) Agents() []*agent.Agent { return s.agents } func (s *Session) AddAgent(ag *agent.Agent) { s.agents = append(s.agents, ag) if s.Ctx != nil { - ag.StartFeedConsumer(s.Ctx) + go agent.ConsumeFeed(s.Ctx, ag) } } @@ -336,7 +336,7 @@ func (s *Session) Resume() error { s.paused = false // Start feed consumers for restored agents (context now available). for _, ag := range s.agents { - ag.StartFeedConsumer(ctx) + go agent.ConsumeFeed(ctx, ag) } go s.saveSession() PublishEvent("session."+s.ID+".resume", "")