// Package manager owns the daemon's set of AgentTeam Server wire connections.
//
// Manager manages creation, registration, lifecycle, and lookup for running
// server connections. It is the single source of truth for the running server
// connection set.
package manager

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"reflect"
	"strings"
	"sync"

	"go.yaml.in/yaml/v3"

	"github.com/tidbcloud/AgentTeam/desktop/internal/profile"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
)

// Constructor builds a ServerConnection from a saved AgentTeam server entry and runtime context.
// Daemon code wires this to wire.New; tests inject fakes.
type Constructor func(server profile.AgentTeamServer, runtime types.ServerConnectionContext) (types.ServerConnection, error)

// ErrManagerClosed is returned when a lifecycle mutation is attempted after
// terminal shutdown admission has closed. A Manager is intentionally not
// restartable after this point: callers must construct a new Manager instead.
var ErrManagerClosed = errors.New("server connection manager is closed")

var ErrServerConnectionAlreadyRegistered = errors.New("server connection is already registered")

var ErrServerConnectionStartSuperseded = errors.New("server connection start was superseded")

var ErrInvalidServerConnectionID = errors.New("server connection ID is empty")

type startOwner struct {
	shutdownGeneration uint64
	serverID           string
	mutationGeneration uint64
	done               chan struct{}
	cancel             context.CancelFunc
	stopCallerBridge   func() bool
}

type constructionError struct {
	id  string
	err error
}

func (e *constructionError) Error() string {
	return fmt.Sprintf("create server connection %q: %v", e.id, e.err)
}

func (e *constructionError) Unwrap() error { return e.err }

type serverIdentitySnapshotter interface {
	ServerIdentitySnapshot() profile.AgentTeamServer
}

// ServerBindingApplier is implemented by running connections that can apply a
// server-resolved workspace binding without stopping their run agentteam.
type ServerBindingApplier interface {
	ApplyServerBinding(previous, current profile.AgentTeamServer) error
}

// Shutdown is the immutable terminal snapshot created by BeginShutdown. It
// owns every connection that published before the close fence and every start
// that was admitted but had not yet finished. Drain is started at most once;
// bounded callers may stop waiting while the already-started cleanup continues.
type Shutdown struct {
	drainOnce sync.Once
	done      chan struct{}
	runtimes  []types.ServerConnection
	owners    []<-chan struct{}
}

// StartDrain starts connection cleanup without waiting for it. Every runtime
// first receives the caller's exact context so a dispatcher-wide deadline is
// shared rather than replaced with per-connection budgets. If that context
// aborts a Stop before its exact run is done, the immutable drain owner keeps an
// unbounded idempotent Stop alive so a later strong waiter can upgrade rather
// than inherit a false-complete bounded attempt.
func (s *Shutdown) StartDrain(ctx context.Context) {
	if s == nil {
		return
	}
	if ctx == nil {
		ctx = context.Background()
	}
	s.drainOnce.Do(func() {
		go func() {
			var stops sync.WaitGroup
			stops.Add(len(s.runtimes))
			for _, runtime := range s.runtimes {
				runtime := runtime
				go func() {
					defer stops.Done()
					if runtime == nil {
						return
					}
					if err := runtime.Stop(ctx); err != nil && ctx.Err() != nil {
						// A bounded caller is allowed to stop waiting, but its
						// deadline must not permanently downgrade the immutable
						// shutdown owner. In particular, wire.Protocol.Stop may
						// return ctx.Err before the exact run's done channel closes.
						// Keep the already-started drain alive with an unbounded,
						// idempotent Stop so a later ordinary Dispatcher Close/Stop
						// can strong-join the same Shutdown instead of observing a
						// false completed boundary.
						_ = runtime.Stop(context.Background())
					}
				}()
			}

			// Owners that had not published at the close fence are responsible
			// for compensating-stop of anything they constructed. Do not mark the
			// shutdown drained until those owners and all published runtimes finish.
			for _, ownerDone := range s.owners {
				if ownerDone != nil {
					<-ownerDone
				}
			}
			stops.Wait()
			close(s.done)
		}()
	})
}

