Fix/memory leak whatsapp reasoning

[alexhoshina]

📝 Description

Fix memory leak issues in the WhatsApp Native Channel:

Issue 1: QR loop blocking in Start()
Issue 2: fire-and-forget reconnection goroutine

Fix issues in the Reasoning Channel that could cause memory leaks:

Each LLM response initiates an untracked goroutine
al.handleReasoning(ctx, response.Reasoning, opts.Channel, ...)

• Each LLM response fire-and-forgets a goroutine to publish reasoning content
• If the message bus's outbound channel is full, the goroutine will block on PublishOutbound
• High concurrency scenarios may lead to goroutine accumulation, consuming significant memory and stack space

🗣️ Type of Change

• 🐞 Bug fix (non-breaking change which fixes an issue)

🤖 AI Code Generation

• 🛠️ Mostly AI-generated (AI draft, Human verified/modified)
• 👨‍💻 Mostly Human-written (Human lead, AI assisted or none)

☑️ Checklist

• My code/docs follow the style of this project.
• I have performed a self-review of my own changes.
• I have updated the documentation accordingly.

[Copilot]

Pull request overview

Fixes potential goroutine/memory leaks in channel integrations by ensuring background work can be canceled and does not block indefinitely on bounded queues.

Changes:

• WhatsApp native channel: move QR handling off the Start() call path; track QR/reconnect goroutines with a WaitGroup and cancel via runCtx.
• WhatsApp native channel: track reconnect goroutine instead of fire-and-forget.
• Agent reasoning publishing: add a bounded publish timeout and log failures as best-effort.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.

File	Description
pkg/channels/whatsapp_native/whatsapp_native.go	Introduces runCtx earlier, moves QR loop to a tracked goroutine, and tracks reconnect goroutines via WaitGroup.
pkg/agent/loop.go	Adds a timeout + error handling for reasoning publishes to avoid indefinite blocking when the outbound bus is full.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

client.GetQRChannel is still called with the parent ctx, but the QR reader goroutine exits on c.runCtx.Done() during Stop(). If GetQRChannel uses its context to stop its internal producer, canceling only runCtx can leave the producer running and blocked trying to send into qrChan after the reader exits. Use c.runCtx (or another context canceled by Stop()) when calling GetQRChannel so the producer is also canceled/closed when stopping.

Suggested change

	qrChan, err := client.GetQRChannel(ctx)
	qrChan, err := client.GetQRChannel(c.runCtx)

[Copilot]

Stop() calls c.wg.Wait() while eventHandler can still call c.wg.Add(1) concurrently (e.g., a *events.Disconnected arriving during shutdown). This is a WaitGroup misuse and can panic with sync: WaitGroup misuse: Add called concurrently with Wait, or let reconnect goroutines run after cleanup. Add a shutdown/stopping guard (protected by a mutex/atomic) that prevents new goroutines from being added once stopping begins, and/or ensure the client can't dispatch events before Wait() is entered.

[Copilot]

Stop(ctx) now waits unconditionally on c.wg.Wait(), but the tracked goroutines can block for an unbounded time (notably reconnectWithBackoff during client.Connect()), and Stop ignores its ctx parameter. Consider making the wait context-aware (e.g., wait in a separate goroutine and select on ctx.Done()), and/or disconnect the client before waiting so Connect() can be interrupted.

Suggested change

	// Wait for background goroutines (QR handler, reconnect) to finish so
	// they don't reference the client/container after cleanup.
	c.wg.Wait()
	c.mu.Lock()
	client := c.client
	container := c.container
	// Grab current client/container so we can disconnect the client before waiting.
	c.mu.Lock()
	client := c.client
	container := c.container
	c.mu.Unlock()
	// Disconnect the client first so any blocking Connect()/reconnect loops
	// can be interrupted before we wait on the goroutines.
	if client != nil {
	client.Disconnect()
	}
	// Wait for background goroutines (QR handler, reconnect) to finish in a
	// context-aware way so Stop can be bounded by ctx.
	done := make(chan struct{})
	go func() {
	c.wg.Wait()
	close(done)
	}()
	select {
	case <-done:
	// All goroutines have finished.
	case <-ctx.Done():
	// Context canceled or timed out; log and proceed with best-effort cleanup.
	logger.WarnC("whatsapp", "Stop context canceled before all goroutines finished: %v", ctx.Err())
	}
	// Now it is safe to clear and close resources.
	c.mu.Lock()
	client = c.client
	container = c.container

[Copilot]

