fix(controller): resolve mutex deadlock between agent and remotemcpserver reconcilers (#1214)

## Summary

Fixes controller reconciliation deadlock by removing the
application-level `upsertLock` mutex. Database-level concurrency control
(atomic upserts and transactions) now handles synchronization.

## Problem

The shared `upsertLock` mutex blocked all Agent reconciliations when
RemoteMCPServer reconciliation performed slow network I/O while holding
the lock.

## Solution

**Phase 1**: Reduced lock scope to exclude network I/O
**Phase 2**: Fixed Agent predicate to ensure Create events are always
processed
**Phase 3**: Removed mutex entirely - database handles concurrency via:
- Atomic upserts: `INSERT ... ON CONFLICT DO UPDATE`
- Transactions: Atomic tool replacement in `RefreshToolsForServer()`

**Phase 4**: Documentation and tests
- Added architecture doc explaining concurrency model
- Added idempotence tests for `StoreAgent` and `StoreToolServer`
- Added IMPORTANT comment to `RefreshToolsForServer` about transaction
scope

---------

Signed-off-by: Josh Bell <[email protected]>

Author: joshadambell

docs/architecture/controller-reconciliation.md

@@ -0,0 +1,90 @@
+# Controller Reconciliation Architecture
+
+This document explains how kagent's Kubernetes controllers reconcile resources and share state.
+
+## Overview
+
+The kagent controller manager runs multiple controllers that share a single `kagentReconciler` instance:
+
+```text
+┌─────────────────────────────────────────────────────────────────┐
+│                    Controller Manager                           │
+│                                                                 │
+│  ┌──────────────────┐  ┌──────────────────┐  ┌───────────────┐ │
+│  │ AgentController  │  │ RemoteMCPServer  │  │ MCPServer     │ │
+│  │                  │  │ Controller       │  │ Controller    │ │
+│  └────────┬─────────┘  └────────┬─────────┘  └───────┬───────┘ │
+│           │                     │                    │         │
+│           └─────────────────────┼────────────────────┘         │
+│                                 │                              │
+│                                 ▼                              │
+│                    ┌────────────────────────┐                  │
+│                    │   kagentReconciler     │                  │
+│                    │   (shared instance)    │                  │
+│                    │                        │                  │
+│                    │  - adkTranslator       │                  │
+│                    │  - kube client         │                  │
+│                    │  - dbClient            │                  │
+│                    └────────────────────────┘                  │
+│                                 │                              │
+│                                 ▼                              │
+│                    ┌────────────────────────┐                  │
+│                    │      SQLite DB         │                  │
+│                    └────────────────────────┘                  │
+└─────────────────────────────────────────────────────────────────┘
+```
+
+## Concurrency Model
+
+The reconciler uses database-level concurrency control instead of application-level locks:
+
+**Atomic Upserts**: Database operations for storing agents and tool servers use SQL `INSERT ... ON CONFLICT DO UPDATE` semantics. This makes the operations idempotent and safe for concurrent execution.
+
+**Transactions**: Tool refresh operations wrap multiple statements (delete all existing tools, insert new tools) in a database transaction to ensure atomicity.
+
+**No Application Locks**: The reconciler does not use mutexes or other Go synchronization primitives. SQLite handles write serialization internally.
+
+## Reconciliation Flows
+
+### Agent Reconciliation
+
+When an Agent CR is created or updated:
+
+1. The `AgentController` receives the event
+2. Delegates to the shared `kagentReconciler`
+3. The reconciler translates the Agent spec into Kubernetes manifests (Deployment, ConfigMap, etc.)
+4. Reconciles the desired state with the cluster (create/update/delete owned resources)
+5. Stores the agent configuration in the SQLite database (atomic upsert)
+6. Updates the Agent status
+
+### RemoteMCPServer Reconciliation
+
+When a RemoteMCPServer CR is created or updated:
+
+1. The RemoteMCPServer controller receives the event
+2. Stores the tool server metadata in the database (atomic upsert)
+3. Connects to the remote MCP server over the network
+4. Lists available tools from the server
+5. Replaces all tools for this server in the database (transaction)
+6. Updates the RemoteMCPServer status with discovered tools
+
+### Key Design Point
+
+Network I/O (connecting to remote MCP servers, listing tools) happens **outside** of database transactions. This prevents long-running network operations from holding database locks and blocking other reconciliations.
+
+## Event Filtering
+
+The `AgentController` uses a custom event predicate to control which Kubernetes events trigger reconciliation:
+
+- **Create events**: Always processed (ensures all agents reconcile on controller startup)
+- **Delete events**: Always processed
+- **Update events**: Only processed if the agent's generation or labels changed
+
+This filtering prevents unnecessary reconciliations when only the agent's status changes.
+
+## Related Files
+
+- [reconciler.go](../../go/internal/controller/reconciler/reconciler.go) - Shared reconciler implementation
+- [agent_controller.go](../../go/internal/controller/agent_controller.go) - Agent controller setup
+- [service.go](../../go/internal/database/service.go) - Database helpers with atomic upserts
+- [client.go](../../go/internal/database/client.go) - Database client implementation

go/internal/controller/agent_controller.go

@@ -70,7 +70,10 @@ func (r *AgentController) SetupWithManager(mgr ctrl.Manager) error {
 		WithOptions(controller.Options{
 			NeedLeaderElection: ptr.To(true),
 		}).
-		For(&v1alpha2.Agent{}, builder.WithPredicates(predicate.Or(predicate.GenerationChangedPredicate{}, predicate.LabelChangedPredicate{})))
+		// Use custom predicate that always processes Create/Delete events,
+		// but filters Update events to only process when generation or labels change.
+		// This fixes a bug where some agents were not reconciled on startup.
+		For(&v1alpha2.Agent{}, builder.WithPredicates(agentPredicate{}))
 
 	// Setup owns relationships for resources created by the Agent controller -
 	// for now ownership of agent resources is handled by the ADK translator
@@ -328,3 +331,48 @@ type typedOwnedObjectPredicate[object metav1.Object] struct {
 func (typedOwnedObjectPredicate[object]) Create(e event.TypedCreateEvent[object]) bool {
 	return false
 }
+
+// agentPredicate is a custom predicate for Agent resources that:
+// - Always processes Create events (ensures all agents are reconciled on startup)
+// - Always processes Delete events
+// - For Update events, only processes if generation or labels changed
+// This fixes a bug where GenerationChangedPredicate combined with LabelChangedPredicate
+// could inconsistently filter out Create events for some agents.
+type agentPredicate struct{}
+
+func (agentPredicate) Create(e event.CreateEvent) bool {
+	// Always process Create events - this ensures all agents are reconciled on startup
+	return true
+}
+
+func (agentPredicate) Delete(e event.DeleteEvent) bool {
+	// Always process Delete events
+	return true
+}
+
+func (agentPredicate) Update(e event.UpdateEvent) bool {
+	if e.ObjectOld == nil || e.ObjectNew == nil {
+		return false
+	}
+	// Process if generation changed (spec was modified)
+	if e.ObjectNew.GetGeneration() != e.ObjectOld.GetGeneration() {
+		return true
+	}
+	// Process if labels changed
+	oldLabels := e.ObjectOld.GetLabels()
+	newLabels := e.ObjectNew.GetLabels()
+	if len(oldLabels) != len(newLabels) {
+		return true
+	}
+	for k, v := range oldLabels {
+		if newLabels[k] != v {
+			return true
+		}
+	}
+	return false
+}
+
+func (agentPredicate) Generic(e event.GenericEvent) bool {
+	// Always process Generic events
+	return true
+}

go/internal/controller/reconciler/reconciler.go

@@ -10,7 +10,6 @@ import (
 	"reflect"
 	"slices"
 	"strings"
-	"sync"
 
 	"github.com/hashicorp/go-multierror"
 	reconcilerutils "github.com/kagent-dev/kagent/go/internal/controller/reconciler/utils"
@@ -56,9 +55,6 @@ type kagentReconciler struct {
 	dbClient database.Client
 
 	defaultModelConfig types.NamespacedName
-
-	// TODO: Remove this lock since we have a DB which we can batch anyway
-	upsertLock sync.Mutex
 }
 
 func NewKagentReconciler(
@@ -620,10 +616,6 @@ func (a *kagentReconciler) deleteObjects(ctx context.Context, objects map[types.
 }
 
 func (a *kagentReconciler) upsertAgent(ctx context.Context, agent *v1alpha2.Agent, agentOutputs *agent_translator.AgentOutputs) error {
-	// lock to prevent races
-	a.upsertLock.Lock()
-	defer a.upsertLock.Unlock()
-
 	id := utils.ConvertToPythonIdentifier(utils.GetObjectRef(agent))
 	dbAgent := &database.Agent{
 		ID:     id,
@@ -639,14 +631,12 @@ func (a *kagentReconciler) upsertAgent(ctx context.Context, agent *v1alpha2.Agen
 }
 
 func (a *kagentReconciler) upsertToolServerForRemoteMCPServer(ctx context.Context, toolServer *database.ToolServer, remoteMcpServer *v1alpha2.RemoteMCPServerSpec, namespace string) ([]*v1alpha2.MCPTool, error) {
-	// lock to prevent races
-	a.upsertLock.Lock()
-	defer a.upsertLock.Unlock()
-
+	// Store tool server - database handles concurrency via atomic upsert
 	if _, err := a.dbClient.StoreToolServer(toolServer); err != nil {
 		return nil, fmt.Errorf("failed to store toolServer %s: %v", toolServer.Name, err)
 	}
 
+	// Create transport and list tools from remote MCP server
 	tsp, err := a.createMcpTransport(ctx, remoteMcpServer, namespace)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create client for toolServer %s: %v", toolServer.Name, err)
@@ -657,6 +647,7 @@ func (a *kagentReconciler) upsertToolServerForRemoteMCPServer(ctx context.Contex
 		return nil, fmt.Errorf("failed to fetch tools for toolServer %s: %v", toolServer.Name, err)
 	}
 
+	// Refresh tools in database - uses transaction for atomicity
 	if err := a.dbClient.RefreshToolsForServer(toolServer.Name, toolServer.GroupKind, tools...); err != nil {
 		return nil, fmt.Errorf("failed to refresh tools for toolServer %s: %v", toolServer.Name, err)
 	}

go/internal/database/client.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"slices"
 	"time"
 
 	"github.com/kagent-dev/kagent/go/api/v1alpha2"
@@ -261,59 +260,36 @@ func (c *clientImpl) ListToolsForServer(serverName string, groupKind string) ([]
 		Clause{Key: "group_kind", Value: groupKind})
 }
 
-// RefreshToolsForServer refreshes a tool server
-// TODO: Use a transaction to ensure atomicity
+// RefreshToolsForServer atomically replaces all tools for a server.
+// Uses a database transaction to ensure consistency under concurrent access.
+//
+// IMPORTANT: This function should only contain fast database operations.
+// Network I/O (e.g., fetching tools from remote MCP servers) must happen
+// BEFORE calling this function, not inside it. Holding a database transaction
+// during slow operations can cause contention and degrade performance.
 func (c *clientImpl) RefreshToolsForServer(serverName string, groupKind string, tools ...*v1alpha2.MCPTool) error {
-	existingTools, err := c.ListToolsForServer(serverName, groupKind)
-	if err != nil {
-		return err
-	}
-
-	// Check if the tool exists in the existing tools
-	// If it does, update it
-	// If it doesn't, create it
-	// If it's in the existing tools but not in the new tools, delete it
-	for _, tool := range tools {
-		existingToolIndex := slices.IndexFunc(existingTools, func(t Tool) bool {
-			return t.ID == tool.Name
-		})
-		if existingToolIndex != -1 {
-			existingTool := existingTools[existingToolIndex]
-			existingTool.ServerName = serverName
-			existingTool.GroupKind = groupKind
-			existingTool.Description = tool.Description
-			err = save(c.db, &existingTool)
-			if err != nil {
-				return err
-			}
-		} else {
-			err = save(c.db, &Tool{
+	return c.db.Transaction(func(tx *gorm.DB) error {
+		// Delete all existing tools for this server in the transaction
+		if err := delete[Tool](tx,
+			Clause{Key: "server_name", Value: serverName},
+			Clause{Key: "group_kind", Value: groupKind}); err != nil {
+			return fmt.Errorf("failed to delete existing tools: %w", err)
+		}
+
+		// Insert all new tools
+		for _, tool := range tools {
+			if err := save(tx, &Tool{
 				ID:          tool.Name,
 				ServerName:  serverName,
 				GroupKind:   groupKind,
 				Description: tool.Description,
-			})
-			if err != nil {
-				return fmt.Errorf("failed to create tool %s: %v", tool.Name, err)
+			}); err != nil {
+				return fmt.Errorf("failed to create tool %s: %w", tool.Name, err)
 			}
 		}
-	}
 
-	// Delete any tools that are in the existing tools but not in the new tools
-	for _, existingTool := range existingTools {
-		if !slices.ContainsFunc(tools, func(t *v1alpha2.MCPTool) bool {
-			return t.Name == existingTool.ID
-		}) {
-			err = delete[Tool](c.db,
-				Clause{Key: "id", Value: existingTool.ID},
-				Clause{Key: "server_name", Value: serverName},
-				Clause{Key: "group_kind", Value: groupKind})
-			if err != nil {
-				return fmt.Errorf("failed to delete tool %s: %v", existingTool.ID, err)
-			}
-		}
-	}
-	return nil
+		return nil
+	})
 }
 
 // ListMessagesForRun retrieves messages for a specific run (helper method)