This is an automated email from the ASF dual-hosted git repository. FreeOnePlus pushed a commit to branch feat/v1-01-domain-child-availability-models in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
commit dd1e9387cc0b242f3839383329d1df7634534565 Author: FreeOnePlus <[email protected]> AuthorDate: Fri Jul 31 01:28:52 2026 +0800 feat: add hierarchical domain contract models --- doris_mcp_server/tools/domain_models.py | 630 ++++++++++++++++++++++++++++++++ test/tools/test_domain_models.py | 545 +++++++++++++++++++++++++++ 2 files changed, 1175 insertions(+) diff --git a/doris_mcp_server/tools/domain_models.py b/doris_mcp_server/tools/domain_models.py new file mode 100644 index 0000000..715b92b --- /dev/null +++ b/doris_mcp_server/tools/domain_models.py @@ -0,0 +1,630 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Immutable wire contracts for hierarchical Doris MCP domain tools.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from datetime import datetime +from enum import StrEnum +from types import MappingProxyType +from typing import Annotated, Any, Literal, Self, cast + +from jsonschema import Draft202012Validator +from jsonschema.exceptions import SchemaError +from pydantic import ( + BaseModel, + ConfigDict, + Field, + JsonValue, + StringConstraints, + field_serializer, + field_validator, + model_validator, +) + +MAX_DOMAIN_MANIFEST_BYTES = 16 * 1024 + +Identifier = Annotated[ + str, + StringConstraints( + min_length=1, + max_length=128, + pattern=r"^[a-z][a-z0-9_]*$", + strip_whitespace=True, + ), +] +BindingName = Annotated[ + str, + StringConstraints( + min_length=1, + max_length=192, + pattern=r"^[A-Za-z_][A-Za-z0-9_.:-]*$", + strip_whitespace=True, + ), +] +NonEmptyText = Annotated[ + str, + StringConstraints(min_length=1, max_length=4096, strip_whitespace=True), +] +RuleIdentifier = Annotated[ + str, + StringConstraints( + min_length=1, + max_length=192, + pattern=r"^[a-z][a-z0-9_.-]*$", + strip_whitespace=True, + ), +] +ReasonCode = Annotated[ + str, + StringConstraints( + min_length=1, + max_length=128, + pattern=r"^[A-Z][A-Z0-9_]*$", + strip_whitespace=True, + ), +] +JsonObject = dict[str, JsonValue] + + +class AvailabilityStatus(StrEnum): + """Authoritative availability states exposed in domain manifests.""" + + AVAILABLE = "available" + DEGRADED = "degraded" + UNAVAILABLE = "unavailable" + MISCONFIGURED = "misconfigured" + UNKNOWN = "unknown" + + +class UnavailableBehavior(StrEnum): + """Manifest behavior when no capability variant can execute.""" + + SHOW_DISABLED = "show_disabled" + HIDE = "hide" + + +class DomainErrorCode(StrEnum): + """Stable errors returned by hierarchical domain tools.""" + + DOMAIN_DISCOVERY_DENIED = "DOMAIN_DISCOVERY_DENIED" + CHILD_TOOL_NOT_FOUND = "CHILD_TOOL_NOT_FOUND" + CHILD_MANIFEST_STALE = "CHILD_MANIFEST_STALE" + CHILD_ARGUMENTS_INVALID = "CHILD_ARGUMENTS_INVALID" + CHILD_RESULT_INVALID = "CHILD_RESULT_INVALID" + CHILD_CAPABILITY_UNAVAILABLE = "CHILD_CAPABILITY_UNAVAILABLE" + CHILD_CONFIRMATION_REQUIRED = "CHILD_CONFIRMATION_REQUIRED" + CHILD_EXECUTION_TIMEOUT = "CHILD_EXECUTION_TIMEOUT" + CHILD_EXECUTION_FAILED = "CHILD_EXECUTION_FAILED" + + +class ContractModel(BaseModel): + """Base class with strict shape and canonical wire serialization.""" + + model_config = ConfigDict( + extra="forbid", + frozen=True, + str_strip_whitespace=True, + validate_default=True, + ) + + def to_wire(self) -> JsonObject: + """Return a JSON-compatible representation with protocol aliases.""" + return cast( + JsonObject, + self.model_dump( + by_alias=True, + mode="json", + exclude_none=True, + ), + ) + + def to_canonical_json(self) -> str: + """Serialize deterministically for manifests, tests, and hashing.""" + return json.dumps( + self.to_wire(), + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + @classmethod + def wire_schema(cls) -> dict[str, Any]: + """Return the deterministic serialization schema for this contract.""" + return cls.model_json_schema(by_alias=True, mode="serialization") + + +class ToolContractAnnotations(ContractModel): + """Risk and side-effect metadata shared by domains and children.""" + + read_only: bool + idempotent: bool + destructive: bool + open_world: bool + requires_confirmation: bool + + @model_validator(mode="after") + def _validate_risk_contract(self) -> Self: + if self.read_only and self.destructive: + raise ValueError("a tool cannot be both read-only and destructive") + if self.destructive and not self.requires_confirmation: + raise ValueError("destructive tools must require confirmation") + return self + + +class CapabilityVariant(ContractModel): + """One deterministic implementation variant for a child capability.""" + + name: Identifier + supported_ranges: Annotated[tuple[NonEmptyText, ...], Field(min_length=1)] + excluded_ranges: tuple[NonEmptyText, ...] = () + deployment_modes: tuple[Identifier, ...] = () + required_features: tuple[Identifier, ...] = () + required_system_objects: tuple[NonEmptyText, ...] = () + required_endpoints: tuple[Identifier, ...] = () + required_providers: tuple[Identifier, ...] = () + required_probes: tuple[Identifier, ...] = () + evidence_quality: Identifier = "native" + callable_when_degraded: bool = False + source_references: tuple[NonEmptyText, ...] = () + + @model_validator(mode="after") + def _validate_variant(self) -> Self: + _require_unique(self.supported_ranges, "supported_ranges") + _require_unique(self.excluded_ranges, "excluded_ranges") + _require_unique(self.deployment_modes, "deployment_modes") + _require_unique(self.required_features, "required_features") + _require_unique(self.required_system_objects, "required_system_objects") + _require_unique(self.required_endpoints, "required_endpoints") + _require_unique(self.required_providers, "required_providers") + _require_unique(self.required_probes, "required_probes") + _require_unique(self.source_references, "source_references") + if not self.source_references and not self.required_probes: + raise ValueError( + "a capability variant requires a source reference or runtime probe" + ) + return self + + +class ChildSupportContract(ContractModel): + """Version and runtime support contract for one child tool.""" + + rule_id: RuleIdentifier + variants: Annotated[tuple[CapabilityVariant, ...], Field(min_length=1)] + unavailable_behavior: UnavailableBehavior + unknown_is_callable: bool = False + tested_versions: tuple[NonEmptyText, ...] = () + + @model_validator(mode="after") + def _validate_contract(self) -> Self: + _require_unique( + tuple(variant.name for variant in self.variants), + "capability variant names", + ) + _require_unique(self.tested_versions, "tested_versions") + return self + + +class CompositeStep(ContractModel): + """One private deterministic step in a composite child DAG.""" + + name: Identifier + handler_name: BindingName + depends_on: tuple[Identifier, ...] = () + + @model_validator(mode="after") + def _validate_dependencies(self) -> Self: + _require_unique(self.depends_on, "step dependencies") + return self + + +class CompositePlan(ContractModel): + """A deterministic DAG for a composite child tool.""" + + steps: Annotated[tuple[CompositeStep, ...], Field(min_length=1)] + required_steps: tuple[Identifier, ...] + optional_steps: tuple[Identifier, ...] + partial_success_policy: Identifier + merge_handler_name: BindingName + + @model_validator(mode="after") + def _validate_plan(self) -> Self: + step_names = tuple(step.name for step in self.steps) + _require_unique(step_names, "composite step names") + _require_unique(self.required_steps, "required_steps") + _require_unique(self.optional_steps, "optional_steps") + + known_steps = set(step_names) + required = set(self.required_steps) + optional = set(self.optional_steps) + if required & optional: + raise ValueError("required_steps and optional_steps must be disjoint") + if required | optional != known_steps: + raise ValueError( + "required_steps and optional_steps must classify every step" + ) + + graph: dict[str, tuple[str, ...]] = {} + for step in self.steps: + unknown = set(step.depends_on) - known_steps + if unknown: + raise ValueError( + f"step {step.name!r} has unknown dependencies: " + f"{sorted(unknown)!r}" + ) + graph[step.name] = step.depends_on + _require_acyclic(graph) + return self + + +class ChildToolDefinition(ContractModel): + """Internal, serializable definition for one exact child tool.""" + + name: Identifier + title: NonEmptyText + canonical_description: NonEmptyText + input_schema: JsonObject + output_schema: JsonObject + handler_name: BindingName + support_contract: ChildSupportContract + authorization_policy: BindingName + annotations: ToolContractAnnotations + composite_plan: CompositePlan | None = None + + @field_validator("input_schema", "output_schema", mode="before") + @classmethod + def _normalize_schema_input(cls, value: Any) -> JsonValue: + return _thaw_json(value) + + @field_validator("input_schema", "output_schema", mode="after") + @classmethod + def _freeze_schemas(cls, value: JsonObject) -> JsonObject: + return _freeze_json_object(value) + + @field_serializer("input_schema", "output_schema") + def _serialize_schemas(self, value: JsonObject) -> JsonObject: + return cast(JsonObject, _thaw_json(value)) + + @model_validator(mode="after") + def _validate_definition(self) -> Self: + if self.canonical_description.startswith("["): + raise ValueError( + "canonical_description must not contain a dynamic status prefix" + ) + _check_object_schema(self.input_schema, "input_schema") + _check_object_schema(self.output_schema, "output_schema") + return self + + +class DomainDefinition(ContractModel): + """Stable definition for one top-level MCP domain tool.""" + + name: Identifier + title: NonEmptyText + description: NonEmptyText + annotations: ToolContractAnnotations + children: Annotated[tuple[ChildToolDefinition, ...], Field(min_length=1)] + enablement_policy: BindingName + discovery_policy: BindingName + max_manifest_bytes: Annotated[ + int, + Field(ge=1, le=MAX_DOMAIN_MANIFEST_BYTES), + ] = MAX_DOMAIN_MANIFEST_BYTES + + @model_validator(mode="after") + def _validate_domain(self) -> Self: + _require_unique( + tuple(child.name for child in self.children), + "child tool names", + ) + return self + + +class Availability(ContractModel): + """Authoritative current availability for one discoverable child.""" + + status: AvailabilityStatus + callable: bool + reason_code: ReasonCode + detected_versions: dict[Identifier, tuple[NonEmptyText, ...]] = Field( + default_factory=dict + ) + active_variant: Identifier | None = None + evidence_sources: tuple[Identifier, ...] = () + limitations: tuple[NonEmptyText, ...] = () + + @field_validator("detected_versions", mode="after") + @classmethod + def _freeze_detected_versions( + cls, + value: dict[str, tuple[str, ...]], + ) -> dict[str, tuple[str, ...]]: + return cast( + dict[str, tuple[str, ...]], + MappingProxyType(dict(value)), + ) + + @field_serializer("detected_versions") + def _serialize_detected_versions( + self, + value: dict[str, tuple[str, ...]], + ) -> dict[str, tuple[str, ...]]: + return dict(value) + + @model_validator(mode="after") + def _validate_availability(self) -> Self: + for component, versions in self.detected_versions.items(): + _require_unique(versions, f"detected_versions[{component}]") + _require_unique(self.evidence_sources, "evidence_sources") + _require_unique(self.limitations, "limitations") + + non_callable_states = { + AvailabilityStatus.UNAVAILABLE, + AvailabilityStatus.MISCONFIGURED, + AvailabilityStatus.UNKNOWN, + } + if self.status in non_callable_states and self.callable: + raise ValueError(f"{self.status.value} availability cannot be callable") + if self.callable and self.active_variant is None: + raise ValueError("callable availability requires an active_variant") + return self + + +class DomainToolRequest(ContractModel): + """Common request shape for discovery and exact child execution.""" + + child_tool: Identifier | None = None + arguments: JsonObject | None = None + manifest_version: NonEmptyText | None = None + + @field_validator("arguments", mode="before") + @classmethod + def _normalize_arguments_input(cls, value: Any) -> JsonValue: + return _thaw_json(value) + + @field_validator("arguments", mode="after") + @classmethod + def _freeze_arguments(cls, value: JsonObject | None) -> JsonObject | None: + return _freeze_json_object(value) if value is not None else None + + @field_serializer("arguments") + def _serialize_arguments(self, value: JsonObject | None) -> JsonObject | None: + return cast(JsonObject | None, _thaw_json(value)) + + @model_validator(mode="after") + def _validate_request_mode(self) -> Self: + if self.child_tool is None and self.arguments: + raise ValueError("arguments require child_tool") + return self + + @property + def is_discovery(self) -> bool: + return self.child_tool is None + + @property + def execution_arguments(self) -> JsonObject: + return dict(self.arguments or {}) + + +class ChildManifestEntry(ContractModel): + """One authorized child entry returned by domain discovery.""" + + name: Identifier + title: NonEmptyText + description: NonEmptyText + input_schema: JsonObject + output_schema: JsonObject + version_support: ChildSupportContract + availability: Availability + annotations: ToolContractAnnotations + + @field_validator("input_schema", "output_schema", mode="before") + @classmethod + def _normalize_schema_input(cls, value: Any) -> JsonValue: + return _thaw_json(value) + + @field_validator("input_schema", "output_schema", mode="after") + @classmethod + def _freeze_schemas(cls, value: JsonObject) -> JsonObject: + return _freeze_json_object(value) + + @field_serializer("input_schema", "output_schema") + def _serialize_schemas(self, value: JsonObject) -> JsonObject: + return cast(JsonObject, _thaw_json(value)) + + @model_validator(mode="after") + def _validate_manifest_entry(self) -> Self: + _check_object_schema(self.input_schema, "input_schema") + _check_object_schema(self.output_schema, "output_schema") + return self + + +class DiscoveryEnvelope(ContractModel): + """Stable structured result for an empty domain discovery call.""" + + mode: Literal["manifest"] + domain: Identifier + manifest_version: NonEmptyText + generated_at: datetime + children: tuple[ChildManifestEntry, ...] + + @model_validator(mode="after") + def _validate_manifest(self) -> Self: + if self.generated_at.utcoffset() is None: + raise ValueError("generated_at must include a timezone") + _require_unique( + tuple(child.name for child in self.children), + "manifest child names", + ) + return self + + +class ExecutionMetadata(ContractModel): + """Bounded metadata for an exact child execution result.""" + + request_id: NonEmptyText + duration_ms: Annotated[float, Field(ge=0)] + source: Identifier + truncated: bool + + +class ExecutionEnvelope(ContractModel): + """Stable structured result for one exact child execution.""" + + mode: Literal["result"] + domain: Identifier + child_tool: Identifier + manifest_version: NonEmptyText + data: JsonValue + metadata: ExecutionMetadata + warnings: tuple[NonEmptyText, ...] = () + + @field_validator("data", mode="before") + @classmethod + def _normalize_data_input(cls, value: Any) -> JsonValue: + return _thaw_json(value) + + @field_validator("data", mode="after") + @classmethod + def _freeze_data(cls, value: JsonValue) -> JsonValue: + return _freeze_json(value) + + @field_serializer("data") + def _serialize_data(self, value: JsonValue) -> JsonValue: + return _thaw_json(value) + + @model_validator(mode="after") + def _validate_warnings(self) -> Self: + _require_unique(self.warnings, "warnings") + return self + + +class StandardError(ContractModel): + """Safe, machine-readable hierarchical tool failure.""" + + code: DomainErrorCode + message: NonEmptyText + retryable: bool + details: JsonObject = Field(default_factory=dict) + + @field_validator("details", mode="before") + @classmethod + def _normalize_details_input(cls, value: Any) -> JsonValue: + return _thaw_json(value) + + @field_validator("details", mode="after") + @classmethod + def _freeze_details(cls, value: JsonObject) -> JsonObject: + return _freeze_json_object(value) + + @field_serializer("details") + def _serialize_details(self, value: JsonObject) -> JsonObject: + return cast(JsonObject, _thaw_json(value)) + + +class ErrorEnvelope(ContractModel): + """Stable structured failure for discovery or execution.""" + + mode: Literal["error"] + domain: Identifier + child_tool: Identifier | None = None + manifest_version: NonEmptyText | None = None + error: StandardError + + +def _require_unique(values: tuple[str, ...], field_name: str) -> None: + if len(values) != len(set(values)): + raise ValueError(f"{field_name} must not contain duplicates") + + +def _check_object_schema(schema: JsonObject, field_name: str) -> None: + if schema.get("type") != "object": + raise ValueError(f"{field_name} root type must be object") + try: + thawed = cast(dict[str, Any], _thaw_json(schema)) + Draft202012Validator.check_schema(thawed) + except SchemaError as exc: + raise ValueError(f"{field_name} is not a valid JSON Schema") from exc + + +def _freeze_json_object(value: JsonObject) -> JsonObject: + return cast(JsonObject, _freeze_json(value)) + + +def _freeze_json(value: JsonValue) -> JsonValue: + if isinstance(value, dict): + return cast( + JsonValue, + MappingProxyType( + {key: _freeze_json(item) for key, item in value.items()} + ), + ) + if isinstance(value, list): + return cast(JsonValue, tuple(_freeze_json(item) for item in value)) + return value + + +def _thaw_json(value: Any) -> JsonValue: + if isinstance(value, Mapping): + return {str(key): _thaw_json(item) for key, item in value.items()} + if isinstance(value, list | tuple): + return [_thaw_json(item) for item in value] + return cast(JsonValue, value) + + +def _require_acyclic(graph: dict[str, tuple[str, ...]]) -> None: + states: dict[str, int] = {} + + def visit(node: str) -> None: + state = states.get(node, 0) + if state == 1: + raise ValueError("composite plan must be acyclic") + if state == 2: + return + states[node] = 1 + for dependency in graph[node]: + visit(dependency) + states[node] = 2 + + for node in graph: + visit(node) + + +__all__ = [ + "MAX_DOMAIN_MANIFEST_BYTES", + "Availability", + "AvailabilityStatus", + "CapabilityVariant", + "ChildManifestEntry", + "ChildSupportContract", + "ChildToolDefinition", + "CompositePlan", + "CompositeStep", + "DiscoveryEnvelope", + "DomainDefinition", + "DomainErrorCode", + "DomainToolRequest", + "ErrorEnvelope", + "ExecutionEnvelope", + "ExecutionMetadata", + "JsonObject", + "StandardError", + "ToolContractAnnotations", + "UnavailableBehavior", +] diff --git a/test/tools/test_domain_models.py b/test/tools/test_domain_models.py new file mode 100644 index 0000000..ecf8b08 --- /dev/null +++ b/test/tools/test_domain_models.py @@ -0,0 +1,545 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import cast + +import pytest +from pydantic import JsonValue, ValidationError + +from doris_mcp_server.tools.domain_models import ( + MAX_DOMAIN_MANIFEST_BYTES, + Availability, + AvailabilityStatus, + CapabilityVariant, + ChildManifestEntry, + ChildSupportContract, + ChildToolDefinition, + CompositePlan, + CompositeStep, + DiscoveryEnvelope, + DomainDefinition, + DomainErrorCode, + DomainToolRequest, + ErrorEnvelope, + ExecutionEnvelope, + ExecutionMetadata, + JsonObject, + StandardError, + ToolContractAnnotations, + UnavailableBehavior, +) + +OBJECT_SCHEMA: JsonObject = { + "type": "object", + "properties": {"database": {"type": "string"}}, + "additionalProperties": False, +} +OUTPUT_SCHEMA: JsonObject = { + "type": "object", + "properties": {"items": {"type": "array"}}, + "required": ["items"], + "additionalProperties": False, +} +READ_ONLY = ToolContractAnnotations( + read_only=True, + idempotent=True, + destructive=False, + open_world=False, + requires_confirmation=False, +) + + +def _variant( + *, + name: str = "information_schema", + probes: tuple[str, ...] = ("information_schema_readable",), + sources: tuple[str, ...] = (), +) -> CapabilityVariant: + return CapabilityVariant( + name=name, + supported_ranges=("project-supported Doris releases",), + required_probes=probes, + source_references=sources, + ) + + +def _support_contract( + *variants: CapabilityVariant, +) -> ChildSupportContract: + return ChildSupportContract( + rule_id="catalog.list_tables.v1", + variants=variants or (_variant(),), + unavailable_behavior=UnavailableBehavior.SHOW_DISABLED, + tested_versions=("4.0.5-rc01",), + ) + + +def _child( + *, + name: str = "list_tables", + handler_name: str = "_list_tables_child", + input_schema: JsonObject | None = None, + output_schema: JsonObject | None = None, + support_contract: ChildSupportContract | None = None, + composite_plan: CompositePlan | None = None, +) -> ChildToolDefinition: + return ChildToolDefinition( + name=name, + title="List Doris Tables", + canonical_description="List visible tables in one Doris database.", + input_schema=input_schema or OBJECT_SCHEMA, + output_schema=output_schema or OUTPUT_SCHEMA, + handler_name=handler_name, + support_contract=support_contract or _support_contract(), + authorization_policy="catalog_metadata", + annotations=READ_ONLY, + composite_plan=composite_plan, + ) + + +def _availability( + *, + status: AvailabilityStatus = AvailabilityStatus.AVAILABLE, + callable_: bool = True, + active_variant: str | None = "information_schema", +) -> Availability: + return Availability( + status=status, + callable=callable_, + reason_code="RUNTIME_PROBE_CONFIRMED", + detected_versions={"fe": ("4.0.5-rc01",)}, + active_variant=active_variant, + evidence_sources=("version", "sql_probe"), + ) + + +def _manifest_child() -> ChildManifestEntry: + child = _child() + return ChildManifestEntry( + name=child.name, + title=child.title, + description="[AVAILABLE] List visible tables in one Doris database.", + input_schema=child.input_schema, + output_schema=child.output_schema, + version_support=child.support_contract, + availability=_availability(), + annotations=child.annotations, + ) + + +def test_domain_definition_is_frozen_and_bounded() -> None: + domain = DomainDefinition( + name="doris_catalog", + title="Doris Catalog", + description="Discover and inspect Doris metadata.", + annotations=READ_ONLY, + children=(_child(),), + enablement_policy="read_only_default", + discovery_policy="catalog_discovery", + ) + + assert domain.max_manifest_bytes == MAX_DOMAIN_MANIFEST_BYTES + with pytest.raises(ValidationError): + domain.name = "doris_query" + + +def test_domain_rejects_duplicate_child_names() -> None: + with pytest.raises(ValidationError, match="child tool names"): + DomainDefinition( + name="doris_catalog", + title="Doris Catalog", + description="Discover and inspect Doris metadata.", + annotations=READ_ONLY, + children=(_child(), _child()), + enablement_policy="read_only_default", + discovery_policy="catalog_discovery", + ) + + +def test_domain_rejects_manifest_budget_above_16_kib() -> None: + with pytest.raises(ValidationError, match="less than or equal"): + DomainDefinition( + name="doris_catalog", + title="Doris Catalog", + description="Discover and inspect Doris metadata.", + annotations=READ_ONLY, + children=(_child(),), + enablement_policy="read_only_default", + discovery_policy="catalog_discovery", + max_manifest_bytes=MAX_DOMAIN_MANIFEST_BYTES + 1, + ) + + +def test_annotations_reject_conflicting_risk_flags() -> None: + with pytest.raises(ValidationError, match="both read-only and destructive"): + ToolContractAnnotations( + read_only=True, + idempotent=False, + destructive=True, + open_world=False, + requires_confirmation=True, + ) + + with pytest.raises(ValidationError, match="must require confirmation"): + ToolContractAnnotations( + read_only=False, + idempotent=False, + destructive=True, + open_world=False, + requires_confirmation=False, + ) + + +def test_capability_variant_requires_evidence_contract() -> None: + with pytest.raises(ValidationError, match="source reference or runtime probe"): + _variant(probes=(), sources=()) + + assert _variant(probes=(), sources=("DORIS_CATALOG_GUIDE",)).name == ( + "information_schema" + ) + + +def test_capability_variant_rejects_duplicate_requirements() -> None: + with pytest.raises(ValidationError, match="required_probes"): + CapabilityVariant( + name="information_schema", + supported_ranges=(">=3.0.0",), + required_probes=("metadata_readable", "metadata_readable"), + ) + + +def test_support_contract_rejects_duplicate_variant_ids() -> None: + with pytest.raises(ValidationError, match="capability variant names"): + _support_contract(_variant(), _variant()) + + +def test_child_rejects_empty_handler_name() -> None: + with pytest.raises(ValidationError, match="handler_name"): + _child(handler_name="") + + +def test_child_rejects_dynamic_status_in_canonical_description() -> None: + payload = _child().model_dump() + payload["canonical_description"] = "[AVAILABLE] List tables." + with pytest.raises(ValidationError, match="dynamic status prefix"): + ChildToolDefinition.model_validate(payload) + + [email protected]("schema_field", ["input_schema", "output_schema"]) +def test_child_rejects_non_object_root_schema(schema_field: str) -> None: + invalid_schema: JsonObject = {"type": "array"} + with pytest.raises(ValidationError, match="root type must be object"): + if schema_field == "input_schema": + _child(input_schema=invalid_schema) + else: + _child(output_schema=invalid_schema) + + +def test_child_rejects_invalid_json_schema() -> None: + with pytest.raises(ValidationError, match="not a valid JSON Schema"): + _child(input_schema={"type": "object", "required": "database"}) + + +def test_child_schemas_are_deeply_immutable() -> None: + child = _child() + with pytest.raises(TypeError): + child.input_schema["title"] = "changed" + + properties = cast(dict[str, JsonValue], child.input_schema["properties"]) + with pytest.raises(TypeError): + properties["database"] = {"type": "integer"} + + assert child.to_wire()["input_schema"] == OBJECT_SCHEMA + + +def test_composite_plan_validates_deterministic_dag() -> None: + plan = CompositePlan( + steps=( + CompositeStep(name="schema", handler_name="_get_schema"), + CompositeStep( + name="comments", + handler_name="_get_comments", + depends_on=("schema",), + ), + ), + required_steps=("schema",), + optional_steps=("comments",), + partial_success_policy="return_required_with_warnings", + merge_handler_name="_merge_table_context", + ) + + assert plan.steps[1].depends_on == ("schema",) + + +def test_composite_plan_rejects_unknown_dependency() -> None: + with pytest.raises(ValidationError, match="unknown dependencies"): + CompositePlan( + steps=( + CompositeStep( + name="comments", + handler_name="_get_comments", + depends_on=("schema",), + ), + ), + required_steps=("comments",), + optional_steps=(), + partial_success_policy="fail_required", + merge_handler_name="_merge", + ) + + +def test_composite_plan_rejects_cycles() -> None: + with pytest.raises(ValidationError, match="must be acyclic"): + CompositePlan( + steps=( + CompositeStep( + name="schema", + handler_name="_get_schema", + depends_on=("comments",), + ), + CompositeStep( + name="comments", + handler_name="_get_comments", + depends_on=("schema",), + ), + ), + required_steps=("schema", "comments"), + optional_steps=(), + partial_success_policy="fail_required", + merge_handler_name="_merge", + ) + + +def test_composite_plan_rejects_overlapping_step_classes() -> None: + with pytest.raises(ValidationError, match="must be disjoint"): + CompositePlan( + steps=(CompositeStep(name="schema", handler_name="_get_schema"),), + required_steps=("schema",), + optional_steps=("schema",), + partial_success_policy="fail_required", + merge_handler_name="_merge", + ) + + +def test_composite_plan_requires_complete_step_classification() -> None: + with pytest.raises(ValidationError, match="classify every step"): + CompositePlan( + steps=(CompositeStep(name="schema", handler_name="_get_schema"),), + required_steps=(), + optional_steps=(), + partial_success_policy="fail_required", + merge_handler_name="_merge", + ) + + [email protected]( + "status", + [ + AvailabilityStatus.UNAVAILABLE, + AvailabilityStatus.MISCONFIGURED, + AvailabilityStatus.UNKNOWN, + ], +) +def test_non_callable_availability_states_fail_closed( + status: AvailabilityStatus, +) -> None: + with pytest.raises(ValidationError, match="cannot be callable"): + _availability(status=status, callable_=True) + + +def test_callable_availability_requires_active_variant() -> None: + with pytest.raises(ValidationError, match="requires an active_variant"): + _availability(callable_=True, active_variant=None) + + +def test_degraded_availability_may_be_callable_or_disabled() -> None: + callable_result = _availability(status=AvailabilityStatus.DEGRADED) + disabled_result = _availability( + status=AvailabilityStatus.DEGRADED, + callable_=False, + active_variant=None, + ) + + assert callable_result.callable is True + assert disabled_result.callable is False + + +def test_domain_request_distinguishes_discovery_and_execution() -> None: + discovery = DomainToolRequest() + empty_discovery = DomainToolRequest(arguments={}) + execution = DomainToolRequest(child_tool="list_tables") + + assert discovery.is_discovery is True + assert empty_discovery.is_discovery is True + assert execution.is_discovery is False + assert execution.execution_arguments == {} + + +def test_domain_request_rejects_arguments_without_child() -> None: + with pytest.raises(ValidationError, match="arguments require child_tool"): + DomainToolRequest(arguments={"database": "analytics"}) + + +def test_request_arguments_and_result_data_are_deeply_immutable() -> None: + request = DomainToolRequest( + child_tool="list_tables", + arguments={"filters": [{"kind": "table"}]}, + ) + arguments = cast(dict[str, JsonValue], request.arguments) + with pytest.raises(TypeError): + arguments["database"] = "changed" + filters = cast(list[JsonValue], arguments["filters"]) + with pytest.raises(TypeError): + filters[0] = {"kind": "view"} + assert request.to_wire()["arguments"] == { + "filters": [{"kind": "table"}] + } + + result = ExecutionEnvelope( + mode="result", + domain="doris_catalog", + child_tool="list_tables", + manifest_version="catalog.7a93f2", + data={"items": [{"name": "orders"}]}, + metadata=ExecutionMetadata( + request_id="req_fixed", + duration_ms=18, + source="doris", + truncated=False, + ), + ) + data = cast(dict[str, JsonValue], result.data) + with pytest.raises(TypeError): + data["items"] = [] + + +def test_discovery_envelope_is_canonical_and_repeatable() -> None: + envelope = DiscoveryEnvelope( + mode="manifest", + domain="doris_catalog", + manifest_version="catalog.7a93f2", + generated_at=datetime(2026, 7, 31, tzinfo=UTC), + children=(_manifest_child(),), + ) + + first = envelope.to_canonical_json() + second = DiscoveryEnvelope.model_validate(envelope.to_wire()).to_canonical_json() + assert first == second + assert first.startswith('{"children":') + assert '"mode":"manifest"' in first + + +def test_discovery_envelope_rejects_naive_timestamp() -> None: + with pytest.raises(ValidationError, match="must include a timezone"): + DiscoveryEnvelope( + mode="manifest", + domain="doris_catalog", + manifest_version="catalog.7a93f2", + generated_at=datetime(2026, 7, 31), + children=(), + ) + + +def test_discovery_envelope_rejects_duplicate_child_ids() -> None: + with pytest.raises(ValidationError, match="manifest child names"): + DiscoveryEnvelope( + mode="manifest", + domain="doris_catalog", + manifest_version="catalog.7a93f2", + generated_at=datetime(2026, 7, 31, tzinfo=UTC), + children=(_manifest_child(), _manifest_child()), + ) + + +def test_execution_envelope_has_stable_wire_shape() -> None: + result = ExecutionEnvelope( + mode="result", + domain="doris_catalog", + child_tool="list_tables", + manifest_version="catalog.7a93f2", + data={"items": [{"name": "orders"}]}, + metadata=ExecutionMetadata( + request_id="req_fixed", + duration_ms=18, + source="doris", + truncated=False, + ), + ) + + assert result.to_wire() == { + "mode": "result", + "domain": "doris_catalog", + "child_tool": "list_tables", + "manifest_version": "catalog.7a93f2", + "data": {"items": [{"name": "orders"}]}, + "metadata": { + "request_id": "req_fixed", + "duration_ms": 18.0, + "source": "doris", + "truncated": False, + }, + "warnings": [], + } + + +def test_standard_error_envelope_uses_fixed_error_codes() -> None: + result = ErrorEnvelope( + mode="error", + domain="doris_catalog", + child_tool="list_tables", + manifest_version="catalog.7a93f2", + error=StandardError( + code=DomainErrorCode.CHILD_MANIFEST_STALE, + message="The domain manifest changed.", + retryable=True, + details={"rediscover": True}, + ), + ) + + error = cast(dict[str, JsonValue], result.to_wire()["error"]) + assert error["code"] == "CHILD_MANIFEST_STALE" + with pytest.raises(ValidationError): + StandardError.model_validate( + { + "code": "NOT_A_STANDARD_ERROR", + "message": "Invalid error code.", + "retryable": False, + } + ) + + +def test_wire_schema_is_deterministic_and_forbids_extra_fields() -> None: + first = DiscoveryEnvelope.wire_schema() + second = DiscoveryEnvelope.wire_schema() + + assert first == second + assert first["additionalProperties"] is False + assert "generated_at" in first["properties"] + + +def test_models_reject_unknown_fields() -> None: + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + Availability.model_validate( + { + "status": AvailabilityStatus.UNKNOWN, + "callable": False, + "reason_code": "PROBE_FAILED", + "guessed_version": "4.0.5", + } + ) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
