Refactor queue checks, abort text, and streaming errors

Move pendingQueueHasSourceEvent into pending_queue.go and implement proper locking (pendingQueuesMu and queue.mu) to safely inspect queue items. Simplify drainPendingQueue to delete the queue map entry and return its items directly. Remove the duplicate helper from abort_helpers.go. Improve formatAbortNotice by capitalizing each sentence part and joining them with ". " for clearer messages. Remove redundant run field assignments in roomRunTarget. Adjust finishStreamingWithFailure to fall through from "cancelled" to "stop" so cancelled streams call End like stop cases and remove some redundant nil checks. These changes tidy concurrency handling, clarify abort messaging, and simplify streaming error handling.

Author: batuhan

bridges/ai/abort_helpers.go

@@ -75,31 +75,15 @@ func formatAbortNotice(result userStopResult) string {
 		if len(parts) == 0 {
 			return "No active or queued turns to stop."
 		}
-		suffix := ""
-		if len(parts) > 1 {
-			suffix = " " + strings.Join(parts[1:], ". ") + "."
+		for i := range parts {
+			parts[i] = strings.ToUpper(parts[i][:1]) + parts[i][1:]
 		}
-		return strings.ToUpper(parts[0][:1]) + parts[0][1:] + "." + suffix
+		return strings.Join(parts, ". ") + "."
 	default:
 		return "No active or queued turns to stop."
 	}
 }
 
-func (oc *AIClient) pendingQueueHasSourceEvent(roomID id.RoomID, sourceEventID id.EventID) bool {
-	if oc == nil || roomID == "" || sourceEventID == "" {
-		return false
-	}
-	queue := oc.getQueueSnapshot(roomID)
-	if queue == nil {
-		return false
-	}
-	for _, item := range queue.items {
-		if item.pending.sourceEventID() == sourceEventID {
-			return true
-		}
-	}
-	return false
-}
 
 func buildStopMetadata(plan userStopPlan, req userStopRequest) *assistantStopMetadata {
 	return &assistantStopMetadata{

bridges/ai/pending_queue.go

@@ -106,20 +106,36 @@ func (oc *AIClient) drainPendingQueue(roomID id.RoomID) []pendingQueueItem {
 		return nil
 	}
 	delete(oc.pendingQueues, roomID)
+	items := queue.items
 	oc.pendingQueuesMu.Unlock()
 
-	queue.mu.Lock()
-	items := slices.Clone(queue.items)
-	queue.items = nil
-	queue.summaryLines = nil
-	queue.droppedCount = 0
-	queue.lastItem = nil
-	queue.mu.Unlock()
-
 	oc.stopQueueTyping(roomID)
 	return items
 }
 
+func (oc *AIClient) pendingQueueHasSourceEvent(roomID id.RoomID, sourceEventID id.EventID) bool {
+	if oc == nil || roomID == "" || sourceEventID == "" {
+		return false
+	}
+	oc.pendingQueuesMu.Lock()
+	queue := oc.pendingQueues[roomID]
+	if queue == nil {
+		oc.pendingQueuesMu.Unlock()
+		return false
+	}
+	queue.mu.Lock()
+	found := false
+	for _, item := range queue.items {
+		if item.pending.sourceEventID() == sourceEventID {
+			found = true
+			break
+		}
+	}
+	queue.mu.Unlock()
+	oc.pendingQueuesMu.Unlock()
+	return found
+}
+
 func (oc *AIClient) removePendingQueueBySourceEvent(roomID id.RoomID, sourceEventID id.EventID) []pendingQueueItem {
 	if oc == nil || roomID == "" || sourceEventID == "" {
 		return nil

bridges/ai/room_runs.go

@@ -137,9 +137,6 @@ func (oc *AIClient) roomRunTarget(roomID id.RoomID) (turnID string, sourceEventI
 	turnID = state.turn.ID()
 	sourceEventID = state.sourceEventID()
 	initialEventID = state.turn.InitialEventID()
-	run.turnID = turnID
-	run.sourceEvent = sourceEventID
-	run.initialEvent = initialEventID
 	return turnID, sourceEventID, initialEventID, state
 }
 

bridges/ai/streaming_error_handling.go

@@ -53,15 +53,13 @@ func (oc *AIClient) finishStreamingWithFailure(
 	switch reason {
 	case "cancelled":
 		state.writer().Abort(ctx, "cancelled")
-		if state != nil && state.turn != nil {
-			state.turn.End(msgconv.MapFinishReason(reason))
-		}
+		fallthrough
 	case "stop":
-		if state != nil && state.turn != nil {
+		if state.turn != nil {
 			state.turn.End(msgconv.MapFinishReason(reason))
 		}
 	default:
-		if state != nil && state.turn != nil {
+		if state.turn != nil {
 			state.turn.EndWithError(err.Error())
 		}
 	}