// Wait starts cleanup if needed and waits for either complete drainage or ctx.
// Cleanup is not canceled as a control goroutine when ctx expires: runtime Stop
// calls have already received ctx, and a context-aborted Stop is continued by
// the drain owner without extending the bounded caller.
func (s *Shutdown) Wait(ctx context.Context) error {
	if s == nil {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	s.StartDrain(ctx)
	select {
	case <-s.done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Manager manages the lifecycle of all registered server connections.
// It handles creation, registration, starting, stopping, and lookup.
// All methods are safe for concurrent use.
type Manager struct {
	mu                  sync.RWMutex
	runtimes            map[string]types.ServerConnection
	runtimeVersions     map[string]uint64
	runtimeCancels      map[string]context.CancelFunc
	runtimeConfigs      map[string]profile.AgentTeamServer
	nextRuntimeVersion  uint64
	runtimeContext      types.ServerConnectionContext
	newServerConnection Constructor
	startGeneration     uint64
	mutationGenerations map[string]uint64
	startOwners         map[*startOwner]struct{}
	shutdown            *Shutdown
}

// New creates a new Manager with the given runtime context and server
// runtime constructor.
func New(runtimeContext types.ServerConnectionContext, newServerConnection Constructor) *Manager {
	return &Manager{
		runtimes:            make(map[string]types.ServerConnection),
		runtimeVersions:     make(map[string]uint64),
		runtimeCancels:      make(map[string]context.CancelFunc),
		runtimeConfigs:      make(map[string]profile.AgentTeamServer),
		nextRuntimeVersion:  1,
		runtimeContext:      runtimeContext,
		newServerConnection: newServerConnection,
		startGeneration:     1,
		mutationGenerations: make(map[string]uint64),
		startOwners:         make(map[*startOwner]struct{}),
	}
}

func (m *Manager) createServerConnection(server profile.AgentTeamServer) (types.ServerConnection, error) {
	if m.newServerConnection == nil {
		return nil, fmt.Errorf("server connection manager has no constructor configured")
	}
	return m.newServerConnection(server, m.runtimeContext)
}

// Register adds a ServerConnection to the manager. If terminal admission is
// already closed, the connection is rejected and synchronously stopped before
// Register returns. Use TryRegister when the caller needs the rejection error.
func (m *Manager) Register(runtime types.ServerConnection) {
	_ = m.TryRegister(runtime)
}

// TryRegister is the error-reporting form of Register.
func (m *Manager) TryRegister(runtime types.ServerConnection) error {
	if m == nil {
		return ErrManagerClosed
	}
	if runtime == nil {
		return fmt.Errorf("register nil server connection")
	}
	m.mu.Lock()
	if m.shutdown != nil {
		m.mu.Unlock()
		_ = runtime.Stop(context.Background())
		return ErrManagerClosed
	}
	id := strings.TrimSpace(runtime.ID())
	if id == "" {
		m.mu.Unlock()
		_ = runtime.Stop(context.Background())
		return ErrInvalidServerConnectionID
	}
	if existing, exists := m.runtimes[id]; exists {
		if sameServerConnection(existing, runtime) {
			m.mu.Unlock()
			return nil
		}
		m.mu.Unlock()
		_ = runtime.Stop(context.Background())
		return fmt.Errorf("%w: %s", ErrServerConnectionAlreadyRegistered, id)
	}
	m.mutationGenerations[id]++
	supersededCancels := m.startCancelsForIDLocked(id)
	m.nextRuntimeVersion++
	m.runtimes[id] = runtime
	m.runtimeVersions[id] = m.nextRuntimeVersion
	delete(m.runtimeConfigs, id)
	m.mu.Unlock()
	cancelStarts(supersededCancels)
	return nil
}

// Unregister removes a ServerConnection from the manager by ID.
func (m *Manager) Unregister(id string) {
	id = strings.TrimSpace(id)
	m.mu.Lock()
	m.mutationGenerations[id]++
	supersededCancels := m.startCancelsForIDLocked(id)
	delete(m.runtimes, id)
	delete(m.runtimeVersions, id)
	delete(m.runtimeConfigs, id)
	lifetimeCancel := m.runtimeCancels[id]
	delete(m.runtimeCancels, id)
	m.mu.Unlock()
	cancelStarts(supersededCancels)
	if lifetimeCancel != nil {
		lifetimeCancel()
	}
}

// ServerConnection returns a registered ServerConnection by ID.
func (m *Manager) ServerConnection(id string) (types.ServerConnection, bool) {
	id = strings.TrimSpace(id)
	m.mu.RLock()
	defer m.mu.RUnlock()
	runtime, ok := m.runtimes[id]
	return runtime, ok
}

// WaitForServerConnection waits for one published server runtime to complete
// its first transport handshake. Callers use this only for Desktop workflows
// that must not expose a newly created local Machine before it is online.
func (m *Manager) WaitForServerConnection(ctx context.Context, id string) error {
	runtime, ok := m.ServerConnection(id)
	if !ok {
		return fmt.Errorf("server connection %q is unavailable", strings.TrimSpace(id))
	}
	waiter, ok := runtime.(interface {
		WaitConnected(context.Context) error
	})
	if !ok {
		return fmt.Errorf("server connection %q does not expose readiness", strings.TrimSpace(id))
	}
	return waiter.WaitConnected(ctx)
}

// ServerConnections returns a snapshot of all registered server connections.
func (m *Manager) ServerConnections() []types.ServerConnection {
	m.mu.RLock()
	defer m.mu.RUnlock()
	result := make([]types.ServerConnection, 0, len(m.runtimes))
	for _, runtime := range m.runtimes {
		result = append(result, runtime)
	}
	return result
}

func (m *Manager) ensureOpen() error {
	if m == nil {
		return ErrManagerClosed
	}
	m.mu.RLock()
	closed := m.shutdown != nil
	m.mu.RUnlock()
	if closed {
		return ErrManagerClosed
	}
	return nil
}

// beginMutation reserves one generation for the complete logical mutation of
// a server ID. Stop, replacement start, and failure compensation must all use
// this same generation. Allocating a new generation between phases would let
// an older request that was blocked in Stop wake after a newer request and
// incorrectly publish itself as the newest runtime.
func (m *Manager) beginMutation(serverID string) (uint64, error) {
	if m == nil {
		return 0, ErrManagerClosed
	}
	serverID = strings.TrimSpace(serverID)
	if serverID == "" {
		return 0, ErrInvalidServerConnectionID
	}

	m.mu.Lock()
	if m.shutdown != nil {
		m.mu.Unlock()
		return 0, ErrManagerClosed
	}
	m.mutationGenerations[serverID]++
	generation := m.mutationGenerations[serverID]
	supersededCancels := m.startCancelsForIDLocked(serverID)
	m.mu.Unlock()
	cancelStarts(supersededCancels)
	return generation, nil
}

func (m *Manager) beginStart(ctx context.Context, serverID string) (*startOwner, context.Context, error) {
	return m.beginStartWithExpectedGeneration(ctx, serverID, nil)
}

// beginStartWithExpectedGeneration admits either a new mutation generation or
// a conditional continuation of one exact generation. The conditional form is
// used only for Reconcile rollback: it must never allocate a newer generation
// and overwrite a runtime published by a concurrent, newer configuration
// mutation.
func (m *Manager) beginStartWithExpectedGeneration(ctx context.Context, serverID string, expectedGeneration *uint64) (*startOwner, context.Context, error) {
	if m == nil {
		return nil, nil, ErrManagerClosed
	}
	if ctx == nil {
		ctx = context.Background()
	}
	// The runtime receives a manager-owned lifetime context. Caller cancellation
	// is bridged only while startup is in flight; successful publication detaches
	// that bridge and transfers cancel ownership to the published runtime entry.
	startCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
	stopCallerBridge := context.AfterFunc(ctx, cancel)
	m.mu.Lock()
	if m.shutdown != nil {
		m.mu.Unlock()
		stopCallerBridge()
		cancel()
		return nil, nil, ErrManagerClosed
	}
	var supersededCancels []context.CancelFunc
	if expectedGeneration == nil {
		m.mutationGenerations[serverID]++
		supersededCancels = m.startCancelsForIDLocked(serverID)
	} else if m.mutationGenerations[serverID] != *expectedGeneration {
		m.mu.Unlock()
		stopCallerBridge()
		cancel()
		return nil, nil, fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, serverID)
	}
	owner := &startOwner{
		shutdownGeneration: m.startGeneration,
		serverID:           serverID,
		mutationGeneration: m.mutationGenerations[serverID],
		done:               make(chan struct{}),
		cancel:             cancel,
		stopCallerBridge:   stopCallerBridge,
	}
	m.startOwners[owner] = struct{}{}
	m.mu.Unlock()
	cancelStarts(supersededCancels)
	return owner, startCtx, nil
}

func (m *Manager) startCancelsForIDLocked(serverID string) []context.CancelFunc {
	cancels := make([]context.CancelFunc, 0)
	for owner := range m.startOwners {
		if owner.serverID == serverID && owner.cancel != nil {
			cancels = append(cancels, owner.cancel)
		}
	}
	return cancels
}

func cancelStarts(cancels []context.CancelFunc) {
	for _, cancel := range cancels {
		cancel()
	}
}

func (m *Manager) finishStart(owner *startOwner) {
	if m == nil || owner == nil {
		return
	}
	m.mu.Lock()
	delete(m.startOwners, owner)
	cancel := owner.cancel
	owner.cancel = nil
	m.mu.Unlock()
	if owner.stopCallerBridge != nil {
		owner.stopCallerBridge()
		owner.stopCallerBridge = nil
	}
	if cancel != nil {
		cancel()
	}
	close(owner.done)
}

func (owner *startOwner) detachCallerCancellation() error {
	if owner == nil || owner.stopCallerBridge == nil {
		return nil
	}
	if !owner.stopCallerBridge() {
		// The bridge callback already won and will cancel (or has canceled) the
		// manager-owned context. Do not publish a runtime whose startup caller
		// had already abandoned the operation.
		return context.Canceled
	}
	owner.stopCallerBridge = nil
	return nil
}

func (m *Manager) startAdmissionError(owner *startOwner) error {
	if m == nil || owner == nil {
		return ErrManagerClosed
	}
	m.mu.RLock()
	defer m.mu.RUnlock()
	if m.shutdown != nil || owner.shutdownGeneration != m.startGeneration {
		return ErrManagerClosed
	}
	if m.mutationGenerations[owner.serverID] != owner.mutationGeneration {
		return fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, owner.serverID)
	}
	return nil
}

func (m *Manager) publishStartedRuntime(owner *startOwner, runtime types.ServerConnection, server profile.AgentTeamServer) error {
	if owner == nil {
		if runtime != nil {
			_ = runtime.Stop(context.Background())
		}
		return ErrManagerClosed
	}
	if runtime == nil || strings.TrimSpace(runtime.ID()) != owner.serverID {
		if runtime != nil {
			_ = runtime.Stop(context.Background())
		}
		return fmt.Errorf("server connection runtime ID does not match admitted ID %q", owner.serverID)
	}
	m.mu.Lock()
	if m.shutdown != nil || owner == nil || owner.shutdownGeneration != m.startGeneration {
		m.mu.Unlock()
		_ = runtime.Stop(context.Background())
		return ErrManagerClosed
	}
	if m.mutationGenerations[owner.serverID] != owner.mutationGeneration {
		m.mu.Unlock()
		_ = runtime.Stop(context.Background())
		return fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, owner.serverID)
	}
	previous := m.runtimes[owner.serverID]
	previousCancel := m.runtimeCancels[owner.serverID]
	m.runtimes[owner.serverID] = runtime
	m.nextRuntimeVersion++
	m.runtimeVersions[owner.serverID] = m.nextRuntimeVersion
	m.runtimeCancels[owner.serverID] = owner.cancel
	m.runtimeConfigs[owner.serverID] = server
	owner.cancel = nil
	m.mu.Unlock()
	if previousCancel != nil {
		previousCancel()
	}
	if previous != nil && !sameServerConnection(previous, runtime) {
		_ = previous.Stop(context.Background())
	}
	return nil
}

