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.warning(
                    TEMPLATE_BUILTIN_DEFECT,
                    template_name=name,
                    error_type=type(exc).__name__,
                    error=safe_error_description(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 = normalize_ascii_lowercase(name)
    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, object] | None = None,
    *,
    locales: list[str] | None = None,
    custom_presets: Mapping[str, dict[str, JsonValue]] | 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,
    )

    _apply_effective_posture(loaded.template, config_dict)

    # 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, object],
    child: dict[str, object],
) -> dict[str, object]:
    """Merge a parent config dict with a child config dict.

    Merge strategies by field:

    - ``company_name``, ``company_type``: child wins if present.
    - ``config``, ``security``, ``budget`` (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``, ``posture``:
      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, object] = {}

    # 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]

    # Deep-merged dict sections: child keys override parent.
    for section in ("config", "security", "budget"):
        parent_section = parent.get(section, {})
        child_section = child.get(section, {})
        if parent_section or child_section:
            result[section] = deep_merge(
                parent_section if isinstance(parent_section, Mapping) else {},
                child_section if isinstance(child_section, Mapping) 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(
            _as_dict_list(parent_agents, "agents"),
            _as_dict_list(child_agents, "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(
            _as_dict_list(parent_depts, "departments"),
            _as_dict_list(child_depts, "departments"),
        )

    # Replace-if-present fields (deep-copied to prevent reference sharing).
    # ``posture`` is child-wins: a child template's declared posture replaces
    # the parent's entirely rather than blending two operating modes.
    for key in ("workflow", "workflow_handoffs", "escalation_paths", "posture"):
        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, JsonValue]] | None = None,
) -> dict[str, JsonValue]:
    """Look up a personality preset by name.

    Custom presets take precedence over builtins; the result is a fresh
    one-level copy with any tuple-valued field normalised to a JSON list.

    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 fresh JSON-shaped copy of the personality configuration dict.

    Raises:
        KeyError: If the preset name is not found in either source.
    """
    key = normalize_ascii_lowercase(name)
    source: Mapping[str, PresetValue | JsonValue] | None = None
    if custom_presets is not None and key in custom_presets:
        source = custom_presets[key]
    elif key in PERSONALITY_PRESETS:
        source = PERSONALITY_PRESETS[key]
    if source is not None:
        return {
            field: list(value) if isinstance(value, (list, tuple)) else value
            for field, value in source.items()
        }
    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 get_strategic_output_default(
    level: SeniorityLevel,
) -> StrategicOutputMode | None:
    """Return the default strategic output mode for a seniority level.

    Args:
        level: Agent seniority level.

    Returns:
        Default strategic output mode, or ``None`` if the level has
        no strategic default (i.e. strategic output is not applicable).
    """
    return STRATEGIC_OUTPUT_DEFAULTS.get(level)

def validate_preset_references(
    template: CompanyTemplate,
    custom_presets: Mapping[str, dict[str, JsonValue]] | 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 = normalize_ascii_lowercase(preset)
        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.

    With *seed*, a fresh single-locale Faker instance is used so the
    shared cached instance is never mutated.  *role* is accepted for
    positional-caller compatibility but does not influence the name;
    *locales* 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 _two_part_name(fake.first_name, fake.last_name)
    except Exception as exc:  # noqa: BLE001 -- criticals re-raised
        reraise_critical(exc)
        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,
            error_type=type(exc).__name__,
            error=safe_error_description(exc),
        )
        # Fall back to a known-safe locale.
        fallback = Faker(["en_US"])
        if seed is not None:
            fallback.seed_instance(seed)
        return _two_part_name(fallback.first_name, fallback.last_name)

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

    A bare string is an explicit example-id pin (``model_id``); a dict maps
    directly onto the :class:`ModelRequirement` capability/family fields.

    Args:
        raw: Either a non-blank model id/alias string, or a dict with
            ``ModelRequirement`` fields.

    Returns:
        Parsed ``ModelRequirement``.

    Raises:
        ValueError: If *raw* is a blank string.
        ValidationError: If *raw* is a dict with invalid fields.
    """
    if isinstance(raw, str):
        pinned = raw.strip()
        if not pinned:
            msg = "Model id reference must be a non-blank string"
            logger.warning(
                TEMPLATE_MODEL_REQUIREMENT_INVALID,
                raw_reference=raw,
                reason="blank_model_id",
            )
            raise ValueError(msg)
        result = ModelRequirement.model_validate({"model_id": pinned})
    else:
        try:
            result = ModelRequirement.model_validate(raw)
        except ValidationError:
            logger.warning(
                TEMPLATE_MODEL_REQUIREMENT_INVALID,
                raw_requirement=raw,
                reason="dict_validation_failed",
            )
            raise

    logger.debug(
        TEMPLATE_MODEL_REQUIREMENT_PARSED,
        model_id=result.model_id,
        priority=result.priority,
        family=result.family,
    )
    return result

def resolve_model_requirement(
    preset_name: str | None = None,
    overrides: dict[str, JsonValue] | None = None,
) -> ModelRequirement:
    """Merge a personality-preset affinity profile with explicit overrides.

    The affinity profile supplies capability defaults (priority, context
    floor, hard requirement flags, optional family hint); any field the
    template states explicitly via *overrides* always wins.

    Args:
        preset_name: Optional personality preset name for affinity lookup.
        overrides: Explicit ``ModelRequirement`` fields from the template
            agent that take precedence over the affinity defaults.

    Returns:
        Resolved ``ModelRequirement``.
    """
    if overrides:
        pin = overrides.get("model_id")
        if isinstance(pin, str) and pin.strip():
            # A pin is selected verbatim and bypasses capability scoring, so
            # affinity flags would be inert against it and are NOT merged in.
            # The full overrides dict is parsed (not just the pin) so any
            # explicit field the template set alongside model_id is preserved,
            # and a contradictory model_id + family/model_pattern pairing is
            # rejected by the ModelRequirement validator.
            pinned = parse_model_requirement(
                {key: value for key, value in overrides.items() if value is not None}
            )
            logger.debug(
                TEMPLATE_MODEL_REQUIREMENT_RESOLVED,
                model_id=pinned.model_id,
                priority=pinned.priority,
                min_context=pinned.min_context,
                family=pinned.family,
                preset=preset_name,
            )
            return pinned

    affinity = MODEL_AFFINITY.get(
        normalize_ascii_lowercase_or_default(preset_name),
        MappingProxyType({}),
    )
    merged: dict[str, JsonValue] = {
        key: affinity[key] for key in _AFFINITY_DEFAULT_KEYS if key in affinity
    }
    if overrides:
        merged.update({k: v for k, v in overrides.items() if v is not None})

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