There are tests for handleReasoning, but none cover the new “bus is full / publish times out” path. Add a test that fills the outbound buffer and uses a short-deadline parent context to assert handleReasoning returns promptly (and does not publish) when PublishOutbound can’t enqueue before the deadline.

[Copilot]

handleReasoning treats reasoning as best-effort, but logs a warning on every publish failure/timeout. Under load (when the outbound bus is full), this can become very noisy and add its own performance/ops overhead. Consider lowering the log level, sampling/rate-limiting, or only warning on unexpected errors (e.g., excluding context deadline/cancel).

[Huaaudio]

Nice fix to save the real memory leak problem. LGTM

[yinwm]

Review: TOCTOU Race Condition in eventHandler

Thanks for addressing these memory leak issues. The overall approach is sound, but there's a critical race condition that still needs to be fixed before merging.

The Problem

In eventHandler(), there's a TOCTOU (Time-Of-Check-Time-Of-Use) race between the stopping check and wg.Add(1):

case *events.Disconnected:
    if c.stopping.Load() {  // Check
        return
    }
    c.reconnectMu.Lock()
    // ...
    c.reconnectMu.Unlock()
    c.wg.Add(1)              // Use - NOT atomic with the check above
    go func() { ... }()

Race scenario:

1. eventHandler checks stopping.Load() → returns false
2. Stop() is called, sets stopping.Store(true)
3. Stop() calls c.wg.Wait() → returns immediately (wg is still 0)
4. Stop() clears c.client = nil and returns
5. eventHandler continues, calls c.wg.Add(1), starts reconnect goroutine
6. Result: Goroutine leak (or worse, sync: WaitGroup misuse panic)

As Copilot noted, this can cause a panic: sync: WaitGroup misuse: Add called concurrently with Wait.

Suggested Fix

Move wg.Add(1) inside the lock, and check stopping while holding the lock:

case *events.Disconnected:
    c.reconnectMu.Lock()
    if c.reconnecting {
        c.reconnectMu.Unlock()
        return
    }
    if c.stopping.Load() {  // Check stopping while holding the lock
        c.reconnectMu.Unlock()
        return
    }
    c.reconnecting = true
    c.wg.Add(1)  // Add to WaitGroup while holding the lock
    c.reconnectMu.Unlock()
    go func() {
        defer c.wg.Done()
        c.reconnectWithBackoff()
    }()

This ensures the stopping check and wg.Add(1) are atomic with respect to Stop().

Minor Observations

• The test coverage and context-aware Stop() improvements look good
• The log level differentiation (Warn for unexpected, Debug for context cancellation) is a nice touch

[nikolasdehor]

Solid fix addressing three distinct memory leak vectors in the WhatsApp native channel. Let me break down each:

1. QR loop blocking in Start() -- Previously the for evt := range qrChan loop ran synchronously in Start(), which meant Start() would not return until QR auth completed (or the channel closed). Now it runs in a goroutine tracked by c.wg. Good.

2. Fire-and-forget reconnect goroutine -- go c.reconnectWithBackoff() was not tracked, so Stop() could return while reconnection was still in progress. Now tracked via c.wg.Add(1) / defer c.wg.Done().

3. Reasoning goroutine accumulation -- handleReasoning used to call PublishOutbound without timeout, so if the outbound bus was full, goroutines would pile up indefinitely. The 5-second timeout with best-effort semantics is the right tradeoff -- reasoning output is supplementary and dropping it is acceptable.

Architecture observations:

• Moving runCtx/runCancel creation before client.Connect() is correct -- it ensures Stop() can cancel the QR flow at any time.
• The stopping atomic bool prevents new goroutines from being spawned after Stop() begins. This closes the race where a disconnect event fires between Stop() start and runCancel().
• The context-aware wg.Wait() in Stop() with fallback to best-effort cleanup is well designed -- prevents Stop() from hanging indefinitely if a goroutine is stuck.
• The test for the bus-full scenario verifies the timeout behavior correctly.

One minor concern (non-blocking): wg.Add(1) in the event handler runs on the whatsmeow event dispatch goroutine. If stopping.Store(true) and wg.Add(1) race, there is a tiny window where wg.Add(1) is called after Stop()'s wg.Wait() has already started, which would panic. The stopping.Load() check before wg.Add(1) mitigates this, but the ordering is not guaranteed by the memory model since stopping.Store and runCancel are not synchronized. In practice the client.Disconnect() call before wg.Wait() should prevent whatsmeow from dispatching more events, so this should be safe. Worth a comment noting this assumption.

LGTM.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

