package home

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand/v2"
	"net"
	"net/http"
	"net/url"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
	"github.com/redis/go-redis/v9/maintnotifications"
	"github.com/router-for-me/CLIProxyAPI/v8/internal/config"
	"github.com/router-for-me/CLIProxyAPI/v8/internal/kvstore"
	"github.com/router-for-me/CLIProxyAPI/v8/sdk/pluginstore"
	log "github.com/sirupsen/logrus"
)

const (
	redisKeyConfig             = "config"
	redisChannelConfig         = "config"
	redisKeyUsage              = "usage"
	redisKeyInFlightSnapshot   = "in-flight-snapshot"
	redisKeyConcurrencyRelease = "concurrency-release"
	redisKeyRequestLog         = "request-log"
	redisKeyAppLog             = "app-log"
	redisKeyPluginStatus       = "plugin-status"
	redisKeyPluginTasks        = "plugin-tasks"
	redisKeyPluginSync         = "plugin-sync"

	homeReconnectInterval                     = time.Second
	homeReconnectFailoverThreshold            = 3
	homeRedisOperationTimeout                 = 3 * time.Second
	homeRefreshOperationTimeout               = 35 * time.Second
	homePluginSyncOperationTimeout            = 2 * time.Minute
	homeSubscriptionReceiveTimeout            = 3 * time.Second
	credentialConcurrencyNodeHeartbeatTimeout = 20 * time.Second
	redisChannelCluster                       = "cluster"
)

const pluginSyncUnsupportedErrorType = "plugin_sync_unsupported"

// DispatchError classifies whether Home may have processed an auth dispatch request.
type DispatchError struct {
	Err       error
	Ambiguous bool
}

func (e *DispatchError) Error() string {
	if e == nil || e.Err == nil {
		return "home auth dispatch failed"
	}
	return e.Err.Error()
}

func (e *DispatchError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Err
}

// NewAmbiguousDispatchError marks a post-send transport failure as requiring a client abort.
func NewAmbiguousDispatchError(err error) error {
	if err == nil {
		return nil
	}
	return &DispatchError{Err: err, Ambiguous: true}
}

// IsAmbiguousDispatchError reports whether Home may have processed the dispatch request.
func IsAmbiguousDispatchError(err error) bool {
	var dispatchErr *DispatchError
	return errors.As(err, &dispatchErr) && dispatchErr.Ambiguous
}

var errClusterDiscoveryTransport = errors.New("home cluster discovery transport failed")

var (
	ErrDisabled              = errors.New("home client disabled")
	ErrNotConnected          = errors.New("home not connected")
	ErrEmptyResponse         = errors.New("home returned empty response")
	ErrAuthNotFound          = errors.New("home auth not found")
	ErrConfigNotFound        = errors.New("home config not found")
	ErrModelsNotFound        = errors.New("home models not found")
	ErrPluginSyncUnsupported = errors.New("home plugin sync is unsupported")
	ErrDispatchFenced        = errors.New("home auth dispatch is fenced")
	// ErrCompareAndSwapUnsupported reports that this Home predates the CAS command.
	ErrCompareAndSwapUnsupported = kvstore.ErrCompareAndSwapUnsupported
)

// isHomeCommandUnsupported reports whether Home rejected a command it does not
// implement. It mirrors isHomeAppLogUnsupported in internal/logging; the two are
// kept separate so the packages stay decoupled.
func isHomeCommandUnsupported(err error) bool {
	for err != nil {
		message := strings.ToLower(strings.TrimSpace(err.Error()))
		if strings.Contains(message, "unknown command") || strings.Contains(message, "unsupported command") {
			return true
		}
		err = errors.Unwrap(err)
	}
	return false
}

// IsMembershipTakeoverUnavailableError reports whether Home cannot preserve the previous membership state.
func IsMembershipTakeoverUnavailableError(err error) bool {
	if err == nil {
		return false
	}
	message := strings.TrimSpace(strings.ToLower(err.Error()))
	return message == "membership_takeover_unavailable" || message == "err membership_takeover_unavailable"
}

// IsLegacyMembershipProtocolError reports whether Home rejected the secure subscription argument count.
func IsLegacyMembershipProtocolError(err error) bool {
	if err == nil {
		return false
	}
	message := strings.TrimSpace(strings.ToLower(err.Error()))
	return message == "wrong number of arguments for 'subscribe' command" || message == "err wrong number of arguments for 'subscribe' command"
}

type clusterNode struct {
	IP          string    `json:"ip"`
	Port        int       `json:"port"`
	ClientCount int       `json:"client_count"`
	IsMaster    bool      `json:"is_master"`
	LastSeenAt  time.Time `json:"last_seen_at"`
}

type clusterNodesEnvelope struct {
	OK    bool          `json:"ok"`
	Nodes []clusterNode `json:"nodes"`
}

type PluginTask struct {
	ID             uint      `json:"id"`
	Operation      string    `json:"operation"`
	PluginID       string    `json:"plugin_id"`
	TargetNodeType string    `json:"target_node_type,omitempty"`
	TargetNodeID   string    `json:"target_node_id,omitempty"`
	CreatedAt      time.Time `json:"created_at"`
	UpdatedAt      time.Time `json:"updated_at"`
}

type KVSetOptions = kvstore.KVSetOptions

type subscriptionCloser interface {
	Close() error
}

type recoveryState uint32

const (
	recoveryStateStable recoveryState = iota
	recoveryStateTakeoverEligible
	recoveryStateSwitching
	recoveryStateSwitchingTakeover
)

type Client struct {
	mu sync.Mutex

	homeCfg  config.HomeConfig
	seedHost string
	seedPort int

	cmd         *redis.Client
	cmdOptions  *redis.Options
	sub         *redis.Client
	release     *redis.Client
	connections map[*homeDispatchConn]struct{}
	closing     chan struct{}
	lifecycle   config.CredentialConcurrencyConfig
	limiter     atomic.Pointer[config.CredentialConcurrencyConfig]
	managed     bool
	// releaseReady belongs to this acknowledged membership and is never copied to a replacement.
	releaseReady bool

	heartbeatOK       atomic.Bool
	dispatchFenced    atomic.Bool
	releaseFenced     atomic.Bool
	ambiguousDispatch atomic.Bool
	// casUnsupported latches when Home does not implement the CAS command.
	// It is deliberately NOT carried across NewLifetime: CAS support is a
	// property of the Home deployment, so re-probing once per client lifetime
	// lets a Home upgrade take effect on the next reconnect instead of
	// requiring a CPA restart. The probe costs one round trip that returns an
	// error without performing any write.
	casUnsupported       atomic.Bool
	testOperationTimeout time.Duration
	recoveryState        atomic.Uint32
	instanceID           string
	legacyMembership     bool
	clusterNodes         []clusterNode
	reconnectFailures    int
	balance              clusterBalanceState
	latency              *clusterLatencyHistory
}

func New(homeCfg config.HomeConfig) *Client {
	return &Client{
		homeCfg:    homeCfg,
		seedHost:   strings.TrimSpace(homeCfg.Host),
		seedPort:   homeCfg.Port,
		instanceID: uuid.NewString(),
	}
}

