Templates¶

def list_templates() -> tuple[TemplateInfo, ...]:
    """Return all available templates (user directory + built-in).

    User templates override built-in ones. Sorted by name.

    Returns:
        Sorted tuple of :class:`TemplateInfo` objects.
    """
    seen: dict[str, TemplateInfo] = {}

    # User templates (higher priority).
    _collect_user_templates(seen)

    # Built-in templates (lower priority).
    for name in sorted(BUILTIN_TEMPLATES):
        if name not in seen:
            try:
                loaded = _load_builtin(name)
                seen[name] = _template_info_from_loaded(
                    name,
                    loaded,
                    "builtin",
                )
            except (TemplateRenderError, TemplateValidationError, OSError) as exc:
                logger.exception(
                    TEMPLATE_BUILTIN_DEFECT,
                    template_name=name,
                    error=str(exc),
                )

    return tuple(info for _, info in sorted(seen.items()))

def list_builtin_templates() -> tuple[str, ...]:
    """Return names of all built-in templates.

    Returns:
        Sorted tuple of built-in template names.
    """
    return tuple(sorted(BUILTIN_TEMPLATES))

def load_template(name: str) -> LoadedTemplate:
    """Load a template by name: user directory first, then builtins.

    Args:
        name: Template name (e.g. ``"startup"``).

    Returns:
        :class:`LoadedTemplate` with validated data and raw YAML.

    Raises:
        TemplateNotFoundError: If no template with *name* exists.
    """
    name_clean = name.strip().lower()
    logger.debug(TEMPLATE_LOAD_START, template_name=name_clean)

    # Sanitize to prevent path traversal (OS-independent).
    if "/" in name_clean or "\\" in name_clean or ".." in name_clean:
        msg = f"Invalid template name {name!r}: must not contain path separators"
        logger.warning(TEMPLATE_LOAD_INVALID_NAME, template_name=name)
        raise TemplateNotFoundError(
            msg,
            locations=(ConfigLocation(file_path=f"<template:{name}>"),),
        )

    # Try user directory first.
    if _USER_TEMPLATES_DIR.is_dir():
        user_path = _USER_TEMPLATES_DIR / f"{name_clean}.yaml"
        if user_path.is_file():
            result = _load_from_file(user_path)
            logger.debug(
                TEMPLATE_LOAD_SUCCESS,
                template_name=name_clean,
                source="user",
            )
            return result

    # Fall back to builtins.
    if name_clean in BUILTIN_TEMPLATES:
        result = _load_builtin(name_clean)
        logger.debug(
            TEMPLATE_LOAD_SUCCESS,
            template_name=name_clean,
            source="builtin",
        )
        return result

    available = list_builtin_templates()
    logger.error(
        TEMPLATE_LOAD_ERROR,
        template_name=name,
        available=list(available),
    )
    msg = f"Unknown template {name!r}. Available: {list(available)}"
    raise TemplateNotFoundError(
        msg,
        locations=(ConfigLocation(file_path=f"<template:{name}>"),),
    )

def load_template_file(path: Path | str) -> LoadedTemplate:
    """Load a template from an explicit file path.

    Args:
        path: Path to the template YAML file.

    Returns:
        :class:`LoadedTemplate` with validated data and raw YAML.

    Raises:
        TemplateNotFoundError: If the file does not exist.
        TemplateValidationError: If validation fails.
    """
    path = Path(path)
    if not path.is_file():
        msg = f"Template file not found: {path}"
        logger.warning(TEMPLATE_LOAD_NOT_FOUND, path=str(path))
        raise TemplateNotFoundError(
            msg,
            locations=(ConfigLocation(file_path=str(path)),),
        )
    return _load_from_file(path)

def render_template(
    loaded: LoadedTemplate,
    variables: dict[str, Any] | None = None,
    *,
    locales: list[str] | None = None,
    custom_presets: Mapping[str, dict[str, Any]] | None = None,
) -> RootConfig:
    """Render a loaded template into a validated RootConfig.

    Resolves template inheritance (``extends``) before validation.

    Args:
        loaded: :class:`LoadedTemplate` from the loader.
        variables: User-supplied variable values (overrides defaults).
        locales: Faker locale codes for auto-name generation.
            Defaults to all Latin-script locales when ``None``.
        custom_presets: Optional mapping of custom preset names to
            personality config dicts for resolving user-defined presets.

    Returns:
        Validated, frozen :class:`RootConfig`.

    Raises:
        TemplateRenderError: If rendering fails.
        TemplateValidationError: If validation fails.
        TemplateInheritanceError: If inheritance resolution fails.
    """
    logger.info(
        TEMPLATE_RENDER_START,
        source_name=loaded.source_name,
    )
    config_dict = _render_to_dict(
        loaded,
        variables,
        locales=locales,
        custom_presets=custom_presets,
    )

    # Merge with defaults and validate.
    merged = deep_merge(default_config_dict(), config_dict)
    result = validate_as_root_config(merged, loaded.source_name)
    logger.info(
        TEMPLATE_RENDER_SUCCESS,
        source_name=loaded.source_name,
    )
    return result

