"""Parse canonical role requirements declared in phase profiles.""" from __future__ import annotations from dataclasses import dataclass from pathlib import Path from .domain.role import RoleCatalogError, normalize_role, role_for_duty class RoleProfileError(ValueError): """Raised when a profile's structured role declaration is invalid.""" @dataclass(frozen=True) class RoleRequirement: role: str min_count: int recommended_count: int max_count: int duty: str dynamic: bool = False def signature(self) -> tuple[str, int, int, int, str, bool]: return ( self.role, self.min_count, self.recommended_count, self.max_count, self.duty, self.dynamic, ) @dataclass(frozen=True) class RoleProfile: roles: tuple[RoleRequirement, ...] def role_signature(self) -> tuple[tuple[str, int, int, int, str, bool], ...]: return tuple(requirement.signature() for requirement in self.roles) _REQUIRED_FIELDS = frozenset({"role", "min", "recommended", "max", "duty"}) _ALLOWED_FIELDS = _REQUIRED_FIELDS | {"dynamic"} def load_role_profile(path: Path) -> RoleProfile: """Load and validate the one fenced YAML ``roles:`` declaration in a profile.""" profile_path = Path(path) try: text = profile_path.read_text(encoding="utf-8") except OSError as exc: raise RoleProfileError(f"cannot read role profile: {profile_path}") from exc raw_roles, errors = _parse_roles_block(text) requirements = tuple( _validate_requirement(raw, index, errors) for index, raw in enumerate(raw_roles, 1) ) _validate_static_role_duplicates(requirements, errors) if errors: raise RoleProfileError("invalid role profile:\n- " + "\n- ".join(errors)) return RoleProfile(requirements) def _parse_roles_block(text: str) -> tuple[list[dict[str, str]], list[str]]: blocks = _yaml_blocks_with_roles(text) if len(blocks) != 1: return [], ["profile must declare exactly one fenced YAML roles block"] lines = blocks[0] if lines == ["roles: []"]: return [], [] if not lines or lines[0] != "roles:": return [], ["roles block must start with roles:"] roles: list[dict[str, str]] = [] errors: list[str] = [] current: dict[str, str] | None = None for line_number, line in enumerate(lines[1:], 2): if not line.strip(): continue if line.startswith(" - "): if current is not None: roles.append(current) current = {} _add_yaml_field(current, line[4:], line_number, errors) continue if line.startswith(" ") and current is not None: _add_yaml_field(current, line[4:], line_number, errors) continue errors.append(f"roles block line {line_number} is not a role field") if current is not None: roles.append(current) elif not errors: errors.append("roles block must contain a role or use roles: []") return roles, errors def _yaml_blocks_with_roles(text: str) -> list[list[str]]: blocks: list[list[str]] = [] current: list[str] | None = None for line in text.splitlines(): if line.strip() == "```yaml": current = [] continue if line.strip() == "```" and current is not None: if any(item.strip().startswith("roles:") for item in current): blocks.append(current) current = None continue if current is not None: current.append(line) return blocks def _add_yaml_field( fields: dict[str, str], line: str, line_number: int, errors: list[str], ) -> None: if ":" not in line: errors.append(f"roles block line {line_number} must contain key: value") return key, value = (part.strip() for part in line.split(":", 1)) if not key or not value: errors.append(f"roles block line {line_number} must contain key: value") return if key in fields: errors.append(f"roles block line {line_number} repeats {key}") return fields[key] = value def _validate_requirement( raw: dict[str, str], index: int, errors: list[str], ) -> RoleRequirement: prefix = f"role requirement {index}" for field in sorted(_REQUIRED_FIELDS - raw.keys()): errors.append(f"{prefix} is missing {field}") for field in sorted(raw.keys() - _ALLOWED_FIELDS): errors.append(f"{prefix} has unknown field {field}") role = _validate_role(raw.get("role", ""), prefix, errors) min_count = _validate_non_negative_integer( raw.get("min", ""), "min", prefix, errors, ) recommended_count = _validate_non_negative_integer( raw.get("recommended", ""), "recommended", prefix, errors, ) max_count = _validate_non_negative_integer( raw.get("max", ""), "max", prefix, errors, ) duty = _validate_duty(raw.get("duty", ""), role, prefix, errors) dynamic = _validate_dynamic(raw.get("dynamic", "false"), prefix, errors) if not (min_count <= recommended_count <= max_count): errors.append(f"{prefix} min must be <= recommended <= max") if dynamic and (min_count != 0 or recommended_count != 0 or max_count != 0): errors.append( f"{prefix} dynamic roles must have min, recommended, and max of zero" ) return RoleRequirement( role, min_count, recommended_count, max_count, duty, dynamic, ) def _validate_role(raw: str, prefix: str, errors: list[str]) -> str: try: return normalize_role(raw) except RoleCatalogError as exc: errors.append(f"{prefix} {exc}") return raw def _validate_duty(raw: str, role: str, prefix: str, errors: list[str]) -> str: try: duty_role = role_for_duty(raw) except RoleCatalogError as exc: errors.append(f"{prefix} {exc}") return raw if duty_role != role: errors.append(f"{prefix} duty {raw} belongs to role {duty_role}, not {role}") return raw def _validate_non_negative_integer(raw: str, field: str, prefix: str, errors: list[str]) -> int: try: value = int(raw) except ValueError: errors.append(f"{prefix} {field} must be a non-negative integer") return 0 if value < 0 or str(value) != raw: errors.append(f"{prefix} {field} must be a non-negative integer") return 0 return value def _validate_dynamic(raw: str, prefix: str, errors: list[str]) -> bool: if raw == "true": return True if raw == "false": return False errors.append(f"{prefix} dynamic must be true or false") return False def _validate_static_role_duplicates( requirements: tuple[RoleRequirement, ...], errors: list[str], ) -> None: static_roles: set[str] = set() for requirement in requirements: if requirement.dynamic: continue if requirement.role in static_roles: errors.append(f"duplicate static role: {requirement.role}") static_roles.add(requirement.role)