Fix pending queue locking and lastItem updates

Rework pending queue locking and item housekeeping to avoid races and stale pointers. getPendingQueue now locks the queue.mu before releasing pendingQueuesMu and consistently applies settings for both new and existing queues. drainPendingQueue clears queue.items and lastItem when removing a queue. removePendingQueueBySourceEvent reassigns lastItem to the new tail if the removed item was the last. enqueuePendingItem removed a now-redundant explicit lock (the returned queue is already locked). Added a unit test to verify lastItem is cleared/reassigned, and tightened an error assertion in the streaming test to use errors.Is for wrapped cancellations.

Author: batuhan

bridges/ai/pending_queue.go

@@ -57,7 +57,6 @@ type pendingQueueDispatchCandidate struct {
 
 func (oc *AIClient) getPendingQueue(roomID id.RoomID, settings airuntime.QueueSettings) *pendingQueue {
 	oc.pendingQueuesMu.Lock()
-	defer oc.pendingQueuesMu.Unlock()
 	queue := oc.pendingQueues[roomID]
 	if queue == nil {
 		queue = &pendingQueue{
@@ -68,20 +67,19 @@ func (oc *AIClient) getPendingQueue(roomID id.RoomID, settings airuntime.QueueSe
 			dropPolicy: settings.DropPolicy,
 		}
 		oc.pendingQueues[roomID] = queue
-	} else {
-		queue.mu.Lock()
-		queue.mode = settings.Mode
-		if settings.DebounceMs >= 0 {
-			queue.debounceMs = settings.DebounceMs
-		}
-		if settings.Cap > 0 {
-			queue.cap = settings.Cap
-		}
-		if settings.DropPolicy != "" {
-			queue.dropPolicy = settings.DropPolicy
-		}
-		queue.mu.Unlock()
 	}
+	queue.mu.Lock()
+	queue.mode = settings.Mode
+	if settings.DebounceMs >= 0 {
+		queue.debounceMs = settings.DebounceMs
+	}
+	if settings.Cap > 0 {
+		queue.cap = settings.Cap
+	}
+	if settings.DropPolicy != "" {
+		queue.dropPolicy = settings.DropPolicy
+	}
+	oc.pendingQueuesMu.Unlock()
 	return queue
 }
 
@@ -99,9 +97,11 @@ func (oc *AIClient) drainPendingQueue(roomID id.RoomID) []pendingQueueItem {
 		oc.pendingQueuesMu.Unlock()
 		return nil
 	}
-	delete(oc.pendingQueues, roomID)
 	queue.mu.Lock()
+	delete(oc.pendingQueues, roomID)
 	items := queue.items
+	queue.items = nil
+	queue.lastItem = nil
 	queue.mu.Unlock()
 	oc.pendingQueuesMu.Unlock()
 
@@ -131,6 +131,13 @@ func (oc *AIClient) removePendingQueueBySourceEvent(roomID id.RoomID, sourceEven
 	}
 	clear(queue.items[len(kept):])
 	queue.items = kept
+	if queue.lastItem != nil && queue.lastItem.pending.sourceEventID() == sourceEventID {
+		queue.lastItem = nil
+		if len(kept) > 0 {
+			lastItem := kept[len(kept)-1]
+			queue.lastItem = &lastItem
+		}
+	}
 	empty := len(queue.items) == 0 && queue.droppedCount == 0
 	if empty {
 		delete(oc.pendingQueues, roomID)
@@ -149,7 +156,6 @@ func (oc *AIClient) enqueuePendingItem(roomID id.RoomID, item pendingQueueItem,
 	if queue == nil {
 		return false
 	}
-	queue.mu.Lock()
 	defer queue.mu.Unlock()
 
 	for _, existing := range queue.items {

bridges/ai/queue_status_test.go

@@ -190,3 +190,33 @@ func TestDispatchOrQueueQueuesBehindExistingPendingWork(t *testing.T) {
 		t.Fatalf("expected room to remain unacquired while backlog exists")
 	}
 }
+
+func TestRemovePendingQueueBySourceEventClearsRemovedLastItem(t *testing.T) {
+	roomID := id.RoomID("!room:example.com")
+	first := pendingQueueItem{pending: pendingMessage{SourceEventID: id.EventID("$one")}}
+	last := pendingQueueItem{pending: pendingMessage{SourceEventID: id.EventID("$two")}}
+	oc := &AIClient{
+		pendingQueues: map[id.RoomID]*pendingQueue{
+			roomID: {
+				items:    []pendingQueueItem{first, last},
+				lastItem: &last,
+			},
+		},
+	}
+
+	removed := oc.removePendingQueueBySourceEvent(roomID, id.EventID("$two"))
+	if len(removed) != 1 {
+		t.Fatalf("expected one removed item, got %d", len(removed))
+	}
+
+	snapshot := oc.getQueueSnapshot(roomID)
+	if snapshot == nil {
+		t.Fatal("expected queue snapshot to remain")
+	}
+	if snapshot.lastItem == nil {
+		t.Fatal("expected lastItem to be reassigned to the new tail")
+	}
+	if got := snapshot.lastItem.pending.sourceEventID(); got != id.EventID("$one") {
+		t.Fatalf("expected lastItem to point at remaining item, got %q", got)
+	}
+}

bridges/ai/streaming_error_handling_test.go

@@ -101,8 +101,8 @@ func TestFinishStreamingWithFailureCancelledEndsTurnAsCancelled(t *testing.T) {
 		"cancelled",
 		context.Canceled,
 	)
-	if err == nil {
-		t.Fatal("expected wrapped cancellation error")
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected wrapped cancellation error, got %#v", err)
 	}
 
 	message := streamui.SnapshotUIMessage(state.turn.UIState())