diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index b841eb4..0fdfec8 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -37,21 +37,21 @@ type Agent struct { resultCache sync.Map // Execution state — owned by the agent, protected by stateMu. - state string // "idle", "thinking", "calling: " - reply string // last assistant response - getCwd func() string // returns session working directory - id string // agent identity (unique principal) - parentID string // immutable ID of the agent that spawned this agent - depth int // sub-agent depth (0=top-level, 1=sub-agent, 2=sub-sub-agent) - activeChildren atomic.Int32 // number of currently-running child sub-agents - fifo Fifo // prompt queue - Feed Feed // streaming input gate (change-detecting) - toolCallCount atomic.Int64 - pendingInject atomic.Pointer[string] - submitMu sync.Mutex // serializes Submit calls (commands + turns) - stateMu sync.RWMutex - signalMu sync.Mutex - signalCh chan struct{} // closed on state change; replaced with a fresh channel + state string // "idle", "thinking", "calling: " + reply string // last assistant response + getCwd func() string // returns session working directory + id string // agent identity (unique principal) + parentID string // immutable ID of the agent that spawned this agent + depth int // sub-agent depth (0=top-level, 1=sub-agent, 2=sub-sub-agent) + activeChildren atomic.Int32 // number of currently-running child sub-agents + fifo Fifo // prompt queue + Feed Feed // streaming input gate (change-detecting) + toolCallCount atomic.Int64 + pendingInject atomic.Pointer[string] + submitMu sync.Mutex // serializes Submit calls (commands + turns) + stateMu sync.RWMutex + signalMu sync.Mutex + signalCh chan struct{} // closed on state change; replaced with a fresh channel // Event handler — receives events from this agent. output EventHandler @@ -700,10 +700,17 @@ func ConsumeFeed(ctx context.Context, ag *Agent) { ag.log.Error("feed consumer: attach: %v", err) return } + feedDone := make(chan struct{}) go func() { - <-ctx.Done() - fsys.Close() + select { + case <-ctx.Done(): + fsys.Close() + case <-feedDone: + } }() + defer close(feedDone) + defer fsys.Close() + defer conn.Close() for ctx.Err() == nil { fid, err := fsys.Open(path, plan9.OREAD) if err != nil {