Function bodies 27 total

    def __init__(self, schema_name: str) -> None:
        super().__init__(
            f"Schema '{schema_name}' not found. "
            f"Available schemas: {sorted(_SUPPORTED_SCHEMAS)}"
        )
        self.schema_name = schema_name

    def __init__(
        self,
        message: str,
        validation_error: jsonschema.exceptions.ValidationError,
    ) -> None:
        super().__init__(message)
        self.validation_error = validation_error

    def __init__(self, path: Path, cause: Exception) -> None:
        super().__init__(
            f"Failed to parse schema file '{path}': {cause}"
        )
        self.path = path
        self.cause = cause

def _resolve_schema_path(schema_name: str) -> Path:
    """Return the filesystem path for a named schema.

    Searches for ``<schema_name>.json`` first, then ``<schema_name>.yaml``.

    Args:
        schema_name: Basename without extension (e.g. ``"tool_canonical_ir_v1"``).

    Returns:
        :class:`pathlib.Path` pointing to the schema file.

    Raises:
        :class:`SchemaNotFoundError`: If neither JSON nor YAML file exists.
    """
    for suffix in (_JSON_SUFFIX, _YAML_SUFFIX):
        candidate = SCHEMAS_DIR / f"{schema_name}{suffix}"
        if candidate.is_file():
            return candidate
    raise SchemaNotFoundError(schema_name)

def load_schema(schema_name: str) -> SchemaDict:
    """Load and parse a canonical AumAI schema by name.

    Results are cached in-process so subsequent calls for the same schema
    name are O(1) dictionary lookups. The cache is keyed by schema name,
    so different names always return different objects.

    Args:
        schema_name: Basename without extension.
            One of: ``tool_canonical_ir_v1``, ``agent_error_taxonomy_v1``,
            ``capsule_format_v1``, ``agent_capability_v1``.

    Returns:
        Parsed schema as a nested dictionary.

    Raises:
        :class:`SchemaNotFoundError`: If the schema file does not exist.
        :class:`SchemaParseError`: If the file cannot be parsed.

    Example::

        >>> schema = load_schema("tool_canonical_ir_v1")
        >>> schema["title"]
        'AumAI Tool Canonical IR'
    """
    path = _resolve_schema_path(schema_name)
    try:
        raw_text = path.read_text(encoding="utf-8")
    except OSError as exc:
        r

def validate(data: dict[str, Any], schema_name: str) -> bool:
    """Validate *data* against the named canonical schema.

    Uses ``jsonschema`` with the JSON Schema Draft 2020-12 validator.
    Both JSON and YAML schemas are supported (YAML is loaded into a dict
    before validation).

    Args:
        data: The dictionary to validate.
        schema_name: Name of the schema to validate against.

    Returns:
        ``True`` if validation passes.

    Raises:
        :class:`SchemaNotFoundError`: If the schema cannot be found.
        :class:`SchemaValidationError`: If *data* fails validation.

    Example::

        >>> validate({"name": "my_tool", ...}, "tool_canonical_ir_v1")
        True
    """
    schema = load_schema(schema_name)
    validator_cls = jsonschema.validators.validator_for(schema)
    validator_cls.check_schema(schema)
    validator = validator_cls(schema)

    try:
        validator.validate(data)
    except jsonschema.exceptions.ValidationError as exc:
       

def validate_quietly(data: dict[str, Any], schema_name: str) -> tuple[bool, str]:
    """Validate *data* against a schema without raising exceptions.

    A convenience wrapper around :func:`validate` that returns a ``(bool,
    message)`` tuple instead of raising. Useful for bulk validation where
    you want to collect all failures rather than short-circuit on the first.

    Args:
        data: The dictionary to validate.
        schema_name: Name of the schema to validate against.

    Returns:
        A ``(is_valid, message)`` tuple. ``message`` is an empty string on
        success, or the first validation error message on failure.
    """
    try:
        validate(data, schema_name)
        return True, ""
    except (SchemaValidationError, SchemaNotFoundError, SchemaParseError) as exc:
        return False, str(exc)

def list_schemas() -> list[str]:
    """Return the names of all bundled canonical schemas.

    Returns:
        Sorted list of schema base names (without file extension).

    Example::

        >>> list_schemas()
        ['agent_capability_v1', 'agent_error_taxonomy_v1', 'capsule_format_v1', 'tool_canonical_ir_v1']
    """
    names: list[str] = []
    for path in SCHEMAS_DIR.iterdir():
        if path.suffix in (_JSON_SUFFIX, _YAML_SUFFIX) and path.stem != "__init__":
            names.append(path.stem)
    return sorted(names)

def get_schema_path(schema_name: str) -> Path:
    """Return the absolute filesystem path to a schema file.

    Useful for tooling that needs to reference the raw file (e.g. editors,
    linters, or documentation generators).

    Args:
        schema_name: Schema base name without extension.

    Returns:
        :class:`pathlib.Path` to the schema file.

    Raises:
        :class:`SchemaNotFoundError`: If the schema file does not exist.
    """
    return _resolve_schema_path(schema_name)