// BeginShutdown atomically closes terminal admission, detaches the published
// runtime set, snapshots all admitted start owners, and cancels their start
// contexts. Repeated or concurrent calls return the same immutable snapshot.
// Call StartDrain/Wait with the desired shared shutdown context to stop it.
func (m *Manager) BeginShutdown() *Shutdown {
	if m == nil {
		return nil
	}
	m.mu.Lock()
	if existing := m.shutdown; existing != nil {
		m.mu.Unlock()
		return existing
	}

	m.startGeneration++
	runtimes := make([]types.ServerConnection, 0, len(m.runtimes))
	for _, runtime := range m.runtimes {
		runtimes = append(runtimes, runtime)
	}
	owners := make([]<-chan struct{}, 0, len(m.startOwners))
	cancels := make([]context.CancelFunc, 0, len(m.startOwners))
	for owner := range m.startOwners {
		owners = append(owners, owner.done)
		if owner.cancel != nil {
			cancels = append(cancels, owner.cancel)
		}
	}
	for _, cancel := range m.runtimeCancels {
		if cancel != nil {
			cancels = append(cancels, cancel)
		}
	}
	shutdown := &Shutdown{
		done:     make(chan struct{}),
		runtimes: runtimes,
		owners:   owners,
	}
	m.shutdown = shutdown
	// Detach immediately so no post-fence accessor can send through a runtime
	// that is already owned by shutdown.
	m.runtimes = make(map[string]types.ServerConnection)
	m.runtimeVersions = make(map[string]uint64)
	m.runtimeCancels = make(map[string]context.CancelFunc)
	m.runtimeConfigs = make(map[string]profile.AgentTeamServer)
	m.mu.Unlock()

	for _, cancel := range cancels {
		cancel()
	}
	return shutdown
}

