"""McpServer Group Aggregate - manages a group of mcp_servers with load balancing. A McpServerGroup is an aggregate root that manages multiple McpServer instances as a single logical unit with automatic load balancing and failover. """ from collections.abc import Callable, Mapping from dataclasses import dataclass, field import hashlib import time from typing import Any from ...errors import bounded_error_type from ...lock_hierarchy import LockLevel, TrackedLock from ...logging_config import get_logger from ..events import CircuitBreakerStateChanged, DomainEvent from ..exceptions import McpServerStartError, CannotStartMcpServerError from ..value_objects import GroupId, GroupState, LoadBalancerStrategy, MemberPriority, MemberWeight, McpServerState from .aggregate import AggregateRoot from .circuit_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitState from .load_balancer import LoadBalancer from .mcp_server import DEAD_NOT_ROUTED_BY_GROUPS, McpServer logger = get_logger(__name__) @dataclass(frozen=True) class CanaryPolicy: """Per-tenant routing policy for a group: explicit pins + a sticky canary split. Resolution order (see :meth:`resolve`): an explicit per-tenant pin wins; otherwise a deterministic, cross-process-stable split routes ``split_pct`` percent of tenants (bucketed by a SHA-256 of ``tenant_id``) to ``canary_member``; otherwise None, meaning "use the load-balancer strategy". """ canary_member: str = "" split_pct: int = 0 pinned_tenants: Mapping[str, str] = field(default_factory=dict) def resolve(self, tenant_id: str) -> str | None: """Return the member id this tenant should route to, or None for the LB.""" pinned = self.pinned_tenants.get(tenant_id) if pinned: return pinned if self.canary_member and self.split_pct > 0: bucket = int(hashlib.sha256(tenant_id.encode()).hexdigest(), 16) % 100 if bucket < self.split_pct: return self.canary_member return None # --- Group-specific Domain Events --- @dataclass class GroupCreated(DomainEvent): """Published when a mcp_server group is created.""" group_id: str strategy: str min_healthy: int @dataclass class GroupMemberAdded(DomainEvent): """Published when a member is added to a group.""" group_id: str member_id: str weight: int priority: int @dataclass class GroupMemberRemoved(DomainEvent): """Published when a member is removed from a group.""" group_id: str member_id: str @dataclass class GroupMemberHealthChanged(DomainEvent): """Published when a member's rotation status changes.""" group_id: str member_id: str in_rotation: bool reason: str = "" @dataclass class GroupStateChanged(DomainEvent): """Published when group state transitions. `healthy_count` and `members_in_rotation_count` mean what they mean on every status surface (#1356). """ group_id: str old_state: str new_state: str healthy_count: int total_count: int members_in_rotation_count: int = 0 @dataclass class GroupCircuitOpened(DomainEvent): """Published when group circuit breaker opens.""" group_id: str failure_count: int @dataclass class GroupCircuitClosed(DomainEvent): """Published when group circuit breaker closes.""" group_id: str @dataclass class GroupUpdated(DomainEvent): """Published when a group configuration is updated.""" group_id: str @dataclass class GroupDeleted(DomainEvent): """Published when a group is deleted.""" group_id: str # --- Group Member --- @dataclass class GroupMember: """A member of a mcp_server group.""" mcp_server: McpServer weight: int = 1 priority: int = 1 in_rotation: bool = False # Currently accepting traffic consecutive_failures: int = 0 consecutive_successes: int = 0 last_selected_at: float = 0.0 @property def id(self) -> str: """Get member's mcp_server ID as string.""" # mcp_server.id returns str (from McpServer class) return str(self.mcp_server.id) # --- McpServer Group Aggregate --- class McpServerGroup(AggregateRoot): """ Aggregate root for a group of load-balanced mcp_servers. Responsibilities: - Manage member lifecycle - Load balancing decisions - Group-level health tracking - Circuit breaker for the entire group Thread-safety: - All public methods are thread-safe - Internal lock prevents concurrent modification """ def __init__( self, group_id: str, strategy: LoadBalancerStrategy = LoadBalancerStrategy.ROUND_ROBIN, min_healthy: int = 1, auto_start: bool = True, unhealthy_threshold: int = 2, healthy_threshold: int = 1, circuit_failure_threshold: int = 10, description: str | None = None, ): """ Initialize a mcp_server group. Args: group_id: Unique identifier for the group strategy: Load balancing strategy min_healthy: Minimum healthy members for HEALTHY state auto_start: Automatically start members when added unhealthy_threshold: Failures before removing from rotation healthy_threshold: Successes before adding back to rotation circuit_failure_threshold: Failures in a row before circuit opens description: Human-readable description """ super().__init__() # Identity self._id = GroupId(group_id) self._description = description # Configuration self._strategy = strategy self._min_healthy = max(1, min_healthy) self._auto_start = auto_start self._unhealthy_threshold = max(1, unhealthy_threshold) self._healthy_threshold = max(1, healthy_threshold) # State self._state = GroupState.INACTIVE self._members: dict[str, GroupMember] = {} self._load_balancer = LoadBalancer(strategy) # Optional per-tenant canary/version routing policy (config-driven). self._canary: CanaryPolicy | None = None # Told whether the circuit is open, after every transition (#1357). self._circuit_listener: Callable[[bool], None] | None = None # Circuit breaker (extracted for SRP). No reset timeout: the breaker's # default is left in place and never read. The breaker consults it only # in `allow_request()`, which the group does not call, so an open group # circuit never half-opens on a timer. It closes through # `_maybe_close_circuit()` once `min_healthy` members are back in # rotation. A group option for the timeout was accepted and ignored # until #1398 removed it. self._circuit_breaker = CircuitBreaker(CircuitBreakerConfig(failure_threshold=circuit_failure_threshold)) self._circuit_breaker._on_state_change = self._on_circuit_breaker_state_change # Threading # Lock hierarchy level: PROVIDER_GROUP (11) # Safe to acquire after: PROVIDER (but avoid holding both) # Safe to acquire before: EVENT_BUS, EVENT_STORE, STDIO_CLIENT self._lock = self._create_lock(group_id) self._record_event( GroupCreated( group_id=group_id, strategy=strategy.value, min_healthy=min_healthy, ) ) @staticmethod def _create_lock(group_id: str) -> TrackedLock: """Create the group's lock, registered in the global ordering. See McpServer._create_lock: the tracking is not optional, and the `except ImportError` this replaces could never fire. """ return TrackedLock(LockLevel.PROVIDER_GROUP, f"McpServerGroup:{group_id}") def _on_circuit_breaker_state_change(self, old_state: Any, new_state: Any) -> None: """Emit a domain event when the circuit breaker transitions between states. Called by CircuitBreaker._fire_state_change outside of the breaker's lock. Every transition the group makes is under its own lock, so the listener hears them in the order they happened. Args: old_state: Previous CircuitState enum value. new_state: New CircuitState enum value. """ self._record_event( CircuitBreakerStateChanged( mcp_server_id=self._id.value, old_state=old_state.value, new_state=new_state.value, ) ) if self._circuit_listener is not None: self._circuit_listener(new_state is CircuitState.OPEN) def observe_circuit(self, listener: Callable[[bool], None]) -> None: """Tell ``listener`` whether the circuit is open: now, and after every transition. Every transition goes through the breaker's state-change callback: opening on failures in a row, closing at ``min_healthy``, ``rebalance()``'s reset. So the listener hears each one, on the path that made it (#1357). The group's own events cannot do that: nothing drains them on the call path or in the health checks, so they never reach the bus from either. Under the group's lock, so the state handed over now cannot land after a transition that followed it. """ with self._lock: self._circuit_listener = listener listener(self._circuit_breaker.is_open) # --- Properties --- @property def id(self) -> str: """Get group ID.""" return self._id.value @property def description(self) -> str | None: """Get group description.""" return self._description @property def state(self) -> GroupState: """Get current group state.""" with self._lock: return self._state @property def strategy(self) -> LoadBalancerStrategy: """Get load balancing strategy.""" return self._strategy @property def healthy_count(self) -> int: """Members that are `ready` and in rotation. Reported, never decided on (#1356). It used to count every member in rotation that was not DEAD, `cold` ones included, so a group could report three healthy members with its circuit open and nothing serving. The group's decisions still count that way, through `_live_rotation_count()`; `members_in_rotation_count` reports rotation size. """ with self._lock: return sum( 1 for m in self._members.values() if m.in_rotation and m.mcp_server.state_snapshot is McpServerState.READY ) @property def members_in_rotation_count(self) -> int: """Members in rotation, whatever their state: the members `members_in_rotation` names.""" with self._lock: return sum(1 for m in self._members.values() if m.in_rotation) def _live_rotation_count(self) -> int: """Members in rotation that are not DEAD: what the group's decisions count. `is_available`, the group state and the `min_healthy` rule that closes an open circuit count these. A `cold` member counts: a group starts its members lazily, and the next call through it starts one. A member whose process crashed stays in rotation, so a call restarts it, but does not count until it is back (#1361). Until #1356 this was `healthy_count`. """ with self._lock: return sum( 1 for m in self._members.values() if m.in_rotation and m.mcp_server.state_snapshot is not McpServerState.DEAD ) @property def total_count(self) -> int: """Total number of members in the group.""" with self._lock: return len(self._members) @property def is_available(self) -> bool: """Can the group accept requests?""" with self._lock: return ( not self._circuit_breaker.is_open and self._state.can_accept_requests and self._live_rotation_count() >= 1 ) @property def circuit_open(self) -> bool: """Is the circuit breaker open?""" return self._circuit_breaker.is_open @property def members(self) -> list[GroupMember]: """Get list of all members.""" with self._lock: return list(self._members.values()) # --- Member Management --- def add_member( self, mcp_server: McpServer, weight: int = 1, priority: int = 1, ) -> None: """ Add a mcp_server to the group. Uses two-phase lock pattern to avoid lock hierarchy violation: Phase 1 (locked): Register member, emit event. Phase 2 (unlocked): Call ensure_ready() if auto_start (mcp_server I/O). Phase 3 (locked): Update rotation state if member still exists. Args: mcp_server: McpServer instance to add weight: Load balancing weight (higher = more traffic) priority: Priority for priority-based selection (lower = higher priority) Raises: ValueError: If member already exists in group """ need_start = False member: GroupMember | None = None member_id: str = "" # Phase 1: Register member under lock with self._lock: member_id = str(mcp_server.id) if member_id in self._members: raise ValueError(f"Member {member_id} already in group {self.id}") # Validate weight and priority validated_weight = MemberWeight(weight) validated_priority = MemberPriority(priority) member = GroupMember( mcp_server=mcp_server, weight=validated_weight.value, priority=validated_priority.value, ) self._members[member_id] = member self._record_event( GroupMemberAdded( group_id=self.id, member_id=member_id, weight=weight, priority=priority, ) ) logger.info(f"Added member {member_id} to group {self.id} (weight={weight}, priority={priority})") need_start = self._auto_start # Phase 2: McpServer I/O outside lock (avoids level-11-holds-level-10) if need_start: self._try_start_member_unlocked(member, member_id) def remove_member(self, member_id: str) -> bool: """ Remove a mcp_server from the group. Args: member_id: ID of the member to remove Returns: True if member was removed, False if not found """ with self._lock: member = self._members.pop(member_id, None) if member: member.in_rotation = False self._update_state() self._record_event( GroupMemberRemoved( group_id=self.id, member_id=member_id, ) ) logger.info(f"Removed member {member_id} from group {self.id}") return True return False def get_member(self, member_id: str) -> GroupMember | None: """Get a member by ID.""" with self._lock: return self._members.get(member_id) def _try_start_member_unlocked(self, member: GroupMember, member_id: str) -> bool: """Try to start a member with two-phase lock pattern. Phase 1 (unlocked): Call ensure_ready() -- mcp_server I/O outside group lock to respect lock hierarchy (McpServer level 10 < McpServerGroup level 11). Phase 2 (locked): Re-acquire lock to update rotation state. Handles the case where the member was removed by another thread during Phase 1. Args: member: The group member to start. member_id: The member's mcp_server ID string. Returns: True if member started and added to rotation. """ # Phase 1: McpServer I/O outside group lock (respects McpServer level 10 < McpServerGroup level 11) try: member.mcp_server.ensure_ready() except (McpServerStartError, CannotStartMcpServerError) as e: # The type only: a start failure's text can carry what the upstream printed. logger.warning( "group_member_start_failed", group_id=self.id, mcp_server_id=member_id, error_type=bounded_error_type(type(e).__qualname__), ) with self._lock: # Only update if member still exists (may have been removed) if member_id in self._members: self._members[member_id].in_rotation = False return False # Read mcp_server state outside group lock to avoid lock order violation # (McpServer._lock level 10 must not be acquired while holding McpServerGroup._lock level 11) mcp_server_state = member.mcp_server.state # Phase 2: Re-acquire group lock to update rotation state with self._lock: # Member may have been removed by another thread during Phase 1 if member_id not in self._members: return False current_member = self._members[member_id] if mcp_server_state == McpServerState.READY: current_member.in_rotation = True current_member.consecutive_failures = 0 current_member.consecutive_successes = 1 self._update_state() self._record_event( GroupMemberHealthChanged( group_id=self.id, member_id=member_id, in_rotation=True, reason="started", ) ) logger.info(f"Member {member_id} started and added to rotation") return True return False # --- Load Balancing --- def set_canary_policy(self, policy: "CanaryPolicy | None") -> None: """Set (or clear) the per-tenant canary/version routing policy.""" with self._lock: self._canary = policy def select_member(self) -> McpServer | None: """Select a member for the next request using the load-balancer strategy.""" return self.select_member_for(None) def select_member_for(self, tenant_id: str | None) -> McpServer | None: """Select a member, applying per-tenant canary routing when configured. Resolution: an explicit per-tenant pin, then a sticky canary split (deterministic by ``tenant_id``), then the load-balancer strategy. A pinned/canary target that is not in rotation falls back to the LB pick, so routing never sends traffic to an out-of-rotation member. The group-level circuit breaker never vetoes a healthy remaining member: as long as at least one member is still ``in_rotation`` it is selectable, even while the group CB is open. The CB only blocks when no member remains in rotation -- that is when the group is genuinely down, which is the breaker's real purpose. This prevents a primary eviction (which opens the group CB) from taking down an otherwise-healthy backup. A member Hangar gave up on, or one a capability block stopped, is never selected, in rotation or not: a group choosing it would make the group the thing that revives it (#1361). A member whose process crashed is selected, as a COLD one is, and selecting it restarts it. Returns: Selected mcp_server or None if no healthy members available. """ with self._lock: self._check_circuit_recovery() available = [m for m in self._members.values() if self._selectable(m)] if not available: # No member remains in rotation: honor the group circuit # breaker and reject rather than hammer a genuinely-down group. return None # Per-tenant canary/version routing (explicit pin or sticky split). if tenant_id is not None and self._canary is not None: target_id = self._canary.resolve(tenant_id) if target_id is not None: target = self._members.get(target_id) if target is not None and self._selectable(target): target.last_selected_at = time.time() return target.mcp_server logger.warning( "canary_target_unavailable_fallback_lb", group_id=str(self.id), tenant_id=tenant_id, target=target_id, ) selected = self._load_balancer.select(available) if selected: selected.last_selected_at = time.time() return selected.mcp_server return None @staticmethod def _selectable(member: GroupMember) -> bool: """In rotation, and not dead for a reason a group does not route to. A snapshot: the member's lock sits below this one's. """ return member.in_rotation and member.mcp_server.dead_reason_snapshot not in DEAD_NOT_ROUTED_BY_GROUPS def _check_circuit_recovery(self) -> None: """Check if circuit just recovered and emit event.""" if not self._circuit_breaker.is_open and self._state == GroupState.DEGRADED: self._record_event(GroupCircuitClosed(group_id=self.id)) logger.info(f"Circuit breaker closed for group {self.id}") self._update_state() # --- Health Reporting --- def report_success(self, member_id: str) -> None: """ Report successful invocation for a member. A success ends the run of failures the circuit counts, so `circuit_failure_threshold` means failures in a row. An open circuit is the exception: it closes only through `_maybe_close_circuit()`, once `min_healthy` members are in rotation. Args: member_id: ID of the member that succeeded """ with self._lock: member = self._members.get(member_id) if not member: return member.consecutive_failures = 0 member.consecutive_successes += 1 self._maybe_add_to_rotation(member, member_id) if self._circuit_breaker.is_open: self._maybe_close_circuit() else: self._end_failure_run() # A crashed member back up counts as healthy again (#1361). self._update_state() def _end_failure_run(self) -> None: """Reset the circuit's failure count on a success while it is not open. The breaker resets it in `record_success()`, and until #1390 the group never called that on a closed circuit. `report_failure()` counted every failure and only `rebalance()` reset, so the threshold counted failures over the life of the process: a group that saw one member failure a day opened its circuit on day ten. Not on an open circuit, which `record_success()` would close at once, skipping the `min_healthy` rule in `_maybe_close_circuit()`. HALF_OPEN goes the breaker's way too, and this success would close it. The group never gets there: the breaker half-opens only in `allow_request()`, which the group does not call, and no saved breaker is restored into a group at startup (#1388). """ self._circuit_breaker.record_success() def _maybe_close_circuit(self) -> None: """Close an open circuit once `min_healthy` members are in rotation. Nothing else closed it. The breaker leaves OPEN through `allow_request()` or `record_success()`, and the group called neither on an open circuit, so a group whose circuit had opened reported `circuit_open: True` and `degraded` after its members were back, until someone ran `rebalance()` (#1355). A success, with enough members in rotation to call the group healthy, is what the breaker was waiting for. """ if not self._circuit_breaker.is_open: return in_rotation = self._live_rotation_count() if in_rotation < self._min_healthy: return self._circuit_breaker.record_success() # OPEN -> CLOSED self._record_event(GroupCircuitClosed(group_id=self.id)) logger.info(f"Circuit breaker closed for group {self.id}: {in_rotation} member(s) in rotation") self._update_state() def _maybe_add_to_rotation(self, member: GroupMember, member_id: str) -> None: """Add member back to rotation if healthy threshold reached.""" if member.in_rotation: return if member.mcp_server.state_snapshot != McpServerState.READY: return if member.consecutive_successes < self._healthy_threshold: return member.in_rotation = True self._record_event( GroupMemberHealthChanged( group_id=self.id, member_id=member_id, in_rotation=True, reason="healthy_threshold_reached", ) ) self._update_state() logger.info(f"Member {member_id} added back to rotation") def report_failure(self, member_id: str) -> None: """ Report failed invocation for a member. Args: member_id: ID of the member that failed """ with self._lock: member = self._members.get(member_id) if not member: return member.consecutive_failures += 1 member.consecutive_successes = 0 self._maybe_remove_from_rotation(member, member_id) self._maybe_open_circuit() self._update_state() def report_member_dead(self, member_id: str) -> None: """A member went DEAD: keep rotation and group state true (#1361). A member Hangar gave up on, or one a capability block stopped, leaves rotation: a call through the group revives neither. A start that succeeds brings it back, through `report_success`. A member whose process crashed stays in rotation, so the next call through the group selects and restarts it; neither `healthy_count` nor the group's decisions count it meanwhile. """ with self._lock: member = self._members.get(member_id) if not member: return reason = member.mcp_server.dead_reason_snapshot if member.in_rotation and reason in DEAD_NOT_ROUTED_BY_GROUPS: member.in_rotation = False member.consecutive_successes = 0 self._record_event( GroupMemberHealthChanged( group_id=self.id, member_id=member_id, in_rotation=False, reason=str(reason), ) ) logger.info(f"Member {member_id} removed from rotation: dead, {reason}") self._update_state() def _maybe_remove_from_rotation(self, member: GroupMember, member_id: str) -> None: """Remove member from rotation if unhealthy threshold reached.""" if member.consecutive_failures < self._unhealthy_threshold: return if not member.in_rotation: return member.in_rotation = False self._record_event( GroupMemberHealthChanged( group_id=self.id, member_id=member_id, in_rotation=False, reason="unhealthy_threshold_reached", ) ) logger.info(f"Member {member_id} removed from rotation after {member.consecutive_failures} failures") def _maybe_open_circuit(self) -> None: """Open circuit breaker if failure threshold reached.""" circuit_just_opened = self._circuit_breaker.record_failure() if not circuit_just_opened: return self._record_event( GroupCircuitOpened( group_id=self.id, failure_count=self._circuit_breaker.failure_count, ) ) logger.warning( f"Circuit breaker opened for group {self.id} after {self._circuit_breaker.failure_count} failures" ) # --- State Management --- def _update_state(self) -> None: """Update group state from the members in rotation that are not DEAD.""" old_state = self._state in_rotation = self._live_rotation_count() total = len(self._members) if self._circuit_breaker.is_open: new_state = GroupState.DEGRADED elif in_rotation == 0: new_state = GroupState.INACTIVE elif in_rotation < self._min_healthy: new_state = GroupState.PARTIAL else: new_state = GroupState.HEALTHY if new_state != old_state: self._state = new_state healthy = self.healthy_count self._record_event( GroupStateChanged( group_id=self.id, old_state=old_state.value, new_state=new_state.value, healthy_count=healthy, total_count=total, members_in_rotation_count=self.members_in_rotation_count, ) ) logger.info( f"Group {self.id} state: {old_state.value} -> {new_state.value} " f"(healthy={healthy}/{total}, not dead in rotation={in_rotation})" ) def rebalance(self) -> None: """ Manually trigger rebalancing. Re-evaluates health of all members and updates rotation. """ with self._lock: for member in self._members.values(): if member.mcp_server.state_snapshot == McpServerState.READY: if not member.in_rotation: member.in_rotation = True member.consecutive_failures = 0 self._record_event( GroupMemberHealthChanged( group_id=self.id, member_id=member.id, in_rotation=True, reason="rebalance", ) ) else: if member.in_rotation: member.in_rotation = False self._record_event( GroupMemberHealthChanged( group_id=self.id, member_id=member.id, in_rotation=False, reason="rebalance", ) ) # Reset load balancer state self._load_balancer.reset() # Reset circuit breaker was_open = self._circuit_breaker.is_open self._circuit_breaker.reset() if was_open: self._record_event(GroupCircuitClosed(group_id=self.id)) self._update_state() logger.info( f"Group {self.id} rebalanced: {self.healthy_count} healthy, " f"{self.members_in_rotation_count} in rotation" ) # --- Lifecycle --- def start_all(self) -> int: """Start all members. Uses two-phase lock pattern to avoid lock hierarchy violation: Phase 1 (locked): Snapshot member references. Phase 2 (unlocked): Call ensure_ready() on each member. Returns: Number of members successfully started. """ # Phase 1: Snapshot members under lock with self._lock: members_snapshot = [(mid, m) for mid, m in self._members.items()] # Phase 2: Start each member outside lock started = 0 for member_id, member in members_snapshot: if self._try_start_member_unlocked(member, member_id): started += 1 return started def stop_all(self) -> None: """Stop all members. Uses two-phase lock pattern to avoid lock hierarchy violation: Phase 1 (locked): Snapshot member references. Phase 2 (unlocked): Call shutdown() on each member. Phase 3 (locked): Update rotation state. """ # Phase 1: Snapshot members under lock with self._lock: members_snapshot = [(mid, m) for mid, m in self._members.items()] # Phase 2: Shutdown each member outside lock for member_id, member in members_snapshot: try: member.mcp_server.shutdown() except Exception as e: # noqa: BLE001 -- fault-barrier: shutdown of one member must not prevent others logger.warning(f"Failed to stop member {member_id}: {e}") # Phase 3: Update state under lock with self._lock: for member_id, _ in members_snapshot: if member_id in self._members: self._members[member_id].in_rotation = False self._update_state() def shutdown(self) -> None: """Shutdown the group and all members.""" self.stop_all() logger.info(f"Group {self.id} shutdown complete") # --- Tools Access --- def get_tools(self) -> list[Any]: """ Get tools from a healthy member. Returns tools from the first healthy member, as all members should have the same tools. """ with self._lock: for member in self._members.values(): if member.in_rotation and member.mcp_server.state_snapshot == McpServerState.READY: return list(member.mcp_server.tools) return [] def get_tool_names(self) -> list[str]: """Get list of tool names from a healthy member.""" with self._lock: for member in self._members.values(): if member.in_rotation and member.mcp_server.state_snapshot == McpServerState.READY: return member.mcp_server.get_tool_names() return [] # --- Configuration Update --- def update( self, strategy: str | None = None, description: str | None = None, min_healthy: int | None = None, ) -> None: """Update mutable configuration fields. Only non-None arguments are applied. Records GroupUpdated event. Args: strategy: New load balancing strategy string (optional). description: New human-readable description (optional). min_healthy: New minimum healthy member count, min 1 (optional). """ with self._lock: if strategy is not None: self._strategy = LoadBalancerStrategy(strategy) if description is not None: self._description = description if min_healthy is not None: self._min_healthy = max(1, min_healthy) self._record_event(GroupUpdated(group_id=self.id)) # --- Serialization --- def to_config_dict(self) -> dict[str, Any]: """Return YAML-compatible config spec dict. Includes mode="group", strategy, min_healthy, auto_start, description (if set), and members list. Returns: Dict with all fields required to reconstruct this group from config. """ with self._lock: spec: dict[str, Any] = { "mode": "group", "strategy": self._strategy.value, "min_healthy": self._min_healthy, "auto_start": self._auto_start, "members": [{"id": m.id, "weight": m.weight, "priority": m.priority} for m in self._members.values()], } if self._description: spec["description"] = self._description return spec def to_status_dict(self) -> dict[str, Any]: """The group's status, read under its lock: one instant. `GET /api/groups/{id}`, `hangar_details`, `hangar_group_list` and `hangar_list` return this dict as it is; `hangar_status` and `hangar_health` report its counts. `healthy_count` counts members that are ready and in rotation, `members_in_rotation_count` members in rotation whatever their state (#1356). """ with self._lock: return { "group_id": self.id, "description": self._description, "state": self._state.value, "strategy": self._strategy.value, "min_healthy": self._min_healthy, "healthy_count": self.healthy_count, "members_in_rotation_count": self.members_in_rotation_count, "total_members": len(self._members), "is_available": self.is_available, "circuit_open": self._circuit_breaker.is_open, "members": [ { "id": m.id, "state": m.mcp_server.state_snapshot.value, "in_rotation": m.in_rotation, "weight": m.weight, "priority": m.priority, "consecutive_failures": m.consecutive_failures, } for m in self._members.values() ], } # legacy aliases ProviderGroup = McpServerGroup