In the QR-login path (Store.ID == nil), Start() now returns immediately but still calls SetRunning(true). That means the channel manager will start outbound workers while the client may not be paired yet, and Send() currently only checks IsConnected() (not Store.ID / login state). Consider delaying SetRunning(true) until pairing completes, or have Send() explicitly detect the unpaired state (e.g., Store.ID == nil) and return a clear temporary/not-running error to avoid futile retries/log noise during QR setup.

[Copilot]

The error logging path checks ctx.Err() to decide whether the PublishOutbound error is expected, but the failure may come from pubCtx timing out (5s) while the parent ctx is still active. In that case ctx.Err() is nil and this will incorrectly warn for the expected "bus full" timeout. Consider basing the decision on err / pubCtx.Err() (e.g., treat context.Canceled/DeadlineExceeded as expected) instead of ctx.Err().

[Copilot]

This test hard-codes the outbound bus buffer size (64) and uses context.Background() when filling it. If the bus buffer size changes in the future, the fill loop could block indefinitely and hang the test suite. Consider filling until PublishOutbound fails using a short timeout context (or introducing a helper that determines/fills capacity) so the test remains robust to buffer-size changes.

Suggested change

	// Fill the outbound bus buffer (default size is 64) so that the
	// next PublishOutbound will block until the context deadline.
	for i := 0; i < 64; i++ {
	err := msgBus.PublishOutbound(context.Background(), bus.OutboundMessage{
	Channel: "filler",
	ChatID: "filler",
	Content: fmt.Sprintf("filler-%d", i),
	})
	if err != nil {
	t.Fatalf("failed to fill bus: %v", err)
	}
	}
	// Fill the outbound bus buffer so that the next PublishOutbound will
	// block until the context deadline. Do not assume a specific buffer
	// size; instead, publish until PublishOutbound fails with a short
	// timeout context.
	const maxFillAttempts = 10000
	filledCount := 0
	for i := 0; i < maxFillAttempts; i++ {
	fillCtx, fillCancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
	err := msgBus.PublishOutbound(fillCtx, bus.OutboundMessage{
	Channel: "filler",
	ChatID: "filler",
	Content: fmt.Sprintf("filler-%d", i),
	})
	fillCancel()
	if err != nil {
	// We assume the bus is full or otherwise unable to accept more
	// messages once PublishOutbound returns an error.
	if filledCount == 0 {
	t.Fatalf("failed to publish to outbound bus even once: %v", err)
	}
	break
	}
	filledCount++
	}
	if filledCount == maxFillAttempts {
	t.Fatalf("failed to reach outbound bus capacity after %d attempts; test may not reflect fullness semantics", maxFillAttempts)
	}

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

c.stopping is set to true in Stop() but never reset in Start(). If this channel instance is ever restarted after a Stop(), the disconnect handler will permanently skip reconnection attempts because eventHandler returns early when stopping is true. Consider resetting stopping (and any related reconnect state) at the beginning of Start(), under the same lock used in Stop()/eventHandler, so lifecycle restarts behave correctly.

[Copilot]

c.wg.Add(1) for the QR handler goroutine is not coordinated with Stop() calling c.wg.Wait(). If Stop() can run concurrently with Start() (the comments imply it can, e.g. during QR-login), there is a real sync.WaitGroup misuse risk: Stop may enter Wait() while Start subsequently calls Add(1), which can panic. To fix, ensure all wg.Add calls are prevented once Stop begins (e.g., guard wg.Add with the same mutex/flag protocol used in eventHandler, or introduce a lifecycle mutex that serializes Start/Stop).

Suggested change

	// promptly. The goroutine is tracked via c.wg and respects
	// c.runCtx for cancellation.
	c.wg.Add(1)
	go func() {
	defer c.wg.Done()
	// promptly. The goroutine respects c.runCtx for cancellation.
	go func() {

[yinwm]

Review Summary

After multiple iterations, this PR has successfully fixed the memory leak issues:

✅ Issues Fixed

1. QR loop blocking Start() - QR event handling moved to background goroutine
2. Fire-and-forget reconnect goroutine - Now tracked with WaitGroup
3. Reasoning goroutine accumulation - Added 5-second timeout
4. TOCTOU race condition - Properly fixed in commit fc28c26

Code Quality

• Start()'s defer cleanup ensures resources are properly released
• Stop() is context-aware, supporting graceful shutdown
• reconnectWithBackoff() correctly responds to runCtx cancellation
• Tests cover timeout behavior when bus is full

Minor Suggestions (Non-blocking)

1. reconnectMu could be renamed to lifecycleMu for clarity
2. Test threshold of 2s could be reduced to 1s for tighter bounds
3. Stop() timeout behavior could be documented in comments

LGTM 👍