// NewLifetime creates a fresh client while preserving cluster failover state.
func (c *Client) NewLifetime() *Client {
	if c == nil {
		return nil
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	next := &Client{
		homeCfg:              c.homeCfg,
		seedHost:             c.seedHost,
		seedPort:             c.seedPort,
		clusterNodes:         append([]clusterNode(nil), c.clusterNodes...),
		reconnectFailures:    c.reconnectFailures,
		testOperationTimeout: c.testOperationTimeout,
		instanceID:           c.instanceID,
		legacyMembership:     c.legacyMembership,
		balance:              c.balance,
		latency:              c.latency,
	}
	next.balance.registered = false
	next.balance.unsupported = false
	next.balance.latencySupported = false
	next.recoveryState.Store(c.recoveryState.Load())
	if next.balance.moving {
		next.homeCfg.Host = next.balance.target.IP
		next.homeCfg.Port = next.balance.target.Port
		if !next.recoveryState.CompareAndSwap(uint32(recoveryStateTakeoverEligible), uint32(recoveryStateSwitchingTakeover)) {
			next.recoveryState.CompareAndSwap(uint32(recoveryStateStable), uint32(recoveryStateSwitching))
		}
	}
	return next
}

// MembershipInstanceID returns the process-scoped Home membership identity.
func (c *Client) MembershipInstanceID() string {
	if c == nil {
		return ""
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.instanceID
}

// LegacyMembership reports whether this subscriber has downgraded to the legacy protocol.
func (c *Client) LegacyMembership() bool {
	if c == nil {
		return false
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.legacyMembership
}

// EnableLegacyMembership permanently downgrades this subscriber lifetime chain.
func (c *Client) EnableLegacyMembership() {
	if c == nil {
		return
	}
	c.mu.Lock()
	c.legacyMembership = true
	c.mu.Unlock()
	c.SuppressTakeover()
}

func (c *Client) Enabled() bool {
	if c == nil {
		return false
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.homeCfg.Enabled
}

func (c *Client) HeartbeatOK() bool {
	if c == nil {
		return false
	}
	if !c.Enabled() {
		return false
	}
	return c.heartbeatOK.Load()
}

// FenceDispatch stops new credential acquisitions while preserving transports and final concurrency releases.
func (c *Client) FenceDispatch() {
	if c != nil {
		c.dispatchFenced.Store(true)
	}
}

// Close permanently ends this client's dispatch and release lifetime.
func (c *Client) Close() {
	if c == nil {
		return
	}
	c.dispatchFenced.Store(true)
	c.releaseFenced.Store(true)
	c.heartbeatOK.Store(false)
	c.mu.Lock()
	commandClient, subscriptionClient, connections := c.detachClientsLocked()
	releaseClient := c.release
	c.release = nil
	closing := c.closing
	c.mu.Unlock()
	closeDetachedClients(commandClient, subscriptionClient, connections)
	if releaseClient != nil {
		_ = releaseClient.Close()
	}
	if closing != nil {
		<-closing
	}
}

// closeBootstrapPools replaces private bootstrap pools without ending the client lifetime.
func (c *Client) closeBootstrapPools() {
	if c == nil {
		return
	}
	c.heartbeatOK.Store(false)
	c.mu.Lock()
	c.releaseReady = false
	commandClient, subscriptionClient, connections := c.detachClientsLocked()
	c.mu.Unlock()
	closeDetachedClients(commandClient, subscriptionClient, connections)
}

// AbortAmbiguousDispatch fences this client after an auth dispatch response is ambiguous.
func (c *Client) AbortAmbiguousDispatch() {
	if c == nil {
		return
	}
	c.ambiguousDispatch.Store(true)
	c.dispatchFenced.Store(true)
	c.releaseFenced.Store(true)
	c.heartbeatOK.Store(false)
	c.mu.Lock()
	commandClient, subscriptionClient, connections := c.detachClientsLocked()
	releaseClient := c.release
	c.release = nil
	c.mu.Unlock()
	for _, conn := range connections {
		_ = conn.Close()
	}
	if commandClient != nil {
		go func() {
			_ = commandClient.Close()
		}()
	}
	if subscriptionClient != nil {
		go func() {
			_ = subscriptionClient.Close()
		}()
	}
	if releaseClient != nil {
		go func() {
			_ = releaseClient.Close()
		}()
	}
}

// AmbiguousDispatch reports whether this lifetime observed an issued dispatch with an unknown delivery result.
func (c *Client) AmbiguousDispatch() bool {
	return c != nil && c.ambiguousDispatch.Load()
}

// SuppressTakeover forces the next subscriber lifetime through normal membership recovery.
func (c *Client) SuppressTakeover() {
	if c == nil {
		return
	}
	if !c.recoveryState.CompareAndSwap(uint32(recoveryStateTakeoverEligible), uint32(recoveryStateStable)) {
		c.recoveryState.CompareAndSwap(uint32(recoveryStateSwitchingTakeover), uint32(recoveryStateSwitching))
	}
}

func (c *Client) detachClientsLocked() (*redis.Client, *redis.Client, []*homeDispatchConn) {
	connections := make([]*homeDispatchConn, 0, len(c.connections))
	for conn := range c.connections {
		connections = append(connections, conn)
	}
	commandClient := c.cmd
	subscriptionClient := c.sub
	c.cmd = nil
	c.cmdOptions = nil
	c.sub = nil
	c.connections = nil
	return commandClient, subscriptionClient, connections
}

func closeDetachedClients(commandClient *redis.Client, subscriptionClient *redis.Client, connections []*homeDispatchConn) {
	for _, conn := range connections {
		_ = conn.Close()
	}
	if commandClient != nil {
		_ = commandClient.Close()
	}
	if subscriptionClient != nil {
		_ = subscriptionClient.Close()
	}
}

func (c *Client) closeClientsLocked() {
	commandClient, subscriptionClient, connections := c.detachClientsLocked()
	releaseClient := c.release
	c.release = nil
	previousClosing := c.closing
	done := make(chan struct{})
	c.closing = done
	go func() {
		defer close(done)
		if previousClosing != nil {
			<-previousClosing
		}
		closeDetachedClients(commandClient, subscriptionClient, connections)
		if releaseClient != nil {
			_ = releaseClient.Close()
		}
	}()
}

func (c *Client) waitForClientsClosed() {
	for {
		c.mu.Lock()
		closing := c.closing
		c.mu.Unlock()
		if closing == nil {
			return
		}
		<-closing
		c.mu.Lock()
		if c.closing == closing {
			c.closing = nil
			c.mu.Unlock()
			return
		}
		c.mu.Unlock()
	}
}

// SetManagedLifetime defers client shutdown to the Service lifetime owner.
func (c *Client) SetManagedLifetime(managed bool) {
	if c == nil {
		return
	}
	c.mu.Lock()
	c.managed = managed
	c.mu.Unlock()
}

func (c *Client) managedLifetime() bool {
	if c == nil {
		return false
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.managed
}

func (c *Client) addr() (string, bool) {
	if c == nil {
		return "", false
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.addrLocked()
}

func (c *Client) addrLocked() (string, bool) {
	host := strings.TrimSpace(c.homeCfg.Host)
	if host == "" {
		return "", false
	}
	if c.homeCfg.Port <= 0 {
		return "", false
	}
	return net.JoinHostPort(host, strconv.Itoa(c.homeCfg.Port)), true
}

func (c *Client) ensureClients() error {
	if c == nil {
		return ErrDisabled
	}
	if c.dispatchFenced.Load() {
		return ErrDispatchFenced
	}
	if !c.Enabled() {
		return ErrDisabled
	}
	c.waitForClientsClosed()
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.dispatchFenced.Load() {
		return ErrDispatchFenced
	}

	addr, ok := c.addrLocked()
	if !ok {
		return fmt.Errorf("home: invalid address (host=%q port=%d)", c.homeCfg.Host, c.homeCfg.Port)
	}

	if c.cmd == nil {
		options, errOptions := c.redisOptionsLocked(addr)
		if errOptions != nil {
			return errOptions
		}
		c.cmdOptions = cloneRedisOptions(options)
		c.cmd = redis.NewClient(options)
	}
	if c.sub == nil {
		options, errOptions := c.redisOptionsLocked(addr)
		if errOptions != nil {
			return errOptions
		}
		c.sub = redis.NewClient(options)
	}
	return nil
}

func (c *Client) redisOptionsLocked(addr string) (*redis.Options, error) {
	options, tlsSettings := c.redisOptionsSnapshotLocked(addr)
	tlsConfig, errTLS := newHomeTLSConfig(tlsSettings, "")
	if errTLS != nil {
		return nil, errTLS
	}
	options.TLSConfig = tlsConfig
	options.Dialer = c.trackedRedisDialer(redis.NewDialer(options))
	return options, nil
}

// redisOptionsSnapshotLocked copies only values; certificate IO can run after releasing c.mu.
func (c *Client) redisOptionsSnapshotLocked(addr string) (*redis.Options, config.HomeTLSConfig) {
	tlsSettings := c.homeCfg.TLS
	serverName := strings.TrimSpace(tlsSettings.ServerName)
	if serverName == "" {
		if tlsSettings.UseTargetServerName {
			serverName = hostFromAddress(addr)
		} else {
			serverName = strings.TrimSpace(c.seedHost)
		}
	}
	if serverName == "" {
		serverName = strings.TrimSpace(c.homeCfg.Host)
	}
	tlsSettings.ServerName = serverName
	dialTimeout := homeRedisOperationTimeout
	readTimeout := homeRedisOperationTimeout
	writeTimeout := homeRedisOperationTimeout
	if c.testOperationTimeout > 0 {
		dialTimeout = c.testOperationTimeout
		readTimeout = c.testOperationTimeout
		writeTimeout = c.testOperationTimeout
	} else if c.cmdOptions != nil {
		if c.cmdOptions.DialTimeout > 0 {
			dialTimeout = c.cmdOptions.DialTimeout
		}
		if c.cmdOptions.ReadTimeout > 0 {
			readTimeout = c.cmdOptions.ReadTimeout
		}
		if c.cmdOptions.WriteTimeout > 0 {
			writeTimeout = c.cmdOptions.WriteTimeout
		}
	}
	options := &redis.Options{
		Addr:                  addr,
		DialTimeout:           dialTimeout,
		ReadTimeout:           readTimeout,
		WriteTimeout:          writeTimeout,
		MaxRetries:            -1,
		DialerRetries:         1,
		ContextTimeoutEnabled: true,
	}
	return options, tlsSettings
}

type homeDispatchConn struct {
	net.Conn
	client *Client
	once   sync.Once
}

func (c *Client) trackedRedisDialer(dialer func(context.Context, string, string) (net.Conn, error)) func(context.Context, string, string) (net.Conn, error) {
	return func(ctx context.Context, network string, address string) (net.Conn, error) {
		conn, errDial := dialer(ctx, network, address)
		if errDial != nil {
			return nil, errDial
		}
		wrapped := &homeDispatchConn{Conn: conn, client: c}
		if c == nil {
			return wrapped, nil
		}
		c.mu.Lock()
		if c.dispatchFenced.Load() {
			c.mu.Unlock()
			_ = wrapped.Close()
			return nil, ErrDispatchFenced
		}
		if c.connections == nil {
			c.connections = make(map[*homeDispatchConn]struct{})
		}
		c.connections[wrapped] = struct{}{}
		c.mu.Unlock()
		return wrapped, nil
	}
}

func (c *homeDispatchConn) untrack() {
	if c == nil {
		return
	}
	c.once.Do(func() {
		if c.client != nil {
			c.client.mu.Lock()
			delete(c.client.connections, c)
			c.client.mu.Unlock()
		}
	})
}

func (c *homeDispatchConn) Close() error {
	if c == nil || c.Conn == nil {
		return net.ErrClosed
	}
	c.untrack()
	return c.Conn.Close()
}

func (c *homeDispatchConn) NetConn() net.Conn {
	if c == nil {
		return nil
	}
	return c.Conn
}

func cloneRedisOptions(options *redis.Options) *redis.Options {
	if options == nil {
		return nil
	}
	cloned := *options
	if options.TLSConfig != nil {
		cloned.TLSConfig = options.TLSConfig.Clone()
	}
	if options.MaintNotificationsConfig != nil {
		maintNotifications := *options.MaintNotificationsConfig
		cloned.MaintNotificationsConfig = &maintNotifications
	}
	return &cloned
}

func hostFromAddress(addr string) string {
	host, _, errSplit := net.SplitHostPort(strings.TrimSpace(addr))
	if errSplit == nil {
		return strings.TrimSpace(host)
	}
	return strings.TrimSpace(addr)
}

func newHomeTLSConfig(cfg config.HomeTLSConfig, fallbackServerName string) (*tls.Config, error) {
	if !cfg.Enable {
		return nil, nil
	}

	serverName := strings.TrimSpace(cfg.ServerName)
	if serverName == "" {
		serverName = strings.TrimSpace(fallbackServerName)
	}

	tlsConfig := &tls.Config{
		MinVersion:         tls.VersionTLS12,
		ServerName:         serverName,
		InsecureSkipVerify: cfg.InsecureSkipVerify,
	}

	clientCertPath := strings.TrimSpace(cfg.ClientCert)
	clientKeyPath := strings.TrimSpace(cfg.ClientKey)
	if clientCertPath != "" || clientKeyPath != "" {
		if clientCertPath == "" || clientKeyPath == "" {
			return nil, fmt.Errorf("home tls: client certificate and key must be set together")
		}
		certPair, errLoad := tls.LoadX509KeyPair(clientCertPath, clientKeyPath)
		if errLoad != nil {
			return nil, fmt.Errorf("home tls: load client certificate: %w", errLoad)
		}
		tlsConfig.Certificates = []tls.Certificate{certPair}
	}

	caCertPath := strings.TrimSpace(cfg.CACert)
	if caCertPath == "" {
		return tlsConfig, nil
	}

	caCertPEM, errRead := os.ReadFile(caCertPath)
	if errRead != nil {
		return nil, fmt.Errorf("home tls: read ca-cert: %w", errRead)
	}

	certPool, errPool := x509.SystemCertPool()
	if errPool != nil || certPool == nil {
		certPool = x509.NewCertPool()
	}
	if !certPool.AppendCertsFromPEM(caCertPEM) {
		return nil, fmt.Errorf("home tls: ca-cert contains no PEM certificates")
	}
	tlsConfig.RootCAs = certPool

	return tlsConfig, nil
}

func (c *Client) commandClient() (*redis.Client, error) {
	if c == nil || c.dispatchFenced.Load() {
		return nil, ErrDispatchFenced
	}
	if errEnsure := c.ensureClients(); errEnsure != nil {
		return nil, errEnsure
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.dispatchFenced.Load() {
		return nil, ErrDispatchFenced
	}
	if c.cmd == nil {
		return nil, ErrNotConnected
	}
	return c.cmd, nil
}

func (c *Client) pluginSyncCommandOptions() (*redis.Options, error) {
	if errEnsure := c.ensureClients(); errEnsure != nil {
		return nil, errEnsure
	}
	c.mu.Lock()
	options := cloneRedisOptions(c.cmdOptions)
	c.mu.Unlock()
	if options == nil {
		return nil, ErrNotConnected
	}
	return options, nil
}

func (c *Client) subscriptionClient() (*redis.Client, error) {
	if errEnsure := c.ensureClients(); errEnsure != nil {
		return nil, errEnsure
	}
	c.mu.Lock()
	sub := c.sub
	c.mu.Unlock()
	if sub == nil {
		return nil, ErrNotConnected
	}
	return sub, nil
}

func (c *Client) Ping(ctx context.Context) error {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return errClient
	}
	return cmd.Ping(ctx).Err()
}

func (c *Client) clusterDiscoveryEnabled() bool {
	if c == nil {
		return false
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.clusterDiscoveryEnabledLocked()
}

func (c *Client) clusterDiscoveryEnabledLocked() bool {
	return !c.homeCfg.DisableClusterDiscovery
}

func (c *Client) refreshBestClusterNode(ctx context.Context) error {
	if !c.clusterDiscoveryEnabled() {
		return nil
	}
	switched, errRefresh := c.refreshClusterNodes(ctx)
	if errRefresh != nil {
		log.Debugf("home cluster nodes unavailable: %v", errRefresh)
		return errRefresh
	}
	if switched {
		if addr, ok := c.addr(); ok {
			log.Infof("home cluster target switched to %s", addr)
		}
	}
	return nil
}

func (c *Client) refreshClusterNodes(ctx context.Context) (bool, error) {
	if !c.clusterDiscoveryEnabled() {
		return false, nil
	}

	if ctx == nil {
		ctx = context.Background()
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return false, fmt.Errorf("%w: %w", errClusterDiscoveryTransport, errClient)
	}
	nodesCommand := cmd.Do(ctx, "CLUSTER", "NODES")
	errDo := nodesCommand.Err()
	if errDo != nil {
		var redisErr redis.Error
		if !errors.As(errDo, &redisErr) {
			return false, fmt.Errorf("%w: %w", errClusterDiscoveryTransport, errDo)
		}
		return false, errDo
	}
	raw, errText := nodesCommand.Text()
	if errText != nil {
		return false, errText
	}
	nodes, errParse := parseClusterNodesPayload([]byte(raw))
	if errParse != nil {
		return false, errParse
	}
	if len(nodes) == 0 {
		return false, nil
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.clusterNodes = nodes
	// Discovery alone does not establish recovery or reset reconnect failures.
	// Keep planned handoffs pinned here; repeated failures may still trigger failover.
	if c.balance.moving || c.balance.returning || c.heartbeatOK.Load() {
		return false, nil
	}
	node, ok := c.chooseClusterNodeLocked(nodes, false, rand.Float64())
	return ok && c.switchToNodeLocked(node), nil
}

func parseClusterNodesPayload(raw []byte) ([]clusterNode, error) {
	var envelope clusterNodesEnvelope
	if errUnmarshal := json.Unmarshal(raw, &envelope); errUnmarshal != nil {
		return nil, errUnmarshal
	}
	return normalizeClusterNodes(envelope.Nodes), nil
}

func (c *Client) updateClusterNodesFromPayload(raw []byte) error {
	if c == nil || !c.clusterDiscoveryEnabled() {
		return nil
	}
	nodes, errParse := parseClusterNodesPayload(raw)
	if errParse != nil {
		return errParse
	}
	c.mu.Lock()
	c.clusterNodes = nodes
	c.mu.Unlock()
	return nil
}

func normalizeClusterNodes(nodes []clusterNode) []clusterNode {
	out := make([]clusterNode, 0, len(nodes))
	seen := make(map[string]bool, len(nodes))
	for _, node := range nodes {
		node.IP = strings.TrimSpace(node.IP)
		if node.IP == "" || node.Port <= 0 || node.Port > 65535 {
			continue
		}
		address := clusterNodeAddress(node)
		if seen[address] {
			continue
		}
		seen[address] = true
		if node.ClientCount < 0 {
			node.ClientCount = 0
		}
		out = append(out, node)
	}
	sort.SliceStable(out, func(i, j int) bool {
		return out[i].ClientCount < out[j].ClientCount
	})
	return out
}

func (c *Client) switchToNodeLocked(node clusterNode) bool {
	host := strings.TrimSpace(node.IP)
	if host == "" || node.Port <= 0 {
		return false
	}
	if strings.TrimSpace(c.homeCfg.Host) == host && c.homeCfg.Port == node.Port {
		return false
	}
	c.releaseReady = false
	c.homeCfg.Host = host
	c.homeCfg.Port = node.Port
	if !c.recoveryState.CompareAndSwap(uint32(recoveryStateStable), uint32(recoveryStateSwitching)) {
		c.recoveryState.CompareAndSwap(uint32(recoveryStateTakeoverEligible), uint32(recoveryStateSwitchingTakeover))
	}
	c.closeClientsLocked()
	return true
}

func (c *Client) markReconnectFailure(reason string) {
	switched, addr := c.failoverAfterReconnectFailure()
	if switched {
		log.Warnf("home control center unavailable after repeated %s failures; switching to %s", reason, addr)
	}
}

func (c *Client) failoverAfterReconnectFailure() (bool, string) {
	if c == nil {
		return false, ""
	}
	c.mu.Lock()
	defer c.mu.Unlock()

	if !c.clusterDiscoveryEnabledLocked() {
		c.reconnectFailures = 0
		return false, ""
	}
	c.reconnectFailures++
	if c.reconnectFailures < homeReconnectFailoverThreshold {
		return false, ""
	}
	c.reconnectFailures = 0

	return c.switchToNextNodeLocked()
}

func (c *Client) failoverAfterSubscriptionTimeout() (bool, string) {
	if c == nil {
		return false, ""
	}
	c.mu.Lock()
	defer c.mu.Unlock()

	if !c.clusterDiscoveryEnabledLocked() {
		c.reconnectFailures = 0
		return false, ""
	}
	c.reconnectFailures = 0
	return c.switchToNextNodeLocked()
}

func (c *Client) switchToNextNodeLocked() (bool, string) {
	currentHost := strings.TrimSpace(c.homeCfg.Host)
	currentPort := c.homeCfg.Port
	node, ok := c.chooseClusterNodeLocked(c.clusterNodes, true, rand.Float64())
	if !ok && strings.TrimSpace(c.seedHost) != "" && c.seedPort > 0 && (c.seedHost != currentHost || c.seedPort != currentPort) {
		node, ok = clusterNode{IP: c.seedHost, Port: c.seedPort}, true
	}
	if ok && c.switchToNodeLocked(node) {
		addr, _ := c.addrLocked()
		return true, addr
	}
	return false, ""
}

func (c *Client) markSubscriptionTimeout() {
	switched, addr := c.failoverAfterSubscriptionTimeout()
	if switched {
		log.Warnf("home subscription heartbeat timeout; switching to %s", addr)
	}
}

func (c *Client) resetReconnectFailures() {
	if c == nil {
		return
	}
	c.mu.Lock()
	c.reconnectFailures = 0
	c.mu.Unlock()
}

func (c *Client) GetConfig(ctx context.Context) ([]byte, error) {
	if errRefresh := c.refreshBestClusterNode(ctx); errors.Is(errRefresh, errClusterDiscoveryTransport) {
		return nil, errRefresh
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, errClient
	}
	raw, err := cmd.Get(ctx, redisKeyConfig).Bytes()
	if errors.Is(err, redis.Nil) {
		return nil, ErrConfigNotFound
	}
	if err != nil {
		return nil, err
	}
	if len(raw) == 0 {
		return nil, ErrEmptyResponse
	}
	return raw, nil
}

func (c *Client) GetModels(ctx context.Context, headers http.Header, query url.Values) ([]byte, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, errClient
	}
	req := modelsRequest{
		Type:    "models",
		Headers: headersToLowerMap(headers),
		Query:   queryToLowerMap(query),
	}
	keyBytes, err := json.Marshal(&req)
	if err != nil {
		return nil, err
	}
	raw, err := cmd.Get(ctx, string(keyBytes)).Bytes()
	if errors.Is(err, redis.Nil) {
		return nil, ErrModelsNotFound
	}
	if err != nil {
		return nil, err
	}
	if len(raw) == 0 {
		return nil, ErrEmptyResponse
	}
	return raw, nil
}

func buildKVSetArgs(key string, value []byte, opts KVSetOptions) ([]any, error) {
	return kvstore.BuildSetArgs(key, value, opts)
}

func durationCeil(value time.Duration, unit time.Duration) int64 {
	return kvstore.DurationCeil(value, unit)
}

func (c *Client) KVGet(ctx context.Context, key string) ([]byte, bool, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, false, errClient
	}
	raw, errGet := cmd.Get(ctx, key).Bytes()
	if errors.Is(errGet, redis.Nil) {
		return nil, false, nil
	}
	if errGet != nil {
		return nil, false, errGet
	}
	return append([]byte(nil), raw...), true, nil
}

func (c *Client) KVSet(ctx context.Context, key string, value []byte, opts KVSetOptions) (bool, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return false, errClient
	}
	args, errArgs := buildKVSetArgs(key, value, opts)
	if errArgs != nil {
		return false, errArgs
	}
	result, errSet := cmd.Do(ctx, append([]any{"SET"}, args...)...).Result()
	if errors.Is(errSet, redis.Nil) {
		return false, nil
	}
	if errSet != nil {
		return false, errSet
	}
	if result == nil {
		return false, nil
	}
	return true, nil
}

func (c *Client) KVSetNX(ctx context.Context, key string, value []byte, ttl time.Duration) (bool, error) {
	opts := KVSetOptions{NX: true}
	if ttl > 0 {
		opts.EX = ttl
	}
	return c.KVSet(ctx, key, value, opts)
}

// KVCompareAndSwap atomically replaces a value only when its current state matches the expected state.
//
// It uses Home's dedicated CAS command:
//
//	CAS <key> <expected-exists 0|1> <expected-value> <new-value> [PX <ttl-ms>]
//
// Omitting PX stores the value without a TTL. Home replies integer 1 when the
// swap happened and integer 0 when the state did not match. Deployments that
// predate CAS reject the command, which latches ErrCompareAndSwapUnsupported for
// this client lifetime so later calls skip the round trip.
func (c *Client) KVCompareAndSwap(ctx context.Context, key string, expected []byte, expectedExists bool, value []byte, ttl time.Duration) (bool, error) {
	if c == nil {
		return false, ErrNotConnected
	}
	if c.casUnsupported.Load() {
		return false, ErrCompareAndSwapUnsupported
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return false, errClient
	}
	expectedFlag := "0"
	if expectedExists {
		expectedFlag = "1"
	}
	args := make([]any, 0, 7)
	args = append(args, "CAS", key, expectedFlag, expected, value)
	if milliseconds := durationCeil(ttl, time.Millisecond); milliseconds > 0 {
		args = append(args, "PX", milliseconds)
	}
	result, errCAS := cmd.Do(ctx, args...).Int64()
	if errCAS != nil {
		if isHomeCommandUnsupported(errCAS) {
			if c.casUnsupported.CompareAndSwap(false, true) {
				log.Warnf("home kv: this Home does not implement the CAS command; Antigravity and Codex reasoning replay are disabled until Home is upgraded")
			}
			return false, ErrCompareAndSwapUnsupported
		}
		return false, errCAS
	}
	return result == 1, nil
}

func (c *Client) KVDel(ctx context.Context, keys ...string) (int64, error) {
	if len(keys) == 0 {
		return 0, nil
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return 0, errClient
	}
	return cmd.Del(ctx, keys...).Result()
}

func (c *Client) KVExpire(ctx context.Context, key string, ttl time.Duration) (bool, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return false, errClient
	}
	return cmd.Expire(ctx, key, ttl).Result()
}

func (c *Client) KVTTL(ctx context.Context, key string) (time.Duration, bool, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return 0, false, errClient
	}
	ttl, errTTL := cmd.TTL(ctx, key).Result()
	if errTTL != nil {
		return 0, false, errTTL
	}
	switch {
	case ttl <= -2*time.Second:
		return 0, false, nil
	case ttl == -1*time.Second:
		return 0, true, nil
	default:
		return ttl, true, nil
	}
}

func (c *Client) KVIncrBy(ctx context.Context, key string, delta int64) (int64, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return 0, errClient
	}
	return cmd.IncrBy(ctx, key, delta).Result()
}

func (c *Client) KVMGet(ctx context.Context, keys ...string) ([][]byte, []bool, error) {
	if len(keys) == 0 {
		return nil, nil, nil
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, nil, errClient
	}
	items, errMGet := cmd.MGet(ctx, keys...).Result()
	if errMGet != nil {
		return nil, nil, errMGet
	}
	values := make([][]byte, len(items))
	found := make([]bool, len(items))
	for i, item := range items {
		switch typed := item.(type) {
		case nil:
			continue
		case string:
			values[i] = []byte(typed)
			found[i] = true
		case []byte:
			values[i] = append([]byte(nil), typed...)
			found[i] = true
		default:
			return nil, nil, fmt.Errorf("home kv: unsupported MGET item type %T", item)
		}
	}
	return values, found, nil
}

func (c *Client) KVMSet(ctx context.Context, pairs map[string][]byte) error {
	if len(pairs) == 0 {
		return nil
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return errClient
	}
	keys := make([]string, 0, len(pairs))
	for key := range pairs {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	args := make([]any, 0, 1+len(keys)*2)
	args = append(args, "MSET")
	for _, key := range keys {
		args = append(args, key, append([]byte(nil), pairs[key]...))
	}
	return cmd.Do(ctx, args...).Err()
}

func headersToLowerMap(headers http.Header) map[string]string {
	if len(headers) == 0 {
		return nil
	}
	out := make(map[string]string, len(headers))
	for key, values := range headers {
		k := strings.ToLower(strings.TrimSpace(key))
		if k == "" {
			continue
		}
		if len(values) == 0 {
			out[k] = ""
			continue
		}
		trimmed := make([]string, 0, len(values))
		for _, v := range values {
			trimmed = append(trimmed, strings.TrimSpace(v))
		}
		out[k] = strings.Join(trimmed, ", ")
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func queryToLowerMap(query url.Values) map[string]string {
	if len(query) == 0 {
		return nil
	}
	out := make(map[string]string, len(query))
	for key, values := range query {
		k := strings.ToLower(strings.TrimSpace(key))
		if k == "" {
			continue
		}
		if len(values) == 0 {
			out[k] = ""
			continue
		}
		trimmed := make([]string, 0, len(values))
		for _, v := range values {
			trimmed = append(trimmed, strings.TrimSpace(v))
		}
		out[k] = strings.Join(trimmed, ", ")
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func newAuthDispatchRequest(requestedModel string, sessionID string, parentSessionID string, headers http.Header, count int, credentialPolicy string, excludedAuthIDs *[]string, pinnedAuthID string) authDispatchRequest {
	if count <= 0 {
		count = 1
	}
	var excludedAuthIDsCopy *[]string
	if excludedAuthIDs != nil {
		// Keep count at one so older Home servers that ignore excluded_auth_ids do
		// not apply their legacy count-based retry cap before CPA can rotate
		// credentials. New Home servers apply retry_round eligibility remotely.
		count = 1
		values := append([]string{}, (*excludedAuthIDs)...)
		excludedAuthIDsCopy = &values
	}
	nodeKind := strings.TrimSpace(headers.Get("X-Node-Kind"))
	if nodeKind == "" {
		nodeKind = strings.TrimSpace(headers.Get("x-node-kind"))
	}
	return authDispatchRequest{
		Type:                "auth",
		Model:               requestedModel,
		Count:               count,
		ConcurrencyProtocol: 1,
		SessionID:           strings.TrimSpace(sessionID),
		ParentSessionID:     strings.TrimSpace(parentSessionID),
		NodeKind:            nodeKind,
		Headers:             headersToLowerMap(headers),
		CredentialPolicy:    strings.TrimSpace(credentialPolicy),
		ExcludedAuthIDs:     excludedAuthIDsCopy,
		PinnedAuthID:        strings.TrimSpace(pinnedAuthID),
	}
}

func newAuthDispatchRequestWithRetryRound(requestedModel string, sessionID string, parentSessionID string, headers http.Header, count int, credentialPolicy string, retryRound int, excludedAuthIDs *[]string, pinnedAuthID string) authDispatchRequest {
	req := newAuthDispatchRequest(requestedModel, sessionID, parentSessionID, headers, count, credentialPolicy, excludedAuthIDs, pinnedAuthID)
	if retryRound < 0 {
		retryRound = 0
	}
	req.RetryRound = &retryRound
	return req
}

func (c *Client) RPopAuth(ctx context.Context, requestedModel string, sessionID string, headers http.Header, count int) ([]byte, error) {
	return c.rPopAuth(ctx, requestedModel, sessionID, "", headers, count, "", nil, nil, "")
}

// RPopAuthWithPolicy requests a Home credential constrained by the supplied fixed policy.
func (c *Client) RPopAuthWithPolicy(ctx context.Context, requestedModel string, sessionID string, headers http.Header, count int, credentialPolicy string) ([]byte, error) {
	return c.rPopAuth(ctx, requestedModel, sessionID, "", headers, count, credentialPolicy, nil, nil, "")
}

// RPopAuthWithConstraints requests a credential using the current retry-round
// exclusions and optional pinned credential constraint.
func (c *Client) RPopAuthWithConstraints(ctx context.Context, requestedModel string, sessionID string, headers http.Header, count int, excludedAuthIDs []string, pinnedAuthID string) ([]byte, error) {
	return c.rPopAuth(ctx, requestedModel, sessionID, "", headers, count, "", nil, &excludedAuthIDs, pinnedAuthID)
}

// RPopAuthWithPolicyAndConstraints combines a fixed credential policy with the
// current retry-round exclusions and optional pinned credential constraint.
func (c *Client) RPopAuthWithPolicyAndConstraints(ctx context.Context, requestedModel string, sessionID string, headers http.Header, count int, credentialPolicy string, excludedAuthIDs []string, pinnedAuthID string) ([]byte, error) {
	return c.rPopAuth(ctx, requestedModel, sessionID, "", headers, count, credentialPolicy, nil, &excludedAuthIDs, pinnedAuthID)
}

// RPopAuthWithRetryRoundConstraints requests a credential with the retry round,
// current-round exclusions, and optional pinned credential constraint.
func (c *Client) RPopAuthWithRetryRoundConstraints(ctx context.Context, requestedModel string, sessionID string, headers http.Header, count int, retryRound int, excludedAuthIDs []string, pinnedAuthID string) ([]byte, error) {
	return c.rPopAuth(ctx, requestedModel, sessionID, "", headers, count, "", &retryRound, &excludedAuthIDs, pinnedAuthID)
}

// RPopAuthWithPolicyAndRetryRoundConstraints combines a credential policy with
// the retry round, current-round exclusions, and optional pin.
func (c *Client) RPopAuthWithPolicyAndRetryRoundConstraints(ctx context.Context, requestedModel string, sessionID string, headers http.Header, count int, credentialPolicy string, retryRound int, excludedAuthIDs []string, pinnedAuthID string) ([]byte, error) {
	return c.rPopAuth(ctx, requestedModel, sessionID, "", headers, count, credentialPolicy, &retryRound, &excludedAuthIDs, pinnedAuthID)
}

// RPopAuthWithSessionHierarchy requests a Home credential with both session ID and parent session ID for hierarchical soft affinity.
func (c *Client) RPopAuthWithSessionHierarchy(ctx context.Context, requestedModel string, sessionID string, parentSessionID string, headers http.Header, count int, credentialPolicy string, retryRound *int, excludedAuthIDs []string, pinnedAuthID string) ([]byte, error) {
	var excludedPtr *[]string
	if len(excludedAuthIDs) > 0 {
		excludedPtr = &excludedAuthIDs
	}
	return c.rPopAuth(ctx, requestedModel, sessionID, parentSessionID, headers, count, credentialPolicy, retryRound, excludedPtr, pinnedAuthID)
}

func (c *Client) rPopAuth(ctx context.Context, requestedModel string, sessionID string, parentSessionID string, headers http.Header, count int, credentialPolicy string, retryRound *int, excludedAuthIDs *[]string, pinnedAuthID string) ([]byte, error) {
	if c == nil || c.dispatchFenced.Load() {
		return nil, ErrDispatchFenced
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if errContext := ctx.Err(); errContext != nil {
		return nil, errContext
	}
	requestedModel = strings.TrimSpace(requestedModel)
	if requestedModel == "" {
		return nil, fmt.Errorf("home: requested model is empty")
	}
	var req authDispatchRequest
	if retryRound == nil {
		req = newAuthDispatchRequest(requestedModel, sessionID, parentSessionID, headers, count, credentialPolicy, excludedAuthIDs, pinnedAuthID)
	} else {
		req = newAuthDispatchRequestWithRetryRound(requestedModel, sessionID, parentSessionID, headers, count, credentialPolicy, *retryRound, excludedAuthIDs, pinnedAuthID)
	}
	keyBytes, errMarshal := json.Marshal(&req)
	if errMarshal != nil {
		return nil, errMarshal
	}
	if c.dispatchFenced.Load() {
		return nil, ErrDispatchFenced
	}
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, errClient
	}
	if c.dispatchFenced.Load() {
		return nil, ErrDispatchFenced
	}
	conn := cmd.Conn()
	defer func() {
		if errClose := conn.Close(); errClose != nil {
			log.Debugf("Home auth dispatch connection close failed: %v", errClose)
		}
	}()
	if errProbe := conn.Ping(ctx).Err(); errProbe != nil {
		return nil, errProbe
	}
	if c.dispatchFenced.Load() {
		return nil, ErrDispatchFenced
	}
	raw, errRPop := conn.RPop(ctx, string(keyBytes)).Bytes()
	if errors.Is(errRPop, redis.Nil) {
		return nil, ErrAuthNotFound
	}
	if errRPop != nil {
		if isAmbiguousIssuedRPopAuthError(errRPop) {
			return nil, NewAmbiguousDispatchError(errRPop)
		}
		return nil, errRPop
	}
	if len(raw) == 0 {
		return nil, ErrEmptyResponse
	}
	return raw, nil
}

func isAmbiguousIssuedRPopAuthError(err error) bool {
	if err == nil || errors.Is(err, redis.Nil) {
		return false
	}
	var redisErr redis.Error
	return !errors.As(err, &redisErr)
}

func (c *Client) GetRefreshAuth(ctx context.Context, authIndex string, accessTokenSHA256 string) ([]byte, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, errClient
	}
	authIndex = strings.TrimSpace(authIndex)
	if authIndex == "" {
		return nil, fmt.Errorf("home: auth_index is empty")
	}
	req := refreshRequest{
		Type:      "refresh",
		AuthIndex: authIndex,
	}
	req.ObservedAccessTokenSHA256 = strings.TrimSpace(accessTokenSHA256)
	keyBytes, err := json.Marshal(&req)
	if err != nil {
		return nil, err
	}

	raw, err := cmd.WithTimeout(homeRefreshOperationTimeout).Get(ctx, string(keyBytes)).Bytes()
	if errors.Is(err, redis.Nil) {
		return nil, ErrAuthNotFound
	}
	if err != nil {
		return nil, err
	}
	if len(raw) == 0 {
		return nil, ErrEmptyResponse
	}
	return raw, nil
}

func (c *Client) LPushUsage(ctx context.Context, payload []byte) error {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return errClient
	}
	if len(payload) == 0 {
		return nil
	}
	return cmd.LPush(ctx, redisKeyUsage, payload).Err()
}

// LPushInFlightSnapshot publishes a bounded in-flight observation frame.
func (c *Client) LPushInFlightSnapshot(ctx context.Context, payload []byte) error {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return errClient
	}
	return cmd.LPush(ctx, redisKeyInFlightSnapshot, payload).Err()
}

// PushConcurrencyRelease sends one cumulative concurrency release frame through an independent client.
func (c *Client) PushConcurrencyRelease(ctx context.Context, frame ConcurrencyReleaseFrame) error {
	if frame.CredentialID == "" || frame.Model == "" || frame.ReleaseSeq <= 0 {
		return fmt.Errorf("invalid concurrency release frame")
	}
	cmd, errClient := c.concurrencyReleaseClient()
	if errClient != nil {
		return errClient
	}
	payload, errMarshal := json.Marshal(frame)
	if errMarshal != nil {
		return fmt.Errorf("marshal concurrency release frame: %w", errMarshal)
	}
	return cmd.Do(ctx, "LPUSH", redisKeyConcurrencyRelease, payload).Err()
}

func (c *Client) concurrencyReleaseClient() (*redis.Client, error) {
	if c == nil || c.releaseFenced.Load() {
		return nil, ErrDispatchFenced
	}
	if !c.Enabled() {
		return nil, ErrDisabled
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	if c.releaseFenced.Load() {
		return nil, ErrDispatchFenced
	}
	state := recoveryState(c.recoveryState.Load())
	// Preparing takeover does not move membership. The acknowledged source owns final releases
	// until Close; an unacknowledged replacement or a switched address does not inherit that right.
	ownsSource := state == recoveryStateTakeoverEligible && c.releaseReady
	if state != recoveryStateStable && !ownsSource {
		return nil, ErrNotConnected
	}
	if c.release != nil {
		return c.release, nil
	}
	addr, ok := c.addrLocked()
	if !ok {
		return nil, fmt.Errorf("home: invalid address (host=%q port=%d)", c.homeCfg.Host, c.homeCfg.Port)
	}
	options, errOptions := c.redisOptionsLocked(addr)
	if errOptions != nil {
		return nil, errOptions
	}
	options.Dialer = redis.NewDialer(options)
	c.release = redis.NewClient(options)
	return c.release, nil
}

func (c *Client) RPushRequestLog(ctx context.Context, payload []byte) error {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return errClient
	}
	if len(payload) == 0 {
		return nil
	}
	return cmd.RPush(ctx, redisKeyRequestLog, payload).Err()
}

func (c *Client) RPushAppLog(ctx context.Context, payload []byte) error {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return errClient
	}
	if len(payload) == 0 {
		return nil
	}
	return cmd.RPush(ctx, redisKeyAppLog, payload).Err()
}

func (c *Client) RPushPluginStatus(ctx context.Context, payload []byte) error {
	options, errOptions := c.pluginSyncCommandOptions()
	if errOptions != nil {
		return errOptions
	}
	if len(payload) == 0 {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	// Plugin cancellation owns only this command's transport, never dispatch or release connections.
	baseDialer := options.Dialer
	if baseDialer == nil {
		baseDialer = pluginSyncDialer(options)
	}
	options.Dialer = func(dialCtx context.Context, network, address string) (net.Conn, error) {
		conn, errDial := baseDialer(dialCtx, network, address)
		if errDial != nil {
			return nil, errDial
		}
		return newPluginSyncCancelableConn(ctx, conn), nil
	}
	statusClient := redis.NewClient(options)
	errPush := statusClient.RPush(ctx, redisKeyPluginStatus, payload).Err()
	errClose := statusClient.Close()
	if errContext := ctx.Err(); errContext != nil {
		return errContext
	}
	if errClose != nil {
		return errors.Join(errPush, fmt.Errorf("close plugin status client: %w", errClose))
	}
	return errPush
}

func (c *Client) GetPluginTasks(ctx context.Context) ([]PluginTask, error) {
	cmd, errClient := c.commandClient()
	if errClient != nil {
		return nil, errClient
	}
	raw, errGet := cmd.Get(ctx, redisKeyPluginTasks).Bytes()
	if errors.Is(errGet, redis.Nil) {
		return nil, nil
	}
	if errGet != nil {
		return nil, errGet
	}
	if len(raw) == 0 {
		return nil, nil
	}
	var tasks []PluginTask
	if errUnmarshal := json.Unmarshal(raw, &tasks); errUnmarshal != nil {
		return nil, errUnmarshal
	}
	return tasks, nil
}

func (c *Client) GetPluginSync(ctx context.Context, request pluginstore.PluginSyncRequest) (pluginstore.PluginSyncResponse, error) {
	options, errOptions := c.pluginSyncCommandOptions()
	if errOptions != nil {
		return pluginstore.PluginSyncResponse{}, errOptions
	}
	payload, errMarshal := json.Marshal(request)
	if errMarshal != nil {
		return pluginstore.PluginSyncResponse{}, fmt.Errorf("marshal plugin sync request: %w", errMarshal)
	}
	requestCmd := redis.NewStringCmd(ctx, "get", redisKeyPluginSync, string(payload))
	if errProcess := processPluginSyncCommand(ctx, options, requestCmd); errProcess != nil {
		if message, ok := pluginSyncUnsupportedMessage(errProcess.Error()); ok {
			return pluginstore.PluginSyncResponse{}, fmt.Errorf("%w: %s", ErrPluginSyncUnsupported, message)
		}
		return pluginstore.PluginSyncResponse{}, errProcess
	}
	raw, errBytes := requestCmd.Bytes()
	if errBytes != nil {
		return pluginstore.PluginSyncResponse{}, errBytes
	}
	defer func() {
		requestCmd.SetVal("")
		for index := range raw {
			raw[index] = 0
		}
	}()
	if len(raw) == 0 {
		return pluginstore.PluginSyncResponse{}, ErrEmptyResponse
	}
	if message, ok := pluginSyncUnsupportedResponse(raw); ok {
		return pluginstore.PluginSyncResponse{}, fmt.Errorf("%w: %s", ErrPluginSyncUnsupported, message)
	}
	var response pluginstore.PluginSyncResponse
	if errUnmarshal := json.Unmarshal(raw, &response); errUnmarshal != nil {
		response.Clear()
		return pluginstore.PluginSyncResponse{}, fmt.Errorf("decode plugin sync response: %w", errUnmarshal)
	}
	if errValidate := response.Validate(time.Now().UTC()); errValidate != nil {
		response.Clear()
		return pluginstore.PluginSyncResponse{}, errValidate
	}
	return response, nil
}

func processPluginSyncCommand(ctx context.Context, options *redis.Options, command redis.Cmder) error {
	if options == nil {
		return ErrNotConnected
	}
	if ctx == nil {
		ctx = context.Background()
	}
	pluginSyncClient := newPluginSyncCommandClient(ctx, options)
	if pluginSyncClient == nil {
		return ErrNotConnected
	}
	errProcess := pluginSyncClient.Process(ctx, command)
	errClose := pluginSyncClient.Close()
	if errContext := ctx.Err(); errContext != nil {
		return errContext
	}
	if errProcess != nil {
		return errProcess
	}
	if errClose != nil {
		return fmt.Errorf("close plugin sync command client: %w", errClose)
	}
	return nil
}

func newPluginSyncCommandClient(ctx context.Context, template *redis.Options) *redis.Client {
	options := cloneRedisOptions(template)
	if options == nil {
		return nil
	}
	options.MaintNotificationsConfig = &maintnotifications.Config{Mode: maintnotifications.ModeDisabled}
	baseDialer := options.Dialer
	if baseDialer == nil {
		baseDialer = pluginSyncDialer(options)
	}
	options.Dialer = func(dialCtx context.Context, network string, address string) (net.Conn, error) {
		conn, errDial := baseDialer(dialCtx, network, address)
		if errDial != nil {
			return nil, errDial
		}
		return newPluginSyncCancelableConn(ctx, conn), nil
	}
	options.ReadTimeout = homePluginSyncOperationTimeout
	options.MaxRetries = -1
	return redis.NewClient(options)
}

func pluginSyncDialer(options *redis.Options) func(context.Context, string, string) (net.Conn, error) {
	return func(ctx context.Context, network string, address string) (net.Conn, error) {
		dialer := &net.Dialer{Timeout: options.DialTimeout, KeepAlive: 5 * time.Minute}
		conn, errDial := dialer.DialContext(ctx, network, address)
		if errDial != nil {
			return nil, errDial
		}
		if options.TLSConfig == nil {
			return conn, nil
		}
		tlsConn := tls.Client(conn, options.TLSConfig)
		if errHandshake := tlsConn.HandshakeContext(ctx); errHandshake != nil {
			return nil, errors.Join(errHandshake, conn.Close())
		}
		return tlsConn, nil
	}
}

type pluginSyncCancelableConn struct {
	net.Conn
	ctx                 context.Context
	stopCancel          func() bool
	once                sync.Once
	closeUnderlyingOnce sync.Once
}

func newPluginSyncCancelableConn(ctx context.Context, conn net.Conn) net.Conn {
	if ctx == nil {
		ctx = context.Background()
	}
	wrapped := &pluginSyncCancelableConn{
		Conn: conn,
		ctx:  ctx,
	}
	wrapped.stopCancel = context.AfterFunc(ctx, wrapped.closeUnderlying)
	if ctx.Err() != nil {
		wrapped.closeUnderlying()
	}
	return wrapped
}

func (c *pluginSyncCancelableConn) closeUnderlying() {
	if c == nil || c.Conn == nil {
		return
	}
	c.closeUnderlyingOnce.Do(func() {
		_ = closeUnderlyingTransport(c.Conn)
	})
}

func closeUnderlyingTransport(conn net.Conn) error {
	if conn == nil {
		return net.ErrClosed
	}
	current := conn
	for {
		if dispatchConn, ok := current.(*homeDispatchConn); ok {
			dispatchConn.untrack()
			if next := dispatchConn.NetConn(); next != nil && next != current {
				current = next
				continue
			}
		}
		if tlsConn, ok := current.(*tls.Conn); ok {
			if netConn := tlsConn.NetConn(); netConn != nil && netConn != current {
				current = netConn
				continue
			}
		}
		type unwrapper interface {
			NetConn() net.Conn
		}
		if u, ok := current.(unwrapper); ok {
			if next := u.NetConn(); next != nil && next != current {
				current = next
				continue
			}
		}
		break
	}
	return current.Close()
}

func (c *pluginSyncCancelableConn) Close() error {
	if c == nil || c.Conn == nil {
		return net.ErrClosed
	}
	defer c.once.Do(func() {
		if c.stopCancel != nil {
			c.stopCancel()
		}
	})
	if c.ctx != nil && c.ctx.Err() != nil {
		c.closeUnderlying()
	}
	return c.Conn.Close()
}

func pluginSyncUnsupportedResponse(raw []byte) (string, bool) {
	var response struct {
		Error struct {
			Code    string `json:"code"`
			Type    string `json:"type"`
			Message string `json:"message"`
		} `json:"error"`
	}
	if errUnmarshal := json.Unmarshal(raw, &response); errUnmarshal != nil {
		return "", false
	}
	if pluginSyncUnsupportedCode(response.Error.Code) || pluginSyncUnsupportedCode(response.Error.Type) {
		message := strings.TrimSpace(response.Error.Message)
		if message == "" {
			message = pluginSyncUnsupportedErrorType
		}
		return message, true
	}
	return pluginSyncUnsupportedMessage(response.Error.Message)
}

func pluginSyncUnsupportedCode(code string) bool {
	return strings.EqualFold(strings.TrimSpace(code), pluginSyncUnsupportedErrorType)
}

func pluginSyncUnsupportedMessage(message string) (string, bool) {
	message = strings.ToLower(strings.TrimSpace(message))
	message = strings.TrimSpace(strings.TrimPrefix(message, "err "))
	switch message {
	case pluginSyncUnsupportedErrorType,
		"unsupported key",
		"wrong number of arguments for 'get' command":
		return message, true
	default:
		return "", false
	}
}

func (c *Client) SetLifecycleConfig(cfg config.CredentialConcurrencyConfig) error {
	if c == nil {
		return ErrDisabled
	}
	cfg = cfg.WithDefaults()
	if errValidate := config.ValidateCredentialConcurrency(cfg); errValidate != nil {
		return fmt.Errorf("validate credential concurrency lifecycle config: %w", errValidate)
	}
	c.mu.Lock()
	c.lifecycle = cfg
	c.mu.Unlock()
	c.limiter.Store(&cfg)
	return nil
}

// LimiterConfig returns the latest immutable, validated Home limiter configuration.
func (c *Client) LimiterConfig() config.CredentialConcurrencyConfig {
	if c == nil {
		return config.CredentialConcurrencyConfig{}.WithDefaults()
	}
	if cfg := c.limiter.Load(); cfg != nil {
		return *cfg
	}
	return config.CredentialConcurrencyConfig{}.WithDefaults()
}

func (c *Client) subscriptionParameters() ([]string, time.Duration) {
	if c == nil {
		return []string{redisChannelConfig}, config.CredentialConcurrencyConfig{}.WithDefaults().CPAHeartbeatTimeout
	}
	c.mu.Lock()
	cfg := c.lifecycle.WithDefaults()
	instanceID := c.instanceID
	legacyMembership := c.legacyMembership
	testTimeout := c.testOperationTimeout
	c.mu.Unlock()

	timeout := cfg.CPAHeartbeatTimeout
	if testTimeout > 0 && cfg.LifecycleConfigRevision == 0 {
		timeout = testTimeout
	}

	args := []string{redisChannelConfig}
	if cfg.LifecycleConfigRevision > 0 {
		args = append(args, strconv.FormatInt(cfg.LifecycleConfigRevision, 10))
		if legacyMembership {
			return args, timeout
		}
		state := recoveryState(c.recoveryState.Load())
		if state == recoveryStateTakeoverEligible || state == recoveryStateSwitchingTakeover {
			args = append(args, "takeover")
		}
		args = append(args, instanceID)
	}
	return args, timeout
}

func (c *Client) markMembershipTakeoverEligible() {
	if c == nil {
		return
	}
	if !c.recoveryState.CompareAndSwap(uint32(recoveryStateStable), uint32(recoveryStateTakeoverEligible)) {
		c.recoveryState.CompareAndSwap(uint32(recoveryStateSwitching), uint32(recoveryStateSwitchingTakeover))
	}
}

func (c *Client) rebuildCommandPoolAndProbe(ctx context.Context) error {
	c.promoteSubscription()
	if errPing := c.Ping(ctx); errPing != nil {
		return errPing
	}
	c.mu.Lock()
	c.releaseReady = true
	c.recoveryState.Store(uint32(recoveryStateStable))
	c.mu.Unlock()
	return nil
}

func (c *Client) promoteSubscription() {
	if c == nil {
		return
	}
	c.mu.Lock()
	commandClient := c.cmd
	c.cmd = nil
	c.cmdOptions = nil
	c.mu.Unlock()
	if commandClient != nil {
		if errClose := commandClient.Close(); errClose != nil {
			log.Warnf("Home bootstrap command client close failed: %v", errClose)
		}
	}
}

func (c *Client) handleSubscriptionPayload(ctx context.Context, channel string, payload string, onConfig func([]byte) error) error {
	payload = strings.TrimSpace(payload)
	if payload == "" {
		return nil
	}

	switch strings.ToLower(strings.TrimSpace(channel)) {
	case redisChannelConfig:
		if onConfig == nil {
			return nil
		}
		return onConfig([]byte(payload))
	case redisChannelCluster:
		if errUpdate := c.updateClusterNodesFromPayload([]byte(payload)); errUpdate != nil {
			return errUpdate
		}
		return c.syncClusterBalance(ctx)
	case redisChannelRebalance:
		return c.executeClusterRebalance(ctx, []byte(payload))
	default:
		return nil
	}
}

// RunConfigSubscriberLifetime runs one GET, SUBSCRIBE, and receive lifetime.
// Reconnection is owned by the service so each replacement can install a new client lifetime.
func (c *Client) RunConfigSubscriberLifetime(ctx context.Context, onConfig func([]byte) error, onReady func()) error {
	if c == nil || !c.Enabled() {
		return ErrDisabled
	}
	if onConfig == nil {
		return fmt.Errorf("home config subscriber callback is nil")
	}
	if ctx == nil {
		ctx = context.Background()
	}

	c.closeBootstrapPools()
	if errEnsure := c.ensureClients(); errEnsure != nil {
		if ctx.Err() == nil {
			c.markReconnectFailure("connect")
		}
		return c.endConfigSubscriberLifetime(errEnsure)
	}

	raw, errGet := c.GetConfig(ctx)
	if errGet != nil {
		if ctx.Err() == nil {
			c.markReconnectFailure("config fetch")
		}
		return c.endConfigSubscriberLifetime(errGet)
	}
	if errApply := onConfig(raw); errApply != nil {
		return c.endConfigSubscriberLifetime(errApply)
	}

	sub, errSubClient := c.subscriptionClient()
	if errSubClient != nil {
		if ctx.Err() == nil {
			c.markReconnectFailure("subscribe client")
		}
		return c.endConfigSubscriberLifetime(errSubClient)
	}
	args, receiveTimeout := c.subscriptionParameters()
	pubsub := sub.Subscribe(ctx, args...)
	if pubsub == nil {
		if ctx.Err() == nil {
			c.markReconnectFailure("subscribe")
		}
		return c.endConfigSubscriberLifetime(ErrNotConnected)
	}

	if errACK := receiveSubscriptionACKs(ctx, pubsub, receiveTimeout, args[:1]); errACK != nil {
		if ctx.Err() == nil {
			c.markReconnectFailure("subscribe")
		}
		return c.endConfigSubscriberLifetimeWithSubscription(errACK, pubsub, "failed ACK")
	}
	heartbeatStarted := time.Now()
	// A protocol-one ACK means Home already committed this membership. Preserve it if the command probe fails.
	if len(args) > 1 {
		c.markMembershipTakeoverEligible()
	}

	if errProbe := c.rebuildCommandPoolAndProbe(ctx); errProbe != nil {
		if ctx.Err() == nil {
			c.markReconnectFailure("command probe")
		}
		return c.endConfigSubscriberLifetimeWithSubscription(errProbe, pubsub, "fresh command probe failure")
	}
	c.resetReconnectFailures()
	c.heartbeatOK.Store(true)
	// Reception must keep monitoring silence while the consumer performs migration RPCs.
	receiveCtx, cancelReceive := context.WithCancelCause(ctx)
	events := make(chan any, 16)
	receiveDone := make(chan struct{})
	c.mu.Lock()
	if c.latency == nil {
		c.latency = &clusterLatencyHistory{entries: make(map[string]clusterLatencyEntry)}
	}
	sampler := &clusterRTTSampler{history: c.latency, node: clusterNode{IP: c.homeCfg.Host, Port: c.homeCfg.Port}}
	c.mu.Unlock()
	failReceive := func(errReceive error) {
		if !errors.Is(context.Cause(receiveCtx), ErrRebalanceRequested) {
			if ctx.Err() == nil {
				if c.heartbeatOK.Load() {
					c.markMembershipTakeoverEligible()
				}
				c.releaseFenced.Store(true)
			}
			// Graceful cancellation stops new work but keeps final releases available until Close.
			c.dispatchFenced.Store(true)
		}
		c.heartbeatOK.Store(false)
		cancelReceive(errReceive)
	}
	stopReceive := context.AfterFunc(receiveCtx, func() { _ = pubsub.Close() })
	latencyDone := make(chan struct{})
	go func() {
		defer close(latencyDone)
		if errLatency := c.runClusterLatency(receiveCtx, pubsub, sampler); errLatency != nil {
			failReceive(errLatency)
		}
	}()
	go func() {
		defer close(receiveDone)
		lastHeartbeat := heartbeatStarted
		for {
			_, timeout := c.subscriptionParameters()
			event, errReceive := receiveSubscriptionEvent(receiveCtx, pubsub, sampler, lastHeartbeat.Add(timeout))
			if errReceive == nil {
				// Only a persisted heartbeat PONG renews liveness. Config, cluster traffic,
				// and RTT echoes cannot extend the deadline while the consumer is busy.
				if pong, ok := event.(*redis.Pong); ok && pong != nil && pong.Payload == "" {
					lastHeartbeat = time.Now()
				}
				select {
				case events <- event:
					continue
				case <-receiveCtx.Done():
					errReceive = context.Cause(receiveCtx)
				default:
					// Bounded backpressure fails the lifetime rather than dropping control messages.
					errReceive = fmt.Errorf("Home subscription consumer backlog exceeded")
				}
			}
			failReceive(errReceive)
			return
		}
	}()
	defer func() {
		cancelReceive(context.Canceled)
		stopReceive()
		_ = pubsub.Close()
		<-receiveDone
		<-latencyDone
		sampler.expire(time.Now(), true)
	}()
	endReceiveFailure := func(err error) error {
		if cause := context.Cause(receiveCtx); cause != nil {
			err = cause
		}
		if ctx.Err() == nil {
			// Processing failures must preserve the committed membership before cleanup clears heartbeatOK.
			if len(args) > 1 {
				c.markMembershipTakeoverEligible()
			}
			if isTimeoutError(err) {
				c.markSubscriptionTimeout()
			} else {
				c.markReconnectFailure("subscription")
			}
		}
		return c.endConfigSubscriberLifetimeWithSubscription(err, pubsub, "subscription failure")
	}
	if errFinish := c.finishClusterRebalance(receiveCtx); errFinish != nil || receiveCtx.Err() != nil {
		return endReceiveFailure(errFinish)
	}
	if onReady != nil {
		onReady()
	}

	for {
		var event any
		select {
		case <-receiveCtx.Done():
			return endReceiveFailure(context.Cause(receiveCtx))
		case event = <-events:
		}
		if receiveCtx.Err() != nil {
			return endReceiveFailure(context.Cause(receiveCtx))
		}
		switch msg := event.(type) {
		case *redis.Message:
			if msg == nil {
				continue
			}
			if errApply := c.handleSubscriptionPayload(receiveCtx, msg.Channel, msg.Payload, onConfig); errApply != nil {
				if receiveCtx.Err() != nil {
					return endReceiveFailure(errApply)
				}
				if errors.Is(errApply, ErrRebalanceRequested) {
					cancelReceive(ErrRebalanceRequested)
					return c.endConfigSubscriberLifetimeWithSubscription(errApply, pubsub, "planned rebalance")
				}
				if strings.EqualFold(strings.TrimSpace(msg.Channel), redisChannelCluster) || strings.EqualFold(strings.TrimSpace(msg.Channel), redisChannelRebalance) {
					return endReceiveFailure(errApply)
				}
				log.Warn("failed to apply config update from home control center, ignoring")
			}
		case *redis.Pong:
			if msg != nil && msg.Payload == "" {
				c.resetReconnectFailures()
			}
		default:
			log.Debugf("home subscription returned unsupported message type %T", event)
		}
	}
}

// receiveSubscriptionEvent preserves the absolute heartbeat deadline across ordinary messages.
func receiveSubscriptionEvent(ctx context.Context, pubsub *redis.PubSub, sampler *clusterRTTSampler, deadline time.Time) (any, error) {
	for {
		remaining := time.Until(deadline)
		if remaining <= 0 {
			return nil, os.ErrDeadlineExceeded
		}
		event, errReceive := pubsub.ReceiveTimeout(ctx, remaining)
		if errReceive != nil {
			return nil, errReceive
		}
		if !time.Now().Before(deadline) {
			return nil, os.ErrDeadlineExceeded
		}
		if _, ok := event.(*redis.Subscription); ok {
			return nil, fmt.Errorf("unexpected Home subscription ACK after initialization")
		}
		if pong, ok := event.(*redis.Pong); ok && pong != nil && strings.HasPrefix(pong.Payload, "home-rtt/") {
			// Timestamp before consumer callbacks, and also exclude stale or duplicate telemetry from liveness.
			sampler.pong(pong.Payload, time.Now())
			continue
		}
		return event, nil
	}
}

func receiveSubscriptionACKs(ctx context.Context, pubsub *redis.PubSub, receiveTimeout time.Duration, channels []string) error {
	if pubsub == nil || len(channels) == 0 {
		return fmt.Errorf("Home subscription ACK is missing")
	}
	for index, channel := range channels {
		event, errReceive := pubsub.ReceiveTimeout(ctx, receiveTimeout)
		if errReceive != nil {
			return errReceive
		}
		ack, ok := event.(*redis.Subscription)
		if !ok || ack == nil || ack.Kind != "subscribe" || ack.Channel != channel || ack.Count != index+1 {
			return fmt.Errorf("invalid Home subscription ACK")
		}
	}
	return nil
}

func (c *Client) endConfigSubscriberLifetime(err error) error {
	c.heartbeatOK.Store(false)
	c.failClusterRebalance()
	if !c.managedLifetime() {
		c.Close()
	}
	return err
}

func (c *Client) endConfigSubscriberLifetimeWithSubscription(err error, subscription subscriptionCloser, reason string) error {
	c.heartbeatOK.Store(false)
	if !errors.Is(err, ErrRebalanceRequested) {
		c.failClusterRebalance()
	}
	if subscription != nil {
		if errClose := subscription.Close(); errClose != nil {
			log.Debugf("Home subscription close after %s: %v", reason, errClose)
		}
	}
	if !c.managedLifetime() {
		c.Close()
	}
	return err
}

// StartConfigSubscriber is retained for callers that do not need the lifetime error.
func (c *Client) StartConfigSubscriber(ctx context.Context, onConfig func([]byte) error) {
	if errRun := c.RunConfigSubscriberLifetime(ctx, onConfig, nil); errRun != nil && !errors.Is(errRun, context.Canceled) {
		log.Warnf("Home config subscription lifetime ended: %v", errRun)
	}
}

func isTimeoutError(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	var netErr net.Error
	return errors.As(err, &netErr) && netErr.Timeout()
}

func sleepWithContext(ctx context.Context, d time.Duration) {
	if d <= 0 {
		return
	}
	timer := time.NewTimer(d)
	defer timer.Stop()
	if ctx == nil {
		<-timer.C
		return
	}
	select {
	case <-ctx.Done():
		return
	case <-timer.C:
		return
	}
}