def merge_template_configs(
    parent: dict[str, Any],
    child: dict[str, Any],
) -> dict[str, Any]:
    """Merge a parent config dict with a child config dict.

    Merge strategies by field:

    - ``company_name``, ``company_type``: child wins if present.
    - ``config`` (dict): deep-merged; child keys override parent.
    - ``agents`` (list): merged by ``(role, department, merge_id)`` key.
    - ``departments`` (list): merged by ``name`` (case-insensitive).
    - ``workflow``, ``workflow_handoffs``, ``escalation_paths``: child
      replaces entirely if present; otherwise inherited from parent.

    Args:
        parent: Rendered parent config dict (post-Jinja2, pre-defaults).
        child: Rendered child config dict (post-Jinja2, pre-defaults).

    Returns:
        New merged config dict.
    """
    logger.debug(TEMPLATE_INHERIT_MERGE, action="start")

    result: dict[str, Any] = {}

    # Scalars: child wins if present.
    for key in ("company_name", "company_type"):
        if key in child and child[key] is not None:
            result[key] = child[key]
        elif key in parent:
            result[key] = parent[key]

    # Config dict: deep merge.
    parent_config = parent.get("config", {})
    child_config = child.get("config", {})
    if parent_config or child_config:
        result["config"] = deep_merge(
            parent_config if isinstance(parent_config, dict) else {},
            child_config if isinstance(child_config, dict) else {},
        )

    # Agents: merge by (role, department, merge_id) key.
    parent_agents = parent.get("agents", [])
    child_agents = child.get("agents", [])
    if parent_agents or child_agents:
        result["agents"] = _merge_agents(parent_agents, child_agents)

    # Departments: merge by name.
    parent_depts = parent.get("departments", [])
    child_depts = child.get("departments", [])
    if parent_depts or child_depts:
        result["departments"] = _merge_departments(parent_depts, child_depts)

    # Replace-if-present fields (deep-copied to prevent reference sharing).
    for key in ("workflow", "workflow_handoffs", "escalation_paths"):
        if key in child and child[key] is not None:
            result[key] = copy.deepcopy(child[key])
        elif key in parent:
            result[key] = copy.deepcopy(parent[key])

    logger.debug(TEMPLATE_INHERIT_MERGE, action="done")
    return result

