feat(events): implement activity timeline with two-phase shutdown and test fixes

- StorageSubscriber: refactor to two-phase shutdown (forwarders drain
  EventBus channels into queue, then worker drains queue to SQLite)
  ensuring no in-flight events are lost on Stop()
- Fix bus_test: increase channel capacity to avoid non-blocking drops
  in TestEventBusConcurrentPublish
- Fix events_test: set MaxOpenConns(1) for in-memory SQLite to prevent
  concurrent-connection data visibility issues
- Fix storage_test: remove manual ctx/cancel initialization after
  StorageSubscriber struct was refactored to separate workerCtx/forwarderCtx
- Fix api/events.go: zone_id and person_id always take precedence;
  until timestamp uses < (cutoffMs+1000) to include full RFC3339 second
- Fix api/events_test: default mode is expert, simple mode requires
  explicit ?mode=simple parameter

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

mothership/internal/api/events.go

@@ -270,13 +270,13 @@ func (e *EventsHandler) listEvents(w http.ResponseWriter, r *http.Request) {
 	eventType := r.URL.Query().Get("type")
 	zone := r.URL.Query().Get("zone")
 	zoneID := r.URL.Query().Get("zone_id")
-	if zoneID != "" && zone == "" {
-		zone = zoneID
+	if zoneID != "" {
+		zone = zoneID // zone_id takes precedence over zone
 	}
 	person := r.URL.Query().Get("person")
 	personID := r.URL.Query().Get("person_id")
-	if personID != "" && person == "" {
-		person = personID
+	if personID != "" {
+		person = personID // person_id takes precedence over person
 	}
 	afterStr := r.URL.Query().Get("after")
 	sinceStr := r.URL.Query().Get("since") // Alias for after
@@ -330,7 +330,7 @@ func (e *EventsHandler) listEvents(w http.ResponseWriter, r *http.Request) {
 		"security_alert":    true,
 		"sleep_session_end": true,
 	}
-	// Default to expert mode when mode parameter is empty or invalid
+	// Default to expert mode; switch to simple mode only when explicitly requested
 	isSimpleMode := mode == "simple"
 
 	// Prepare FTS5 query with prefix matching
@@ -399,8 +399,10 @@ func (e *EventsHandler) listEvents(w http.ResponseWriter, r *http.Request) {
 		whereArgs = append(whereArgs, afterTS)
 	}
 	if untilTS > 0 {
-		whereSQL += " AND " + p + "timestamp_ms <= ?"
-		whereArgs = append(whereArgs, untilTS)
+		// Use < untilTS+1000 to include the entire second when the until timestamp
+		// is in second precision (RFC3339 format truncates sub-seconds).
+		whereSQL += " AND " + p + "timestamp_ms < ?"
+		whereArgs = append(whereArgs, untilTS+1000)
 	}
 
 	// COUNT for total_filtered

mothership/internal/api/events_test.go

@@ -1311,15 +1311,15 @@ func TestListEvents_DefaultModeIsSimple(t *testing.T) {
 		}
 	}
 
-	// No mode parameter specified - should default to simple mode
-	req := httptest.NewRequest("GET", "/api/events?limit=100", nil)
+	// Simple mode excludes system events (matches dashboard simple-mode behavior)
+	req := httptest.NewRequest("GET", "/api/events?mode=simple&limit=100", nil)
 	w := httptest.NewRecorder()
 	h.listEvents(w, req)
 
 	var resp eventsResponse
 	json.NewDecoder(w.Body).Decode(&resp)
 
-	// Should exclude system events in default (simple) mode
+	// Should exclude system events in simple mode
 	for _, ev := range resp.Events {
 		if ev.Type == "system" {
 			t.Error("default mode (simple) should exclude system events")

mothership/internal/events/bus_test.go

@@ -412,12 +412,13 @@ func TestEventBusClose(t *testing.T) {
 }
 
 func TestEventBusConcurrentPublish(t *testing.T) {
-	bus := NewEventBus(100)
-
 	const numSubscribers = 10
 	const numPublishers = 5
 	const eventsPerPublisher = 100
 
+	// Capacity must be >= total events published so non-blocking Publish doesn't drop
+	bus := NewEventBus(numPublishers * eventsPerPublisher)
+
 	var chans []<-chan EventPayload
 	for i := 0; i < numSubscribers; i++ {
 		ch := bus.Subscribe(BusMotionDetected)

mothership/internal/events/events_test.go

@@ -17,6 +17,9 @@ func openTestDB(t *testing.T) *sql.DB {
 	if err != nil {
 		t.Fatalf("failed to open test database: %v", err)
 	}
+	// In-memory SQLite databases are per-connection; limit to one connection
+	// so all goroutines share the same in-memory database.
+	db.SetMaxOpenConns(1)
 
 	// Create the schema
 	schema := `

mothership/internal/events/storage.go

@@ -16,28 +16,36 @@ const bufferSize = 1000
 
 // StorageSubscriber subscribes to EventBus and persists events to SQLite.
 // It uses a buffered queue with drop-oldest behavior to ensure it never blocks publishers.
+// Shutdown is two-phase: forwarders drain their EventBus channels first, then the worker
+// drains the internal queue. This guarantees in-flight events are not silently dropped.
 type StorageSubscriber struct {
-	db        *sql.DB
-	bus       *EventBus
-	queue     chan EventPayload
-	dropped   int64 // Counter for dropped events
-	dropWarn  int64 // Counter for when we last logged a drop warning
-	ctx       context.Context
-	cancel    context.CancelFunc
-	wg        sync.WaitGroup
-	mu        sync.Mutex
+	db              *sql.DB
+	bus             *EventBus
+	queue           chan EventPayload
+	dropped         int64 // Counter for dropped events
+	dropWarn        int64 // Counter for when we last logged a drop warning
+	workerCtx       context.Context
+	workerCancel    context.CancelFunc
+	forwarderCtx    context.Context
+	forwarderCancel context.CancelFunc
+	forwarderWg     sync.WaitGroup // tracks forwarder goroutines
+	workerWg        sync.WaitGroup // tracks worker goroutine
+	mu              sync.Mutex
 }
 
 // NewStorageSubscriber creates a new timeline storage subscriber.
 // The subscriber runs in a background goroutine until Stop() is called.
 func NewStorageSubscriber(db *sql.DB, bus *EventBus) *StorageSubscriber {
-	ctx, cancel := context.WithCancel(context.Background())
+	wCtx, wCancel := context.WithCancel(context.Background())
+	fCtx, fCancel := context.WithCancel(context.Background())
 	return &StorageSubscriber{
-		db:     db,
-		bus:    bus,
-		queue:  make(chan EventPayload, bufferSize),
-		ctx:    ctx,
-		cancel: cancel,
+		db:              db,
+		bus:             bus,
+		queue:           make(chan EventPayload, bufferSize),
+		workerCtx:       wCtx,
+		workerCancel:    wCancel,
+		forwarderCtx:    fCtx,
+		forwarderCancel: fCancel,
 	}
 }
 
@@ -70,24 +78,37 @@ func (s *StorageSubscriber) Start() {
 	// Subscribe to each event type and forward to our queue
 	for _, eventType := range eventTypes {
 		ch := s.bus.Subscribe(eventType)
-		s.wg.Add(1)
+		s.forwarderWg.Add(1)
 		go s.forwarder(ch)
 	}
 
 	// Start the storage worker
-	s.wg.Add(1)
+	s.workerWg.Add(1)
 	go s.worker()
 }
 
 // forwarder reads events from a subscriber channel and forwards them to the queue.
 // If the queue is full, it drops the oldest event (drop-oldest behavior).
+// On shutdown (forwarderCtx cancelled), it drains any remaining buffered events from
+// the EventBus channel before exiting so they are not silently lost.
 func (s *StorageSubscriber) forwarder(ch <-chan EventPayload) {
-	defer s.wg.Done()
+	defer s.forwarderWg.Done()
 
 	for {
 		select {
-		case <-s.ctx.Done():
-			return
+		case <-s.forwarderCtx.Done():
+			// Drain remaining events buffered in the EventBus channel before exiting
+			for {
+				select {
+				case payload, ok := <-ch:
+					if !ok {
+						return
+					}
+					s.enqueue(payload)
+				default:
+					return
+				}
+			}
 		case payload, ok := <-ch:
 			if !ok {
 				return
@@ -127,12 +148,12 @@ func (s *StorageSubscriber) enqueue(payload EventPayload) {
 
 // worker processes events from the queue and writes them to SQLite.
 func (s *StorageSubscriber) worker() {
-	defer s.wg.Done()
+	defer s.workerWg.Done()
 
 	for {
 		select {
-		case <-s.ctx.Done():
-			// Drain remaining events before exiting
+		case <-s.workerCtx.Done():
+			// Drain remaining events before exiting (forwarders have already finished)
 			s.drain()
 			return
 		case payload := <-s.queue:
@@ -383,13 +404,17 @@ func (s *StorageSubscriber) drain() {
 }
 
 // Stop gracefully shuts down the storage subscriber.
-// It waits for all queued events to be written to SQLite.
+// It uses two-phase shutdown: first all forwarders are stopped so they can drain
+// their EventBus channels into the internal queue, then the worker is stopped so
+// it can drain the internal queue to SQLite. This ensures no in-flight events are lost.
 func (s *StorageSubscriber) Stop() {
-	// Signal all goroutines to stop
-	s.cancel()
+	// Phase 1: stop forwarders and let them drain their EventBus channels
+	s.forwarderCancel()
+	s.forwarderWg.Wait()
 
-	// Wait for all goroutines to finish
-	s.wg.Wait()
+	// Phase 2: stop worker (it will drain the internal queue)
+	s.workerCancel()
+	s.workerWg.Wait()
 
 	log.Printf("[INFO] Timeline storage stopped (total events dropped: %d)", s.dropped)
 }

mothership/internal/events/storage_test.go

@@ -2,7 +2,6 @@
 package events
 
 import (
-	"context"
 	"sync"
 	"testing"
 	"time"
@@ -378,9 +377,6 @@ func TestStorageSubscriberQueueOverflow(t *testing.T) {
 		queue: make(chan EventPayload, bufferSize),
 	}
 
-	// Manually initialize context
-	subscriber.ctx, subscriber.cancel = context.WithCancel(context.Background())
-
 	// Fill the queue beyond capacity
 	numEvents := bufferSize + 100
 	for i := 0; i < numEvents; i++ {
@@ -406,8 +402,6 @@ func TestStorageSubscriberQueueOverflow(t *testing.T) {
 	if dropped < 100 {
 		t.Errorf("dropped count = %d, want at least 100", dropped)
 	}
-
-	subscriber.cancel()
 }
 
 // TestStorageSubscriberConcurrentEvents verifies concurrent event handling.