feed: BlockOnce with base init, ConsumeFeed via lib9p client
- BlockOnce handler initializes base to current hash on fresh open, then blocks until hash changes. Same pattern as statewait. - ConsumeFeed is a plain function: dials 9P, reads feed in a loop (open → block → data → close → repeat), submits to agent. - Context cancellation closes the 9P client, unblocking Read(). - Called as go ConsumeFeed(ctx, ag) from AddAgent and session resume.
This commit is contained in:
parent
e2695d6f50
commit
b5cb8c397d
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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", "")
|
||||
|
|
|
|||
Loading…
Reference in New Issue