def get_personality_preset(
    name: str,
    custom_presets: Mapping[str, dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Look up a personality preset by name.

    Custom presets are checked first (higher precedence), then builtins.

    Args:
        name: Preset name (case-insensitive, whitespace-stripped).
        custom_presets: Optional mapping of custom preset names to
            personality config dicts.  Keys must be lowercased.

    Returns:
        A *copy* of the personality configuration dict.

    Raises:
        KeyError: If the preset name is not found in either source.
    """
    key = name.strip().lower()
    if custom_presets is not None and key in custom_presets:
        return copy.deepcopy(custom_presets[key])
    if key in PERSONALITY_PRESETS:
        return dict(PERSONALITY_PRESETS[key])
    available = sorted(PERSONALITY_PRESETS)
    if custom_presets:
        available = sorted({*available, *custom_presets})
    msg = f"Unknown personality preset {name!r}. Available: {available}"
    logger.warning(
        TEMPLATE_PERSONALITY_PRESET_UNKNOWN,
        preset_name=name,
        available=available,
    )
    raise KeyError(msg)

def validate_preset_references(
    template: CompanyTemplate,
    custom_presets: Mapping[str, dict[str, Any]] | None = None,
) -> tuple[str, ...]:
    """Check all agent personality_preset references against known presets.

    Returns a tuple of warning messages for unknown presets.  Does not
    raise -- purely advisory for pre-flight validation and template
    import/export scenarios.

    Args:
        template: Parsed template to validate.
        custom_presets: Optional custom preset mapping.  Keys must
            be lowercased.

    Returns:
        Tuple of warning strings (empty when all presets are known).
    """
    issues: list[str] = []
    for agent_cfg in template.agents:
        preset = agent_cfg.personality_preset
        if preset is None:
            continue
        key = preset.strip().lower()
        if custom_presets is not None and key in custom_presets:
            continue
        if key in PERSONALITY_PRESETS:
            continue
        issues.append(
            f"Agent {agent_cfg.role!r} references unknown personality preset {preset!r}"
        )
    return tuple(issues)

def generate_auto_name(
    role: str,  # noqa: ARG001
    *,
    seed: int | None = None,
    locales: list[str] | None = None,
) -> str:
    """Generate an internationally diverse agent name using Faker.

    When *seed* is provided, a local ``random.Random`` deterministically
    selects a locale, then a **fresh** single-locale Faker instance
    generates the name -- the cached instance is never mutated.

    The *role* parameter is accepted because callers
    (``setup_agents.py``, ``renderer.py``) pass it positionally;
    it does not influence name generation.

    Args:
        role: The agent's role name.  Unused since the switch from
            role-based name pools to Faker.
        seed: Optional random seed for deterministic naming.
        locales: Faker locale codes to draw from.  Defaults to all
            Latin-script locales when ``None`` or empty.

    Returns:
        A generated full name string.
    """
    import random  # noqa: PLC0415

    from faker import Faker  # noqa: PLC0415

    from synthorg.templates.locales import ALL_LATIN_LOCALES  # noqa: PLC0415

    locale_list = locales or list(ALL_LATIN_LOCALES)
    try:
        if seed is not None:
            rng = random.Random(seed)  # noqa: S311
            chosen_locale = rng.choice(locale_list)
            # Fresh instance -- never mutate the shared cached one.
            fake = Faker([chosen_locale])
            fake.seed_instance(seed)
        else:
            fake = _get_faker(tuple(locale_list))
        return str(fake.name())
    except MemoryError, RecursionError:
        raise
    except Exception:
        from synthorg.observability.events.template import (  # noqa: PLC0415
            TEMPLATE_NAME_GEN_FAKER_ERROR,
        )

        logger.warning(
            TEMPLATE_NAME_GEN_FAKER_ERROR,
            locales=locale_list[:5],
            seed=seed,
            exc_info=True,
        )
        # Fall back to a known-safe locale.
        fallback = Faker(["en_US"])
        if seed is not None:
            fallback.seed_instance(seed)
        return str(fallback.name())

def parse_model_requirement(raw: str | dict[str, Any]) -> ModelRequirement:
    """Parse a model requirement from a string tier or dict.

    Backward-compatible: accepts the legacy ``"medium"`` string format
    used by existing template YAML files as well as the new dict format.

    Args:
        raw: Either a tier string (``"large"``, ``"medium"``, ``"small"``)
            or a dict with ``ModelRequirement`` fields.

    Returns:
        Parsed ``ModelRequirement``.

    Raises:
        ValueError: If *raw* is a string not in the valid tier set.
        ValidationError: If *raw* is a dict with invalid fields.
    """
    if isinstance(raw, str):
        key = raw.strip().lower()
        if key not in _VALID_TIERS:
            msg = f"Invalid model tier {raw!r}. Valid tiers: {sorted(_VALID_TIERS)}"
            logger.warning(
                TEMPLATE_MODEL_REQUIREMENT_INVALID,
                raw_tier=raw,
                valid_tiers=sorted(_VALID_TIERS),
            )
            raise ValueError(msg)
        result = ModelRequirement(tier=key)  # type: ignore[arg-type]
    else:
        try:
            result = ModelRequirement(**raw)
        except ValidationError:
            logger.warning(
                TEMPLATE_MODEL_REQUIREMENT_INVALID,
                raw_requirement=raw,
                reason="dict_validation_failed",
                exc_info=True,
            )
            raise

    logger.debug(
        TEMPLATE_MODEL_REQUIREMENT_PARSED,
        tier=result.tier,
        priority=result.priority,
    )
    return result

def resolve_model_requirement(
    tier_str: str,
    preset_name: str | None = None,
) -> ModelRequirement:
    """Merge a template tier alias with personality-preset affinity.

    The template's tier always wins.  Affinity provides ``priority``
    and ``min_context`` defaults based on the personality preset.

    Args:
        tier_str: Tier alias from the template agent config.
        preset_name: Optional personality preset name for affinity lookup.

    Returns:
        Resolved ``ModelRequirement``.
    """
    affinity: dict[str, Any] = dict(
        MODEL_AFFINITY.get((preset_name or "").strip().lower(), {}),
    )

    merged: dict[str, Any] = {"tier": tier_str.strip().lower()}
    # Affinity values fill in priority and min_context when available.
    if "priority" in affinity:
        merged["priority"] = affinity["priority"]
    if "min_context" in affinity:
        merged["min_context"] = affinity["min_context"]

    result = parse_model_requirement(merged)
    logger.debug(
        TEMPLATE_MODEL_REQUIREMENT_RESOLVED,
        tier=result.tier,
        priority=result.priority,
        min_context=result.min_context,
        preset=preset_name,
    )
    return result

def match_model(
    requirement: ModelRequirement,
    available: tuple[ProviderModelConfig, ...],
) -> tuple[ProviderModelConfig | None, float]:
    """Select the best model for a requirement from available models.

    Models are classified into cost-based tiers (thirds by input cost),
    then ranked within the matching tier according to the requirement's
    priority axis.

    Args:
        requirement: Structured model requirement.
        available: Tuple of available models from a single provider.

    Returns:
        Tuple of (best matching model or None, score 0-1).
    """
    if not available:
        return None, 0.0

    # Filter by minimum context window.
    candidates = [m for m in available if m.max_context >= requirement.min_context]
    if not candidates:
        return None, 0.0

    # Classify into tiers by cost.
    tier_models = _classify_tiers(candidates)
    tier_candidates = tier_models.get(requirement.tier, [])

    # Fall back to next-best tier if exact tier is empty.
    if not tier_candidates:
        for fallback in _TIER_FALLBACK[requirement.tier]:
            tier_candidates = tier_models.get(fallback, [])
            if tier_candidates:
                break

    if not tier_candidates:
        return None, 0.0

    # Rank within tier by priority axis.
    best = _rank_by_priority(tier_candidates, requirement.priority)
    score = _compute_score(best, requirement, tier_candidates)
    return best, score

def match_all_agents(
    agents: list[dict[str, Any]],
    providers: dict[str, Any],
) -> list[ModelMatch]:
    """Batch-match template agents to provider models.

    For each agent, resolves its model requirement and finds the best
    model across all configured providers.

    Note:
        The *agents* list is shallow-copied from the caller. Each dict
        is shared, so nested mutable values (e.g. ``personality``) are
        **not** copied. This function only reads agent dicts.

    Args:
        agents: List of expanded agent config dicts.  Model requirement
            resolution uses three paths (checked in order):

            - ``model_requirement`` (``ModelRequirement``): used directly.
            - ``model_requirement`` (dict): deserialized to
              ``ModelRequirement`` via ``parse_model_requirement``.
            - ``tier`` (str) + optional ``personality_preset`` (str):
              resolved via ``resolve_model_requirement`` with
              personality-based affinity defaults.
        providers: Provider name -> provider config mapping.  Each
            provider config must have a ``models`` attribute returning
            a tuple of ``ProviderModelConfig``.

    Returns:
        List of ``ModelMatch`` results.  Agents may be omitted from
        the result when no models exist across any provider or when
        requirement resolution fails.  Agents with a viable provider
        but no tier match get a ``ModelMatch`` with score 0 and the
        first available provider/model as a fallback.
    """
    from synthorg.templates.model_requirements import (  # noqa: PLC0415
        ModelRequirement,
        parse_model_requirement,
        resolve_model_requirement,
    )

    results: list[ModelMatch] = []

    # Flatten all models across providers for fallback.
    all_models: list[tuple[str, ProviderModelConfig]] = [
        (pname, m) for pname, pcfg in providers.items() for m in pcfg.models
    ]

    for idx, agent in enumerate(agents):
        resolved = _resolve_agent_requirement(
            agent,
            idx,
            ModelRequirement,
            parse_model_requirement,
            resolve_model_requirement,
        )
        if resolved is None:
            continue
        req, tier = resolved

        best_provider: str | None = None
        best_model: ProviderModelConfig | None = None
        best_score = 0.0

        # Try each provider.
        for pname, pcfg in providers.items():
            model, score = match_model(req, pcfg.models)
            if model is not None and score > best_score:
                best_provider = pname
                best_model = model
                best_score = score

        if best_provider is not None and best_model is not None:
            logger.debug(
                TEMPLATE_MODEL_MATCH_SUCCESS,
                agent_index=idx,
                provider=best_provider,
                model=best_model.id,
                score=best_score,
            )
            results.append(
                ModelMatch(
                    agent_index=idx,
                    provider_name=best_provider,
                    model_id=best_model.id,
                    tier=tier,
                    score=best_score,
                ),
            )
        elif all_models:
            # Fallback: assign first available model with score 0.
            fb_provider, fb_model = all_models[0]
            logger.warning(
                TEMPLATE_MODEL_MATCH_FAILED,
                agent_index=idx,
                tier=tier,
                fallback_provider=fb_provider,
                fallback_model=fb_model.id,
            )
            results.append(
                ModelMatch(
                    agent_index=idx,
                    provider_name=fb_provider,
                    model_id=fb_model.id,
                    tier=tier,
                    score=0.0,
                ),
            )
        else:
            logger.warning(
                TEMPLATE_MODEL_MATCH_FAILED,
                agent_index=idx,
                tier=tier,
                reason="no_models_available",
            )

    return results