This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new 99b81a0 feat: add hierarchical domain contract models (#165)
99b81a0 is described below
commit 99b81a0967a8ea85f97a1cf394c9743d73263337
Author: Yijia Su <[email protected]>
AuthorDate: Fri Jul 31 01:32:51 2026 +0800
feat: add hierarchical domain contract models (#165)
Define the immutable machine contracts required by the 1.0.0 hierarchical
tool architecture.
---
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]
