fix: convert remaining isStarted bool to atomic.Bool and add panic recovery to handleEvents goroutines

- EventBusModule.isStarted: bool → atomic.Bool (module.go)
- CustomMemoryEventBus.isStarted: bool → atomic.Bool (custom_memory.go)
- Update test helpers to use .Load() for atomic bool access
- Add defer recover() to handleEvents goroutine wrapper in memory.go and durable_memory.go

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: intel352

modules/eventbus/bdd_multi_engine_error_isolation_test.go

@@ -30,7 +30,7 @@ func (ctx *EventBusBDDTestContext) oneEngineEncountersAnError() error {
 	}
 
 	// Ensure service is started before trying to publish
-	if !ctx.service.isStarted {
+	if !ctx.service.isStarted.Load() {
 		err := ctx.service.Start(context.Background())
 		if err != nil {
 			return fmt.Errorf("failed to start eventbus: %w", err)

modules/eventbus/bdd_multi_engine_test.go

@@ -136,7 +136,7 @@ func (ctx *EventBusBDDTestContext) iPublishAnEventToTopic(topic string) error {
 	ctx.publishedTopics[topic] = true
 
 	// Start the service if not already started
-	if !ctx.service.isStarted {
+	if !ctx.service.isStarted.Load() {
 		err := ctx.service.Start(context.Background())
 		if err != nil {
 			return fmt.Errorf("failed to start eventbus: %w", err)

modules/eventbus/custom_memory.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"log/slog"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -19,7 +20,7 @@ type CustomMemoryEventBus struct {
 	topicMutex    sync.RWMutex
 	ctx           context.Context
 	cancel        context.CancelFunc
-	isStarted     bool
+	isStarted     atomic.Bool
 	eventMetrics  *EventMetrics
 	eventFilters  []EventFilter
 }
@@ -189,7 +190,7 @@ func NewCustomMemoryEventBus(config map[string]interface{}) (EventBus, error) {
 
 // Start initializes the custom memory event bus
 func (c *CustomMemoryEventBus) Start(ctx context.Context) error {
-	if c.isStarted {
+	if c.isStarted.Load() {
 		return nil
 	}
 
@@ -200,7 +201,7 @@ func (c *CustomMemoryEventBus) Start(ctx context.Context) error {
 		go c.metricsCollector()
 	}
 
-	c.isStarted = true
+	c.isStarted.Store(true)
 	slog.Info("Custom memory event bus started with enhanced features",
 		"metricsEnabled", c.config.EnableMetrics,
 		"filterCount", len(c.eventFilters))
@@ -209,7 +210,7 @@ func (c *CustomMemoryEventBus) Start(ctx context.Context) error {
 
 // Stop shuts down the custom memory event bus
 func (c *CustomMemoryEventBus) Stop(ctx context.Context) error {
-	if !c.isStarted {
+	if !c.isStarted.Load() {
 		return nil
 	}
 
@@ -227,7 +228,7 @@ func (c *CustomMemoryEventBus) Stop(ctx context.Context) error {
 	}
 	c.topicMutex.Unlock()
 
-	c.isStarted = false
+	c.isStarted.Store(false)
 	slog.Info("Custom memory event bus stopped",
 		"totalEvents", c.eventMetrics.TotalEvents,
 		"topics", len(c.eventMetrics.EventsPerTopic))
@@ -236,7 +237,7 @@ func (c *CustomMemoryEventBus) Stop(ctx context.Context) error {
 
 // Publish sends an event to the specified topic with custom filtering and metrics
 func (c *CustomMemoryEventBus) Publish(ctx context.Context, event Event) error {
-	if !c.isStarted {
+	if !c.isStarted.Load() {
 		return ErrEventBusNotStarted
 	}
 
@@ -309,7 +310,7 @@ func (c *CustomMemoryEventBus) SubscribeAsync(ctx context.Context, topic string,
 
 // subscribe is the internal implementation for both Subscribe and SubscribeAsync
 func (c *CustomMemoryEventBus) subscribe(ctx context.Context, topic string, handler EventHandler, isAsync bool) (Subscription, error) {
-	if !c.isStarted {
+	if !c.isStarted.Load() {
 		return nil, ErrEventBusNotStarted
 	}
 
@@ -347,7 +348,7 @@ func (c *CustomMemoryEventBus) subscribe(ctx context.Context, topic string, hand
 
 // Unsubscribe removes a subscription
 func (c *CustomMemoryEventBus) Unsubscribe(ctx context.Context, subscription Subscription) error {
-	if !c.isStarted {
+	if !c.isStarted.Load() {
 		return ErrEventBusNotStarted
 	}
 

modules/eventbus/durable_memory.go

@@ -339,6 +339,11 @@ func (d *DurableMemoryEventBus) subscribe(_ context.Context, topic string, handl
 	started := make(chan struct{})
 	d.wg.Add(1)
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("handleEvents panic recovered", "error", r, "topic", sub.topic)
+			}
+		}()
 		close(started)
 		d.handleEvents(sub)
 	}()

modules/eventbus/memory.go

@@ -360,6 +360,11 @@ func (m *MemoryEventBus) subscribe(ctx context.Context, topic string, handler Ev
 	started := make(chan struct{})
 	m.wg.Add(1)
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("handleEvents panic recovered", "error", r, "topic", sub.topic)
+			}
+		}()
 		close(started) // Signal that the goroutine has started
 		m.handleEvents(sub)
 	}()

modules/eventbus/module.go

@@ -111,6 +111,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/GoCodeAlone/modular"
@@ -146,7 +147,7 @@ type EventBusModule struct {
 	logger    modular.Logger
 	router    *EngineRouter
 	mutex     sync.RWMutex
-	isStarted bool
+	isStarted atomic.Bool
 	subject   modular.Subject // For event observation (guarded by mutex)
 }
 
@@ -290,7 +291,7 @@ func (m *EventBusModule) Start(ctx context.Context) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
-	if m.isStarted {
+	if m.isStarted.Load() {
 		return nil
 	}
 
@@ -300,7 +301,7 @@ func (m *EventBusModule) Start(ctx context.Context) error {
 		return fmt.Errorf("starting engine router: %w", err)
 	}
 
-	m.isStarted = true
+	m.isStarted.Store(true)
 	if m.config.IsMultiEngine() {
 		m.logger.Info("Event bus started with multiple engines",
 			"engines", m.router.GetEngineNames())
@@ -358,7 +359,7 @@ func (m *EventBusModule) Stop(ctx context.Context) error {
 
 	m.mutex.Lock()
 
-	if !m.isStarted {
+	if !m.isStarted.Load() {
 		m.mutex.Unlock()
 		return nil
 	}
@@ -370,7 +371,7 @@ func (m *EventBusModule) Stop(ctx context.Context) error {
 		return fmt.Errorf("stopping engine router: %w", err)
 	}
 
-	m.isStarted = false
+	m.isStarted.Store(false)
 	engineName := "unknown"
 	if m.config != nil {
 		engineName = m.config.Engine