"""Normalize canonical role selections and legacy provider CLI inputs.""" from __future__ import annotations import json from dataclasses import dataclass from types import MappingProxyType from typing import Mapping, Sequence from .domain.host import ( CurrentSessionModelAttestation, HostSessionContext, ) from .domain.role import RoleCatalogError, normalize_role from .role_requirements import RoleProfile, RoleRequirement class ModelSelectionInputError(ValueError): """Raised before prepare side effects when role selection is invalid.""" @dataclass(frozen=True) class CanonicalModelSelection: role_counts: Mapping[str, int] role_models: Mapping[str, tuple[str, ...]] provider_constraints: Mapping[str, tuple[str, ...]] legacy_model_values: Mapping[str, Mapping[str, str]] legacy_sources: tuple[str, ...] def __post_init__(self) -> None: model_values = { role: MappingProxyType(dict(values)) for role, values in self.legacy_model_values.items() } object.__setattr__( self, "role_counts", MappingProxyType(dict(self.role_counts)), ) object.__setattr__( self, "role_models", MappingProxyType(dict(self.role_models)), ) object.__setattr__( self, "provider_constraints", MappingProxyType(dict(self.provider_constraints)), ) object.__setattr__( self, "legacy_model_values", MappingProxyType(model_values), ) def normalize_model_selection_inputs( *, profile: RoleProfile, role_counts_raw: Sequence[str], role_models_raw: Sequence[str], workers: str, worker_model: str, critic: str, executor: str, lead_provider: str, lead_model: str, report_writer_provider: str, report_writer_model: str, provider_models: Mapping[str, str], ) -> CanonicalModelSelection: """Return one ordered role-selection view for canonical and legacy inputs.""" counts = _parse_role_counts(role_counts_raw) models = _parse_role_models(role_models_raw) legacy = _legacy_selection( profile=profile, workers=workers, worker_model=worker_model, critic=critic, executor=executor, lead_provider=lead_provider, lead_model=lead_model, report_writer_provider=report_writer_provider, report_writer_model=report_writer_model, provider_models=provider_models, ) # 같은 값이면 정본을 남기고 레거시를 접는다. 다르면 prepare 전에 거부. legacy = _reconcile_legacy_with_canonical(counts, models, legacy) merged_counts = {**counts, **legacy.role_counts} merged_models = {**models, **legacy.role_models} _validate_selection(profile, merged_counts, merged_models, counts) return CanonicalModelSelection( merged_counts, merged_models, legacy.provider_constraints, legacy.legacy_model_values, legacy.legacy_sources, ) def _parse_role_counts(raw_values: Sequence[str]) -> dict[str, int]: counts: dict[str, int] = {} for raw in raw_values: role, value = _split_role_value(raw, "--role-count") if role in counts: raise ModelSelectionInputError(f"duplicate --role-count role: {role}") try: count = int(value) except ValueError as exc: raise ModelSelectionInputError( f"--role-count requires role=: {raw!r}" ) from exc if count < 0 or str(count) != value: raise ModelSelectionInputError( f"--role-count requires role=: {raw!r}" ) counts[role] = count return counts def _parse_role_models(raw_values: Sequence[str]) -> dict[str, tuple[str, ...]]: grouped: dict[str, list[str]] = {} for raw in raw_values: role, model_ref = _split_role_value(raw, "--role-model") grouped.setdefault(role, []).append(model_ref) return {role: tuple(values) for role, values in grouped.items()} def _split_role_value(raw: str, option: str) -> tuple[str, str]: role_raw, separator, value = raw.partition("=") role_raw = role_raw.strip() value = value.strip() if not separator or not role_raw or not value: raise ModelSelectionInputError(f"{option} requires role=value: {raw!r}") try: role = normalize_role(role_raw) except RoleCatalogError as exc: raise ModelSelectionInputError(str(exc)) from exc return role, value def _legacy_selection( *, profile: RoleProfile, workers: str, worker_model: str, critic: str, executor: str, lead_provider: str, lead_model: str, report_writer_provider: str, report_writer_model: str, provider_models: Mapping[str, str], ) -> CanonicalModelSelection: counts: dict[str, int] = {} constraints: dict[str, tuple[str, ...]] = {} values: dict[str, dict[str, str]] = {} sources: list[str] = [] needs_initial = bool( workers.strip() or worker_model.strip() or any(model.strip() for model in provider_models.values()) ) initial = ( _initial_cross_verification_requirement(profile) if needs_initial else None ) if workers.strip() and initial is not None: providers = _legacy_worker_providers(workers) _validate_legacy_workers(initial, providers) counts[initial.role] = len(providers) constraints[initial.role] = providers report_writer = _requirement(profile, "report-writer") if report_writer is not None: counts["report-writer"] = report_writer.recommended_count sources.append("--workers") worker_values = _parse_worker_models(worker_model, provider_models) if worker_values and initial is not None: values[initial.role] = worker_values sources.append("worker-model") _add_provider_selection( constraints, values, sources, "leader", lead_provider, lead_model, "lead" ) _add_provider_selection( constraints, values, sources, "report-writer", report_writer_provider, report_writer_model, "report-writer", ) _add_simple_legacy_roles(counts, constraints, sources, critic, executor) return CanonicalModelSelection(counts, {}, constraints, values, tuple(sources)) def _initial_cross_verification_requirement(profile: RoleProfile) -> RoleRequirement: for requirement in profile.roles: if not requirement.dynamic and requirement.role in { "analyser", "designer", "planner", "verifier" }: return requirement raise ModelSelectionInputError( "profile has no initial cross-verification role for legacy --workers" ) def _legacy_worker_providers(workers: str) -> tuple[str, ...]: tokens = tuple( token.strip().lower() for token in workers.split(",") if token.strip() ) return tuple(token for token in tokens if token != "report-writer") def _validate_legacy_workers( requirement: RoleRequirement, providers: tuple[str, ...], ) -> None: # minDistinctProviders 제거. 모델 참조 중복 거부는 Task 2. if ( len(providers) < requirement.min_count or len(providers) > requirement.max_count ): raise ModelSelectionInputError( f"legacy --workers selects {len(providers)} {requirement.role} instances; " f"profile requires {requirement.min_count}..{requirement.max_count}" ) def _parse_worker_models( worker_model: str, provider_models: Mapping[str, str], ) -> dict[str, str]: values = { provider.strip().lower(): model.strip() for provider, model in provider_models.items() if model.strip() } for item in worker_model.split(","): if not item.strip(): continue provider, separator, model = item.partition("=") provider = provider.strip().lower() model = model.strip() if not separator or not provider or not model: raise ModelSelectionInputError( "--worker-model must use provider=model entries separated by commas" ) if provider in values: raise ModelSelectionInputError( f"duplicate legacy worker model provider: {provider}" ) values[provider] = model return values def _add_provider_selection( constraints: dict[str, tuple[str, ...]], values: dict[str, dict[str, str]], sources: list[str], role: str, provider_raw: str, model_raw: str, source: str, ) -> None: provider = provider_raw.strip().lower() model = model_raw.strip() if not provider and not model: return constraints[role] = (provider,) if model: values[role] = {provider: model} sources.append(source) def _add_simple_legacy_roles( counts: dict[str, int], constraints: dict[str, tuple[str, ...]], sources: list[str], critic_raw: str, executor_raw: str, ) -> None: critic = critic_raw.strip().lower() if critic: counts["critic"] = 0 if critic == "off" else 1 if critic != "off": constraints["critic"] = (critic,) sources.append("critic") executor = executor_raw.strip().lower() if executor: constraints["implementer"] = (executor,) sources.append("executor") def _reconcile_legacy_with_canonical( canonical_counts: Mapping[str, int], canonical_models: Mapping[str, tuple[str, ...]], legacy: CanonicalModelSelection, ) -> CanonicalModelSelection: """충돌하는 레거시 입력을 거부하고, 같은 값 겹침은 정본만 남긴다.""" canonical_roles = set(canonical_counts) | set(canonical_models) legacy_roles = ( set(legacy.role_counts) | set(legacy.provider_constraints) | set(legacy.legacy_model_values) ) overlap = sorted(canonical_roles & legacy_roles) if not overlap: return legacy counts = dict(legacy.role_counts) constraints = dict(legacy.provider_constraints) values = { role: dict(models) for role, models in legacy.legacy_model_values.items() } for role in overlap: if not _legacy_matches_canonical( role, canonical_counts, canonical_models, legacy, ): raise ModelSelectionInputError( "canonical and legacy model selections conflict for role " f"{role!r}" ) # 정본 모델이 있으면 레거시 공급자 제약을 접어 materialize 덮어쓰기를 막는다. if role in canonical_models: constraints.pop(role, None) values.pop(role, None) if role in canonical_counts: counts.pop(role, None) return CanonicalModelSelection( counts, {}, constraints, values, legacy.legacy_sources, ) def _legacy_matches_canonical( role: str, canonical_counts: Mapping[str, int], canonical_models: Mapping[str, tuple[str, ...]], legacy: CanonicalModelSelection, ) -> bool: """레거시(--workers 등)가 변환 후 정본과 같은 값으로 수렴하는지 본다.""" leg_count = legacy.role_counts.get(role) can_count = canonical_counts.get(role) if can_count is not None and leg_count is not None and can_count != leg_count: return False leg_providers = legacy.provider_constraints.get(role) can_models = canonical_models.get(role) if can_models is not None and leg_providers is not None: can_providers = tuple( _provider_from_model_ref(model_ref) for model_ref in can_models ) if can_providers != tuple( provider.lower() for provider in leg_providers ): return False # 워커 목록은 칸 전체 순서다. 칸 수가 다르면 다른 선택이다. if len(can_models) != len(leg_providers): return False leg_models = legacy.legacy_model_values.get(role, {}) for provider, model_ref in zip(leg_providers, can_models): explicit = leg_models.get(provider) if not explicit: continue model_id = ( model_ref.split("/", 1)[1] if "/" in model_ref else model_ref ) if explicit != model_id and model_ref != f"{provider}/{explicit}": return False return True if can_count is not None and leg_providers is not None: if can_count != len(leg_providers): return False if ( can_models is not None and leg_count is not None and leg_providers is None and len(can_models) > leg_count ): return False return True def _provider_from_model_ref(model_ref: str) -> str: provider, separator, _model = model_ref.partition("/") if separator: return provider.strip().lower() return model_ref.strip().lower() def _validate_selection( profile: RoleProfile, counts: Mapping[str, int], models: Mapping[str, tuple[str, ...]], explicit_counts: Mapping[str, int], ) -> None: requirements = {row.role: row for row in profile.roles if not row.dynamic} requirements["leader"] = RoleRequirement("leader", 1, 1, 1, "lead") for role in sorted(set(counts) | set(models)): requirement = requirements.get(role) if requirement is None: raise ModelSelectionInputError( f"role {role!r} is not static in the profile" ) _validate_role_count( role, counts, requirement, explicit=role in explicit_counts, ) selected_count = counts.get(role, requirement.recommended_count) model_count = len(models.get(role, ())) if model_count > selected_count: if role not in explicit_counts: raise ModelSelectionInputError( f"set --role-count {role}={model_count} before assigning " f"{model_count} models" ) raise ModelSelectionInputError( f"role {role!r} has {model_count} models for {selected_count} instances" ) def _validate_role_count( role: str, counts: Mapping[str, int], requirement: RoleRequirement, *, explicit: bool, ) -> None: if role not in counts: return count = counts[role] fixed = requirement.min_count == requirement.max_count if explicit and fixed: raise ModelSelectionInputError( f"role {role!r} has a fixed quantity and cannot receive --role-count" ) if fixed and count != requirement.min_count: raise ModelSelectionInputError( f"role {role!r} has a fixed quantity of {requirement.min_count}" ) if count < requirement.min_count or count > requirement.max_count: raise ModelSelectionInputError( f"role {role!r} count must be in " f"{requirement.min_count}..{requirement.max_count}: {count}" ) def _requirement(profile: RoleProfile, role: str) -> RoleRequirement | None: return next((row for row in profile.roles if row.role == role), None) def serialize_host_session_context(context: HostSessionContext) -> str: """Serialize host-owned session facts as stable canonical JSON.""" current = context.current_model payload = { "availableFunctions": sorted(context.available_functions), "currentModel": { "effort": current.effort, "level": current.level, "normalizedModelRef": current.normalized_model_ref, "observedModel": current.observed_model, "providerId": current.provider_id, "source": current.source, }, "entryMode": context.entry_mode, "hostId": context.host_id, "interactionSurface": context.interaction_surface, } return json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ) def deserialize_host_session_context(payload: str) -> HostSessionContext: """Deserialize only the shared context schema; never infer model identity.""" if not payload: return HostSessionContext( host_id="", entry_mode="spawn-process", available_functions=frozenset(), interaction_surface="unavailable", current_model=CurrentSessionModelAttestation.unknown(""), ) try: raw = json.loads(payload) except json.JSONDecodeError as exc: raise ModelSelectionInputError("invalid host session context JSON") from exc if not isinstance(raw, dict): raise ModelSelectionInputError("host session context must be a JSON object") current = _deserialize_attestation(raw.get("currentModel")) return HostSessionContext( host_id=_required_string(raw, "hostId"), entry_mode=_enum(raw, "entryMode", {"spawn-process", "current-session"}), available_functions=frozenset(_string_list(raw, "availableFunctions")), interaction_surface=_required_string(raw, "interactionSurface"), current_model=current, ) def _deserialize_attestation(raw: object) -> CurrentSessionModelAttestation: if raw is None: return CurrentSessionModelAttestation.unknown("") if not isinstance(raw, dict): raise ModelSelectionInputError("currentModel must be a JSON object") level = _enum(raw, "level", {"exact", "channel", "unknown"}) source = _enum( raw, "source", {"host-session-metadata", "native-api", "cli-handoff", "unavailable"}, ) observed = _nullable_string(raw, "observedModel") normalized = _nullable_string(raw, "normalizedModelRef") # 키 없으면 null. 호스트 어댑터가 아직 안 채운 경우를 허용한다. effort = _nullable_string(raw, "effort") if level == "unknown" and (observed is not None or normalized is not None): raise ModelSelectionInputError("unknown currentModel must use null model fields") if level != "unknown" and normalized is None: raise ModelSelectionInputError( "attested currentModel requires normalizedModelRef" ) return CurrentSessionModelAttestation( _required_string(raw, "providerId"), observed, normalized, level, source, effort, ) def _required_string(raw: Mapping[str, object], key: str) -> str: value = raw.get(key) if not isinstance(value, str): raise ModelSelectionInputError(f"host session context {key} must be a string") return value def _nullable_string(raw: Mapping[str, object], key: str) -> str | None: value = raw.get(key) if value is not None and not isinstance(value, str): raise ModelSelectionInputError( f"host session context {key} must be a string or null" ) return value def _enum(raw: Mapping[str, object], key: str, allowed: set[str]) -> str: value = _required_string(raw, key) if value not in allowed: raise ModelSelectionInputError( f"host session context {key} must be one of: {', '.join(sorted(allowed))}" ) return value def _string_list(raw: Mapping[str, object], key: str) -> tuple[str, ...]: value = raw.get(key) if not isinstance(value, list) or not all( isinstance(item, str) for item in value ): raise ModelSelectionInputError( f"host session context {key} must be an array of strings" ) return tuple(value)