// Start creates all enabled server connections from config, registers them, and starts them.
func (m *Manager) Start(ctx context.Context) error {
	if err := m.ensureOpen(); err != nil {
		return err
	}
	mgr := profile.Default()
	for _, server := range mgr.Servers().All() {
		if !server.Enabled {
			continue
		}

		if err := m.startConfiguredServerConnection(ctx, server); err != nil {
			if errors.Is(err, ErrManagerClosed) {
				return err
			}
			var createErr *constructionError
			if errors.As(err, &createErr) {
				slog.Error("failed to create server connection", "id", server.ID, "error", createErr.err)
				continue
			}
			return fmt.Errorf("failed to start server connection %q: %w", server.ID, err)
		}
		slog.Info("registered agentteam server connection", "id", server.ID, "agent_id", server.AgentID)
	}
	return nil
}

// Stop terminally closes admission and waits for every published runtime and
// already-admitted start under ctx. The Manager cannot be restarted afterward.
func (m *Manager) Stop(ctx context.Context) {
	shutdown := m.BeginShutdown()
	_ = shutdown.Wait(ctx)
}

// Enable creates, registers, and starts a single server connection.
// It is called when a server entry is toggled on via the dashboard API.
func (m *Manager) Enable(id string) error {
	if m == nil {
		return nil
	}
	if err := m.ensureOpen(); err != nil {
		return err
	}
	id = strings.TrimSpace(id)
	if id == "" {
		return ErrInvalidServerConnectionID
	}

	mgr := profile.Default()
	chCfg, ok := mgr.Servers().Get(id)
	if !ok {
		return fmt.Errorf("server %q not found", id)
	}
	generation, err := m.beginMutation(id)
	if err != nil {
		return err
	}
	// Enable owns the exact ID, not merely the profile flag. Retire any stray
	// runtime before publishing the configured generation.
	if err := m.stopServerConnectionForGeneration(id, generation); err != nil {
		return err
	}
	return m.startConfiguredServerConnectionForGeneration(context.Background(), chCfg, generation)
}

// Disable stops and unregisters a server connection.
func (m *Manager) Disable(id string) error {
	if m == nil {
		return nil
	}
	generation, err := m.beginMutation(id)
	if err != nil {
		return err
	}
	return m.stopServerConnectionForGeneration(id, generation)
}

// Reconcile applies a single server configuration mutation to the running
// server connection set without relying on external file watchers.
func (m *Manager) Reconcile(previous, current *profile.AgentTeamServer) error {
	return m.ReconcileContext(context.Background(), previous, current)
}

