From 57984dfa3fa9219991fe55a4c9e213cc5ebbab4b Mon Sep 17 00:00:00 2001 From: Ollie Agent Date: Sun, 9 Aug 2026 01:08:41 +0200 Subject: [PATCH] agent: replace sync.Cond with close-channel signal for WaitChange MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sync.Cond is the wrong primitive for Go — it requires holding a mutex during Wait and needs context.AfterFunc hacks to integrate with ctx cancellation. Replace with a simple signal channel: notifyChange() closes it and swaps in a fresh one; WaitChange() selects on the channel and ctx.Done(). Simpler, idiomatic, no spurious wakeups. --- agent/agent.go | 45 +++++++++++++++++++-------------------------- 1 file changed, 19 insertions(+), 26 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index 676984e..8d33510 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -41,10 +41,10 @@ type Agent struct { fifo Fifo // prompt queue toolCallCount atomic.Int64 pendingInject atomic.Pointer[string] - submitMu sync.Mutex // serializes Submit calls (commands + turns) - stateMu sync.RWMutex - changeMu sync.Mutex - changeCond *sync.Cond + 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 @@ -171,43 +171,36 @@ func (ag *Agent) setReply(reply string) { // notifyChange wakes all goroutines waiting on state changes. func (ag *Agent) notifyChange() { - ag.changeMu.Lock() - ag.changeCond.Broadcast() - ag.changeMu.Unlock() + ag.signalMu.Lock() + close(ag.signalCh) + ag.signalCh = make(chan struct{}) + ag.signalMu.Unlock() } // 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) { - done := make(chan struct{}) - context.AfterFunc(ctx, func() { - ag.changeMu.Lock() - ag.changeCond.Broadcast() - ag.changeMu.Unlock() - close(done) - }) - - ag.changeMu.Lock() for { + // Snapshot the signal channel before reading state. + ag.signalMu.Lock() + ch := ag.signalCh + ag.signalMu.Unlock() + var val string switch field { case WatchState: val = ag.State() default: - ag.changeMu.Unlock() return "", false } if val != current { - ag.changeMu.Unlock() return val, true } - if ctx.Err() != nil { - ag.changeMu.Unlock() - return "", false - } - ag.changeCond.Wait() - if ctx.Err() != nil { - ag.changeMu.Unlock() + // Wait for either a state change or context cancellation. + select { + case <-ch: + // State changed — loop to re-check. + case <-ctx.Done(): return "", false } } @@ -588,7 +581,7 @@ func NewAgent(cfg AgentCfg) *Agent { flush: cfg.Flush, state: "idle", } - ag.changeCond = sync.NewCond(&ag.changeMu) + ag.signalCh = make(chan struct{}) ag.wireToolsChanged() return ag }