def _validate_sha256(value: str) -> str:
    if not _SHA256_PATTERN.match(value):
        raise ValueError(
            f"'{value}' is not a valid SHA-256 hash (expected 'sha256:<64 hex chars>')"
        )
    return value

    def validate_name_pattern(cls, value: str) -> str:
        if not _TOOL_NAME_PATTERN.match(value):
            raise ValueError(
                f"Tool name '{value}' must start with a letter and contain only "
                "alphanumerics, underscores, hyphens, and dots."
            )
        return value

    def validate_deprecation_coherence(self) -> "ToolCanonicalIR":
        if self.metadata.deprecation_message and not self.metadata.deprecated:
            raise ValueError(
                "deprecation_message is set but deprecated is False. "
                "Set deprecated=True if providing a deprecation_message."
            )
        return self

    def validate_code_category_alignment(cls, value: int) -> int:
        # Codes must be in [100, 699] — already enforced by ge/le.
        # Additionally validate the hundreds digit is in [1, 6].
        prefix = value // 100
        if prefix < 1 or prefix > 6:
            raise ValueError(
                f"Error code {value} has hundreds prefix {prefix}. "
                "Valid range is 1xx-6xx."
            )
        return value

    def validate_category_matches_code_prefix(self) -> "AgentError":
        """Ensure the category enum aligns with the code's hundreds digit."""
        prefix_to_category: dict[int, ErrorCategory] = {
            1: ErrorCategory.PLANNING,
            2: ErrorCategory.TOOL_EXECUTION,
            3: ErrorCategory.CONTEXT,
            4: ErrorCategory.SECURITY,
            5: ErrorCategory.RESOURCE,
            6: ErrorCategory.ORCHESTRATION,
        }
        expected = prefix_to_category[self.code // 100]
        if self.category != expected:
            raise ValueError(
                f"Code {self.code} belongs to category '{expected.value}' "
                f"but category '{self.category.value}' was provided."
            )
        return self

    def validate_screaming_snake(cls, value: str) -> str:
        if not _SCREAMING_SNAKE_PATTERN.match(value):
            raise ValueError(
                f"Error name '{value}' must be SCREAMING_SNAKE_CASE "
                "(uppercase letters, digits, underscores only)."
            )
        return value

    def validate_error_codes_match_prefix(self) -> "TaxonomyCategory":
        for entry in self.errors:
            if entry.code // 100 != self.code_prefix:
                raise ValueError(
                    f"Error code {entry.code} does not match category prefix "
                    f"{self.code_prefix}xx."
                )
        return self

    def validate_no_duplicate_codes(self) -> "AgentErrorTaxonomy":
        seen_codes: set[int] = set()
        for category in self.categories:
            for entry in category.errors:
                if entry.code in seen_codes:
                    raise ValueError(
                        f"Duplicate error code {entry.code} found in taxonomy."
                    )
                seen_codes.add(entry.code)
        return self

    def validate_total_coherence(self) -> "TokenUsage":
        """If total_tokens is non-zero, verify it is consistent with parts."""
        computed = self.input_tokens + self.output_tokens
        if self.total_tokens != 0 and self.total_tokens < computed:
            raise ValueError(
                f"total_tokens ({self.total_tokens}) is less than "
                f"input_tokens + output_tokens ({computed})."
            )
        return self

    def validate_timing_order(self) -> "CapsuleStep":
        if self.ended_at < self.started_at:
            raise ValueError(
                f"ended_at ({self.ended_at}) must not be before "
                f"started_at ({self.started_at}) for step {self.step_index}."
            )
        return self

    def validate_tool_fields_coherence(self) -> "CapsuleStep":
        """tool_name should only be present for tool_call/tool_result steps."""
        tool_step_types = {StepType.TOOL_CALL, StepType.TOOL_RESULT}
        if self.tool_name and self.step_type not in tool_step_types:
            raise ValueError(
                f"tool_name is only valid for {tool_step_types} steps, "
                f"not '{self.step_type}'."
            )
        return self

    def validate_capsule_id_format(cls, value: str) -> str:
        if not _CAPSULE_ID_PATTERN.match(value):
            raise ValueError(
                f"capsule_id '{value}' must match pattern 'capsule_<26 alphanumeric chars>'."
            )
        return value

    def validate_schema_version(cls, value: str) -> str:
        if value != "1.0.0":
            raise ValueError(
                f"schema_version must be '1.0.0' for v1 capsules, got '{value}'."
            )
        return value

    def validate_trace_indices_contiguous(self) -> "RunCapsule":
        """Execution trace step indices must be 0-based and contiguous."""
        for expected_idx, step in enumerate(self.execution_trace):
            if step.step_index != expected_idx:
                raise ValueError(
                    f"Execution trace step at position {expected_idx} has "
                    f"step_index={step.step_index}. Indices must be contiguous "
                    f"and zero-based."
                )
        return self

    def validate_image_only_for_docker(self) -> "SandboxConfig":
        if self.image and self.tier != SandboxTier.DOCKER_LOCAL:
            raise ValueError(
                "image is only applicable when tier is 'docker_local'."
            )
        return self

    def validate_allowlist_has_rules(self) -> "NetworkConfig":
        if self.egress_mode == EgressMode.ALLOWLIST_ONLY and not self.egress_rules:
            raise ValueError(
                "egress_rules must be non-empty when egress_mode is 'allowlist_only'."
            )
        return self

    def validate_production_sandbox_requirements(self) -> "AgentCapability":
        """Production agents must not use sandbox tier 'none'."""
        if (
            self.metadata.environment == Environment.PRODUCTION
            and self.sandbox.tier == SandboxTier.NONE
        ):
            raise ValueError(
                "Production agents must not use sandbox tier 'none'. "
                "Use e2b_micro or higher."
            )
        return self

    def validate_pii_requires_confidential_classification(
        self,
    ) -> "AgentCapability":
        """Agents handling PII must declare at minimum 'confidential' classification."""
        if self.permissions.pii_access:
            elevated = {DataClassification.CONFIDENTIAL, DataClassification.RESTRICTED}
            if not any(dc in elevated for dc in self.permissions.data_classifications):
                raise ValueError(
                    "Agents with pii_access=True must declare at least "
                    "'confidential' in data_classifications."
                )
        return self