// ReconcileContext applies one mutation while honoring the caller's
// transaction deadline. Runtime Stop keeps its own immutable cleanup owner,
// so a deadline may release the profile fence without abandoning convergence.
func (m *Manager) ReconcileContext(ctx context.Context, previous, current *profile.AgentTeamServer) error {
	if ctx == nil {
		ctx = context.Background()
	}
	if m == nil {
		return nil
	}
	if err := m.ensureOpen(); err != nil {
		return err
	}
	if previous != nil && strings.TrimSpace(previous.ID) == "" {
		return ErrInvalidServerConnectionID
	}
	if current != nil && strings.TrimSpace(current.ID) == "" {
		return ErrInvalidServerConnectionID
	}

	if previous == nil && current == nil {
		return nil
	}
	prevID := ""
	if previous != nil {
		prevID = strings.TrimSpace(previous.ID)
	}
	nextID := ""
	if current != nil {
		nextID = strings.TrimSpace(current.ID)
	}
	if prevID != "" && nextID != "" && prevID != nextID {
		return fmt.Errorf("reconcile server IDs differ: %q vs %q", prevID, nextID)
	}
	serverID := nextID
	if serverID == "" {
		serverID = prevID
	}
	generation, err := m.beginMutation(serverID)
	if err != nil {
		return err
	}

	// A disabled or deleted durable snapshot owns the runtime by ID as well:
	// stop any stray connection and invalidate every older admitted start even
	// when the previous snapshot was also disabled.
	if current == nil || !current.Enabled {
		return m.stopServerConnectionForGenerationContext(ctx, serverID, generation)
	}

	if previous != nil && previous.Enabled && onlyWorkspaceBindingChanged(*previous, *current) {
		return m.applyWorkspaceBindingForGeneration(serverID, generation, *previous, *current)
	}

	// An equivalent enabled update still invalidates an older transaction. Keep
	// an already-published runtime, but repair a missing runtime from the exact
	// committed snapshot.
	if previous != nil && previous.Enabled && serverConnectionEquivalent(*previous, *current) {
		m.mu.RLock()
		_, running := m.runtimes[serverID]
		runtimeConfig, configKnown := m.runtimeConfigs[serverID]
		m.mu.RUnlock()
		if running && configKnown && serverConnectionEquivalent(runtimeConfig, *current) {
			return nil
		}
		// The profile delta is runtime-equivalent, but the currently published
		// runtime is absent or belongs to an older/unknown configuration. Repair it
		// instead of treating mere map presence as proof of convergence.
	}

	if err := m.stopServerConnectionForGenerationContext(ctx, serverID, generation); err != nil {
		return err
	}
	failedGeneration, err := m.startConfiguredServerConnectionTracked(ctx, *current, &generation)
	if err != nil {
		// A newer mutation or terminal shutdown already owns the ID. Rolling back
		// would allocate an even newer generation and overwrite that winner.
		if errors.Is(err, ErrServerConnectionStartSuperseded) || errors.Is(err, ErrManagerClosed) {
			return fmt.Errorf("restart server connection %q: %w", serverID, err)
		}
		if previous != nil && previous.Enabled {
			if rollbackErr := m.startConfiguredServerConnectionForGeneration(context.Background(), *previous, failedGeneration); rollbackErr != nil {
				return fmt.Errorf("restart server connection %q: %w (rollback failed: %v)", serverID, err, rollbackErr)
			}
		}
		return fmt.Errorf("restart server connection %q: %w", serverID, err)
	}
	return nil
}

// PrepareDelete closes the current runtime generation for one exact profile
// RepairServer stops any residual runtime for server.ID and starts a fresh
// connection from the supplied snapshot. Used by the dashboard repair path to
// recover from a server-side identity change (workspace rebind, agent re-issue)
// without taking the broader Manager through a full Reconcile cycle. The
// caller-supplied snapshot is authoritative: profile.Default is intentionally
// not consulted so a concurrent dashboard edit cannot flip the replacement
// back to the older identity mid-repair.
func (m *Manager) RepairServer(ctx context.Context, server profile.AgentTeamServer) error {
	if m == nil {
		return nil
	}
	if err := m.ensureOpen(); err != nil {
		return err
	}
	serverID := strings.TrimSpace(server.ID)
	if serverID == "" {
		return ErrInvalidServerConnectionID
	}
	generation, err := m.beginMutation(serverID)
	if err != nil {
		return err
	}
	if err := m.stopServerConnectionForGenerationContext(ctx, serverID, generation); err != nil {
		return err
	}
	if !server.Enabled {
		return nil
	}
	if err := m.startConfiguredServerConnectionForGeneration(ctx, server, generation); err != nil {
		return err
	}
	runtime, ok := m.runtimeForGeneration(serverID, generation)
	if !ok {
		return fmt.Errorf("replacement server connection %q changed concurrently", serverID)
	}
	snapshotter, ok := runtime.(serverIdentitySnapshotter)
	if !ok || !serverRuntimeIdentityMatches(snapshotter.ServerIdentitySnapshot(), server) {
		verificationErr := fmt.Errorf("replacement server connection %q identity verification failed", serverID)
		if cleanupErr := runtime.Stop(ctx); cleanupErr != nil {
			return fmt.Errorf("%w (cleanup failed: %v; residual runtime retained)", verificationErr, cleanupErr)
		}
		m.removeRuntimeForGeneration(serverID, generation, runtime)
		return verificationErr
	}
	if _, ok := m.runtimeForGeneration(serverID, generation); !ok {
		if cleanupErr := runtime.Stop(ctx); cleanupErr != nil {
			return fmt.Errorf("replacement server connection %q changed after identity verification (cleanup failed: %v)", serverID, cleanupErr)
		}
		return fmt.Errorf("replacement server connection %q changed after identity verification", serverID)
	}
	return nil
}

