diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index 7a97c50..d6ade3c 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -688,8 +688,8 @@ func (ag *Agent) CompactionModel() string { } // Queue pushes a prompt onto the agent's FIFO. -func (ag *Agent) Queue(prompt string) { - ag.fifo.Push(prompt) +func (ag *Agent) Queue(prompt string) error { + return ag.fifo.Push(prompt) } // PopQueue pops the next prompt from the FIFO. diff --git a/cmd/olliesrv/internal/agent/fifo.go b/cmd/olliesrv/internal/agent/fifo.go index a4f5dfb..7efe980 100644 --- a/cmd/olliesrv/internal/agent/fifo.go +++ b/cmd/olliesrv/internal/agent/fifo.go @@ -1,18 +1,32 @@ package agent -import "sync" +import ( + "fmt" + "sync" +) + +const ( + fifoMaxItems = 256 + fifoMaxBytes = 4 * 1024 * 1024 +) // Fifo is a thread-safe queue for buffered prompts. type Fifo struct { mu sync.Mutex items []string + bytes int } // Push appends a prompt to the queue. -func (f *Fifo) Push(s string) { +func (f *Fifo) Push(s string) error { f.mu.Lock() + defer f.mu.Unlock() + if len(s) > fifoMaxBytes || len(f.items) >= fifoMaxItems || f.bytes+len(s) > fifoMaxBytes { + return fmt.Errorf("prompt queue limit reached") + } f.items = append(f.items, s) - f.mu.Unlock() + f.bytes += len(s) + return nil } // Pop removes and returns the next prompt. @@ -24,6 +38,8 @@ func (f *Fifo) Pop() (string, bool) { return "", false } s := f.items[0] + f.items[0] = "" f.items = f.items[1:] + f.bytes -= len(s) return s, true } diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index ae22776..f9072b4 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -639,7 +639,9 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl if input == "" { return nil } - a.Queue(input) + if err := a.Queue(input); err != nil { + return err + } // If the agent is idle, trigger processing. if !a.IsRunning() { if !startAsync(s.Ctx, func() {