func (m *Manager) runtimeForGeneration(id string, generation uint64) (types.ServerConnection, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	if m.shutdown != nil || m.mutationGenerations[id] != generation {
		return nil, false
	}
	runtime, ok := m.runtimes[id]
	return runtime, ok
}

func (m *Manager) applyWorkspaceBindingForGeneration(
	id string,
	generation uint64,
	previous, current profile.AgentTeamServer,
) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.shutdown != nil || m.mutationGenerations[id] != generation {
		return fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, id)
	}
	runtime, running := m.runtimes[id]
	runtimeConfig, configKnown := m.runtimeConfigs[id]

	if !running {
		return fmt.Errorf("server connection %q not found", id)
	}
	if !configKnown || !serverConnectionEquivalent(runtimeConfig, previous) {
		return fmt.Errorf("server connection %q runtime configuration changed concurrently", id)
	}
	applier, ok := runtime.(ServerBindingApplier)
	if !ok {
		return fmt.Errorf("server connection %q cannot apply a live binding", id)
	}
	if err := applier.ApplyServerBinding(previous, current); err != nil {
		return err
	}
	m.runtimeConfigs[id] = current
	return nil
}

func (m *Manager) removeRuntimeForGeneration(
	id string,
	generation uint64,
	runtime types.ServerConnection,
) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.shutdown != nil || m.mutationGenerations[id] != generation {
		return
	}
	if current, ok := m.runtimes[id]; !ok || !sameServerConnection(current, runtime) {
		return
	}
	delete(m.runtimes, id)
	delete(m.runtimeVersions, id)
	if cancel := m.runtimeCancels[id]; cancel != nil {
		cancel()
	}
	delete(m.runtimeCancels, id)
	delete(m.runtimeConfigs, id)
}

// PrepareDelete returns a compensation closure that re-publishes a server
// runtime snapshot if the calling dashboard delete transaction fails to commit
// to the durable profile. The runtime is captured as it was before the
// snapshot and returns a generation-bound compensation. The caller invokes the
// compensation only when the durable profile delete failed while that same
// snapshot remained current. A concurrent enable/reconcile/disable advances
// the per-ID mutation generation, so an older delete failure can never restore
// its stale runtime over the newer winner.
func (m *Manager) PrepareDelete(previous *profile.AgentTeamServer) (func() error, error) {
	if m == nil {
		return func() error { return nil }, nil
	}
	if err := m.ensureOpen(); err != nil {
		return nil, err
	}
	if previous == nil {
		return nil, fmt.Errorf("previous server configuration is required")
	}
	if strings.TrimSpace(previous.ID) == "" {
		return nil, ErrInvalidServerConnectionID
	}

	generation, err := m.beginMutation(previous.ID)
	if err != nil {
		return nil, err
	}
	if err := m.stopServerConnectionForGeneration(previous.ID, generation); err != nil {
		return nil, err
	}
	if !previous.Enabled {
		return func() error { return nil }, nil
	}
	snapshot := *previous
	var once sync.Once
	var rollbackErr error
	return func() error {
		once.Do(func() {
			rollbackErr = m.startConfiguredServerConnectionForGeneration(context.Background(), snapshot, generation)
		})
		return rollbackErr
	}, nil
}

func (m *Manager) startConfiguredServerConnection(ctx context.Context, server profile.AgentTeamServer) error {
	_, err := m.startConfiguredServerConnectionTracked(ctx, server, nil)
	return err
}

func (m *Manager) startConfiguredServerConnectionForGeneration(ctx context.Context, server profile.AgentTeamServer, expectedGeneration uint64) error {
	_, err := m.startConfiguredServerConnectionTracked(ctx, server, &expectedGeneration)
	return err
}

func (m *Manager) startConfiguredServerConnectionTracked(ctx context.Context, server profile.AgentTeamServer, expectedGeneration *uint64) (uint64, error) {
	serverID := strings.TrimSpace(server.ID)
	if serverID == "" {
		return 0, ErrInvalidServerConnectionID
	}
	owner, startCtx, err := m.beginStartWithExpectedGeneration(ctx, serverID, expectedGeneration)
	if err != nil {
		return 0, err
	}
	defer m.finishStart(owner)
	generation := owner.mutationGeneration

	runtime, err := m.createServerConnection(server)
	if err != nil {
		if admissionErr := m.startAdmissionError(owner); admissionErr != nil {
			return generation, fmt.Errorf("%w: create server connection %q: %v", admissionErr, server.ID, err)
		}
		return generation, &constructionError{id: server.ID, err: err}
	}
	if runtime == nil {
		return generation, &constructionError{id: server.ID, err: fmt.Errorf("constructor returned nil runtime")}
	}
	if admissionErr := m.startAdmissionError(owner); admissionErr != nil {
		return generation, m.cleanupUnpublishedRuntime(ctx, owner, runtime, server, admissionErr)
	}

	if err := runtime.Start(startCtx); err != nil {
		startErr := fmt.Errorf("start server connection %q: %w", server.ID, err)
		if admissionErr := m.startAdmissionError(owner); admissionErr != nil {
			startErr = fmt.Errorf("%w: start server connection %q: %v", admissionErr, server.ID, err)
		}
		return generation, m.cleanupUnpublishedRuntime(ctx, owner, runtime, server, startErr)
	}
	if err := owner.detachCallerCancellation(); err != nil {
		if admissionErr := m.startAdmissionError(owner); admissionErr != nil {
			err = admissionErr
		}
		return generation, m.cleanupUnpublishedRuntime(
			ctx,
			owner, runtime, server,
			fmt.Errorf("start server connection %q: %w", server.ID, err),
		)
	}
	if err := startCtx.Err(); err != nil {
		if admissionErr := m.startAdmissionError(owner); admissionErr != nil {
			err = admissionErr
		}
		return generation, m.cleanupUnpublishedRuntime(
			ctx,
			owner, runtime, server,
			fmt.Errorf("start server connection %q: %w", server.ID, err),
		)
	}
	if err := m.publishStartedRuntime(owner, runtime, server); err != nil {
		return generation, err
	}
	slog.Info("enabled server connection", "id", server.ID, "agent_id", server.AgentID)

	return generation, nil
}

func (m *Manager) cleanupUnpublishedRuntime(
	ctx context.Context,
	owner *startOwner,
	runtime types.ServerConnection,
	server profile.AgentTeamServer,
	cause error,
) error {
	if runtime == nil {
		return cause
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if cleanupErr := runtime.Stop(ctx); cleanupErr == nil {
		return cause
	} else if retainErr := m.retainResidualRuntime(owner, runtime, server); retainErr != nil {
		go func() { _ = runtime.Stop(context.Background()) }()
		return fmt.Errorf("%w (cleanup failed: %v; residual ownership failed: %v)", cause, cleanupErr, retainErr)
	} else {
		return fmt.Errorf("%w (cleanup failed: %v; residual runtime retained)", cause, cleanupErr)
	}
}

func (m *Manager) retainResidualRuntime(
	owner *startOwner,
	runtime types.ServerConnection,
	server profile.AgentTeamServer,
) error {
	if m == nil || owner == nil || runtime == nil {
		return ErrManagerClosed
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.shutdown != nil || owner.shutdownGeneration != m.startGeneration {
		return ErrManagerClosed
	}
	if m.mutationGenerations[owner.serverID] != owner.mutationGeneration {
		return fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, owner.serverID)
	}
	if existing := m.runtimes[owner.serverID]; existing != nil && !sameServerConnection(existing, runtime) {
		return fmt.Errorf("%w: %s", ErrServerConnectionAlreadyRegistered, owner.serverID)
	}
	m.runtimes[owner.serverID] = runtime
	m.nextRuntimeVersion++
	m.runtimeVersions[owner.serverID] = m.nextRuntimeVersion
	m.runtimeCancels[owner.serverID] = owner.cancel
	m.runtimeConfigs[owner.serverID] = server
	owner.cancel = nil
	return nil
}

func (m *Manager) stopServerConnection(id string) error {
	_, err := m.stopServerConnectionTracked(id)
	return err
}

func (m *Manager) stopServerConnectionTracked(id string) (uint64, error) {
	if m == nil {
		return 0, nil
	}
	id = strings.TrimSpace(id)
	if id == "" {
		return 0, ErrInvalidServerConnectionID
	}
	generation, err := m.beginMutation(id)
	if err != nil {
		return 0, err
	}
	return generation, m.stopServerConnectionForGeneration(id, generation)
}

func (m *Manager) stopServerConnectionForGeneration(id string, generation uint64) error {
	return m.stopServerConnectionForGenerationContext(context.Background(), id, generation)
}

func (m *Manager) stopServerConnectionForGenerationContext(ctx context.Context, id string, generation uint64) error {
	if m == nil {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	id = strings.TrimSpace(id)
	if id == "" {
		return ErrInvalidServerConnectionID
	}
	m.mu.Lock()
	if m.shutdown != nil {
		m.mu.Unlock()
		return ErrManagerClosed
	}
	if m.mutationGenerations[id] != generation {
		m.mu.Unlock()
		return fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, id)
	}
	runtime, ok := m.runtimes[id]
	version := m.runtimeVersions[id]
	lifetimeCancel := m.runtimeCancels[id]
	m.mu.Unlock()
	if lifetimeCancel != nil {
		lifetimeCancel()
	}
	if !ok {
		return nil
	}

	// Runtime replacement is an exact generation boundary. A bounded Stop that
	// returns before its internal run has drained would leave an unowned old
	// connection alive while a replacement starts, so mutation paths strong-join
	// the runtime before publishing the next generation. Dispatcher shutdown has
	// a separate bounded-wait API whose immutable Shutdown owner continues the
	// strong join in the background.
	if err := runtime.Stop(ctx); err != nil {
		if ctx.Err() != nil {
			go m.finishTimedOutRuntimeStop(runtime, id, generation, version)
		}
		return fmt.Errorf("stop server connection %q: %w", id, err)
	}
	m.mu.Lock()
	if m.shutdown != nil {
		m.mu.Unlock()
		return ErrManagerClosed
	}
	if m.mutationGenerations[id] != generation {
		m.mu.Unlock()
		return fmt.Errorf("%w: %s", ErrServerConnectionStartSuperseded, id)
	}
	if _, exists := m.runtimes[id]; exists && m.runtimeVersions[id] == version {
		delete(m.runtimes, id)
		delete(m.runtimeVersions, id)
		delete(m.runtimeCancels, id)
		delete(m.runtimeConfigs, id)
	}
	m.mu.Unlock()
	slog.Info("disabled server connection", "id", id)
	return nil
}

func (m *Manager) finishTimedOutRuntimeStop(
	runtime types.ServerConnection,
	id string,
	generation, version uint64,
) {
	if runtime == nil || runtime.Stop(context.Background()) != nil {
		return
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.shutdown != nil || m.mutationGenerations[id] != generation {
		return
	}
	if current, exists := m.runtimes[id]; exists &&
		sameServerConnection(current, runtime) &&
		m.runtimeVersions[id] == version {
		delete(m.runtimes, id)
		delete(m.runtimeVersions, id)
		delete(m.runtimeCancels, id)
		delete(m.runtimeConfigs, id)
	}
}

func sameServerConnection(left, right types.ServerConnection) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	lv := reflect.ValueOf(left)
	rv := reflect.ValueOf(right)
	if lv.Type() != rv.Type() {
		return false
	}
	switch lv.Kind() {
	case reflect.Chan, reflect.Func, reflect.Map, reflect.Pointer, reflect.Slice, reflect.UnsafePointer:
		return lv.Pointer() == rv.Pointer()
	default:
		return lv.Comparable() && lv.Interface() == rv.Interface()
	}
}

func serverConnectionEquivalent(previous, current profile.AgentTeamServer) bool {
	if strings.TrimSpace(previous.AgentID) != strings.TrimSpace(current.AgentID) {
		return false
	}
	if !serverConnectionConfigEquivalent(previous.Config, current.Config) {
		return false
	}
	return strings.TrimSpace(previous.ServerURL) == strings.TrimSpace(current.ServerURL) &&
		strings.TrimSpace(previous.APIKey) == strings.TrimSpace(current.APIKey) &&
		strings.TrimSpace(previous.WorkspaceID) == strings.TrimSpace(current.WorkspaceID) &&
		strings.TrimSpace(previous.MachineID) == strings.TrimSpace(current.MachineID) &&
		strings.TrimSpace(previous.AuthToken) == strings.TrimSpace(current.AuthToken)
}

func onlyWorkspaceBindingChanged(previous, current profile.AgentTeamServer) bool {
	if strings.TrimSpace(previous.WorkspaceID) == strings.TrimSpace(current.WorkspaceID) {
		return false
	}
	previous.WorkspaceID = current.WorkspaceID
	previous.Config = current.Config
	return serverConnectionEquivalent(previous, current)
}

func serverRuntimeIdentityMatches(actual, expected profile.AgentTeamServer) bool {
	if strings.TrimSpace(actual.ID) != strings.TrimSpace(expected.ID) ||
		actual.Enabled != expected.Enabled ||
		strings.TrimSpace(actual.ServerURL) != strings.TrimSpace(expected.ServerURL) ||
		strings.TrimSpace(actual.APIKey) != strings.TrimSpace(expected.APIKey) ||
		strings.TrimSpace(actual.WorkspaceID) != strings.TrimSpace(expected.WorkspaceID) ||
		strings.TrimSpace(actual.MachineID) != strings.TrimSpace(expected.MachineID) ||
		strings.TrimSpace(actual.AgentID) != strings.TrimSpace(expected.AgentID) {
		return false
	}
	actualToken := strings.TrimSpace(actual.AuthToken)
	if actualToken == "" {
		actualToken = strings.TrimSpace(actual.APIKey)
	}
	expectedToken := strings.TrimSpace(expected.AuthToken)
	if expectedToken == "" {
		expectedToken = strings.TrimSpace(expected.APIKey)
	}
	return actualToken == expectedToken
}

func serverConnectionConfigEquivalent(previous, current types.RawMessage) bool {
	if len(previous) == 0 && len(current) == 0 {
		return true
	}

	var previousValue any
	if err := yaml.Unmarshal(previous, &previousValue); err != nil {
		return string(previous) == string(current)
	}

	var currentValue any
	if err := yaml.Unmarshal(current, &currentValue); err != nil {
		return string(previous) == string(current)
	}

	return reflect.DeepEqual(previousValue, currentValue)
}