契约优先：接口定义先行

“

系列：S10 设计模式 · 第 12 篇 | 难度：高级 | 阅读时间：18 分钟

#TL;DR

契约优先（Contract-First）模式要求在编写实现代码之前先定义接口契约。coomia-dip 使用 Protobuf 定义 gRPC 服务契约，使用 OpenAPI 定义 REST API 契约，使用 JSON Schema 定义 Ontology Schema 契约。
契约是 Layer 之间的协作边界——只要契约不变，各 Layer 可以独立开发、测试和部署。
coomia-dip 实现了契约兼容性检查（向后兼容、向前兼容）、契约版本管理和自动代码生成，确保多团队协作的稳定性。

#引言：先有接口还是先有实现

在多团队协作开发中，一个典型的问题是：团队 A 开发 Control Layer，团队 B 开发 Data Layer，他们之间通过 gRPC 通信。如果没有预先约定的接口定义：

Code

团队 A 认为 CreateObject 返回 object_id: string
团队 B 认为 CreateObject 返回 id: int64
→ 集成时发现不兼容，返工

契约优先模式的核心：先定义接口，再写实现。接口定义就是团队间的协作契约。

#一、Protobuf 作为 gRPC 契约

#1.1 Layer 间 gRPC 服务定义

coomia-dip 所有 Layer 间通信使用 Protobuf 定义的 gRPC 服务：

PROTOBUF

// proto/ontology/v1/ontology_service.proto
syntax = "proto3";
package onto.ontology.v1;

import "google/protobuf/timestamp.proto";
import "google/protobuf/struct.proto";

// Ontology Service - Control Layer
service OntologyService {
    // ObjectType CRUD
    rpc CreateObjectType(CreateObjectTypeRequest) returns (CreateObjectTypeResponse);
    rpc GetObjectType(GetObjectTypeRequest) returns (ObjectType);
    rpc ListObjectTypes(ListObjectTypesRequest) returns (ListObjectTypesResponse);
    rpc UpdateObjectType(UpdateObjectTypeRequest) returns (ObjectType);
    rpc DeleteObjectType(DeleteObjectTypeRequest) returns (DeleteObjectTypeResponse);

    // LinkType CRUD
    rpc CreateLinkType(CreateLinkTypeRequest) returns (CreateLinkTypeResponse);
    rpc GetLinkType(GetLinkTypeRequest) returns (LinkType);
    rpc ListLinkTypes(ListLinkTypesRequest) returns (ListLinkTypesResponse);

    // Action management
    rpc RegisterAction(RegisterActionRequest) returns (RegisterActionResponse);
    rpc ExecuteAction(ExecuteActionRequest) returns (ExecuteActionResponse);
}

message CreateObjectTypeRequest {
    string name = 1;
    string display_name = 2;
    string description = 3;
    google.protobuf.Struct schema = 4;
    string tenant_id = 5;
    string world_id = 6;
}

message CreateObjectTypeResponse {
    string type_id = 1;
    int32 version = 2;
    google.protobuf.Timestamp created_at = 3;
}

message ObjectType {
    string type_id = 1;
    string name = 2;
    string display_name = 3;
    string description = 4;
    google.protobuf.Struct schema = 5;
    int32 version = 6;
    string tenant_id = 7;
    string world_id = 8;
    google.protobuf.Timestamp created_at = 9;
    google.protobuf.Timestamp updated_at = 10;
}

#1.2 契约目录结构

Code

proto/
├── ontology/v1/
│   ├── ontology_service.proto    # Ontology 服务定义
│   ├── object_types.proto        # ObjectType 消息定义
│   └── link_types.proto          # LinkType 消息定义
├── data/v1/
│   ├── data_service.proto        # Data Layer 服务定义
│   ├── query_service.proto       # 查询服务定义
│   └── storage_types.proto       # 存储类型定义
├── reasoning/v1/
│   ├── reasoning_service.proto   # 推理服务定义
│   └── rule_types.proto          # 规则类型定义
├── agent/v1/
│   ├── agent_service.proto       # Agent 服务定义
│   └── task_types.proto          # 任务类型定义
└── common/v1/
    ├── error.proto               # 统一错误定义
    ├── pagination.proto          # 分页定义
    └── metadata.proto            # 元数据定义

#二、契约兼容性管理

#2.1 兼容性规则

Python

class CompatibilityChecker:
    """Check Protobuf schema compatibility."""

    def check_backward_compatibility(
        self, old_proto: str, new_proto: str
    ) -> list[str]:
        """Check if new proto is backward compatible with old."""
        violations = []
        old_desc = self._parse(old_proto)
        new_desc = self._parse(new_proto)

        # 规则 1: 不能删除已有字段
        for field in old_desc.fields:
            if field.number not in [f.number for f in new_desc.fields]:
                violations.append(
                    f"Field {field.name} (number {field.number}) was removed"
                )

        # 规则 2: 不能修改已有字段的类型
        for old_field in old_desc.fields:
            for new_field in new_desc.fields:
                if old_field.number == new_field.number:
                    if old_field.type != new_field.type:
                        violations.append(
                            f"Field {old_field.name}: type changed from "
                            f"{old_field.type} to {new_field.type}"
                        )

        # 规则 3: 不能修改已有字段的编号
        old_names = {f.name: f.number for f in old_desc.fields}
        new_names = {f.name: f.number for f in new_desc.fields}
        for name in set(old_names) & set(new_names):
            if old_names[name] != new_names[name]:
                violations.append(
                    f"Field {name}: number changed from "
                    f"{old_names[name]} to {new_names[name]}"
                )

        # 规则 4: 不能删除 RPC 方法
        for rpc in old_desc.services[0].methods:
            if rpc.name not in [m.name for m in new_desc.services[0].methods]:
                violations.append(f"RPC method {rpc.name} was removed")

        return violations

    def check_forward_compatibility(
        self, old_proto: str, new_proto: str
    ) -> list[str]:
        """Check if old consumers can handle new proto messages."""
        violations = []
        # 新增的 required 字段会破坏前向兼容
        old_desc = self._parse(old_proto)
        new_desc = self._parse(new_proto)

        new_fields = {f.number for f in new_desc.fields}
        old_fields = {f.number for f in old_desc.fields}
        added = new_fields - old_fields

        for new_field in new_desc.fields:
            if new_field.number in added and new_field.label == "required":
                violations.append(
                    f"New required field {new_field.name} breaks forward compatibility"
                )

        return violations

#2.2 CI/CD 中的契约检查

Python

class ContractCICheck:
    """Run contract compatibility checks in CI/CD pipeline."""

    async def run_checks(self, pr_branch: str, main_branch: str) -> dict:
        """Compare proto files between PR and main branch."""
        changed_protos = await self._git.diff_files(
            main_branch, pr_branch, pattern="*.proto"
        )

        results = {"passed": True, "violations": []}

        for proto_path in changed_protos:
            old_content = await self._git.get_file(main_branch, proto_path)
            new_content = await self._git.get_file(pr_branch, proto_path)

            if old_content:  # 修改已有 proto
                violations = self._checker.check_backward_compatibility(
                    old_content, new_content
                )
                if violations:
                    results["passed"] = False
                    results["violations"].extend([
                        {"file": proto_path, "violation": v}
                        for v in violations
                    ])

        return results

#三、OpenAPI 契约

#3.1 REST API 定义

对外暴露的 REST API 使用 OpenAPI 3.0 定义：

YAML

# openapi/v1/ontology-api.yaml
openapi: 3.0.3
info:
  title: coomia-dip Ontology API
  version: 1.0.0
  description: REST API for coomia-dip Ontology operations

paths:
  /api/v1/ontology/types:
    get:
      operationId: listObjectTypes
      summary: List all ObjectTypes
      parameters:
        - name: tenant_id
          in: header
          required: true
          schema:
            type: string
        - name: page_size
          in: query
          schema:
            type: integer
            default: 20
      responses:
        '200':
          description: Successful response
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ListObjectTypesResponse'
    post:
      operationId: createObjectType
      summary: Create a new ObjectType
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateObjectTypeRequest'
      responses:
        '201':
          description: Created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ObjectType'

components:
  schemas:
    ObjectType:
      type: object
      required: [type_id, name, schema]
      properties:
        type_id:
          type: string
        name:
          type: string
        display_name:
          type: string
        description:
          type: string
        schema:
          type: object
        version:
          type: integer

#3.2 从 OpenAPI 自动生成代码

Python

class OpenAPICodeGenerator:
    """Generate client and server code from OpenAPI spec."""

    async def generate_python_client(
        self, spec_path: str, output_dir: str
    ) -> None:
        """Generate Python SDK client from OpenAPI spec."""
        spec = self._load_spec(spec_path)

        for path, methods in spec["paths"].items():
            for method, operation in methods.items():
                self._generate_method(
                    operation["operationId"],
                    method.upper(),
                    path,
                    operation,
                    output_dir,
                )

    async def generate_typescript_client(
        self, spec_path: str, output_dir: str
    ) -> None:
        """Generate TypeScript OSDK client from OpenAPI spec."""
        pass

    def _generate_method(
        self, operation_id: str, method: str, path: str,
        operation: dict, output_dir: str
    ) -> None:
        """Generate a single API method."""
        pass

#四、Ontology Schema 契约

#4.1 JSON Schema 定义 ObjectType

Python

class OntologySchemaContract:
    """Define and validate Ontology Schema contracts."""

    OBJECT_TYPE_META_SCHEMA = {
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "type": "object",
        "required": ["name", "properties"],
        "properties": {
            "name": {"type": "string", "pattern": "^[A-Z][a-zA-Z0-9]+$"},
            "display_name": {"type": "string"},
            "description": {"type": "string"},
            "properties": {
                "type": "object",
                "additionalProperties": {
                    "$ref": "#/$defs/PropertyDefinition"
                },
            },
            "primary_key": {"type": "string"},
            "state_machine": {"$ref": "#/$defs/StateMachineDefinition"},
        },
        "$defs": {
            "PropertyDefinition": {
                "type": "object",
                "required": ["type"],
                "properties": {
                    "type": {
                        "type": "string",
                        "enum": ["string", "integer", "float", "boolean",
                                 "datetime", "array", "object", "reference"],
                    },
                    "required": {"type": "boolean", "default": False},
                    "description": {"type": "string"},
                    "default": {},
                    "constraints": {"type": "object"},
                },
            },
        },
    }

    async def validate_schema(self, schema: dict) -> list[str]:
        """Validate an ObjectType schema against the meta-schema."""
        import jsonschema
        errors = []
        try:
            jsonschema.validate(schema, self.OBJECT_TYPE_META_SCHEMA)
        except jsonschema.ValidationError as e:
            errors.append(str(e.message))
        return errors

    async def check_evolution_compatibility(
        self, old_schema: dict, new_schema: dict
    ) -> list[str]:
        """Check if schema evolution is compatible."""
        errors = []
        old_props = old_schema.get("properties", {})
        new_props = new_schema.get("properties", {})

        # 不能删除已有属性
        removed = set(old_props) - set(new_props)
        if removed:
            errors.append(f"Properties removed: {removed}")

        # 不能修改已有属性的类型
        for name in set(old_props) & set(new_props):
            if old_props[name].get("type") != new_props[name].get("type"):
                errors.append(
                    f"Property '{name}' type changed from "
                    f"'{old_props[name].get('type')}' to '{new_props[name].get('type')}'"
                )

        # 新增 required 属性必须有默认值
        for name in set(new_props) - set(old_props):
            if new_props[name].get("required") and "default" not in new_props[name]:
                errors.append(
                    f"New required property '{name}' must have a default value"
                )

        return errors

#五、契约测试

#5.1 消费者驱动的契约测试

Python

class ContractTest:
    """Consumer-driven contract testing."""

    async def verify_provider(
        self,
        provider_name: str,
        contract: dict,
    ) -> dict:
        """Verify that a provider satisfies the contract."""
        results = {"passed": True, "failures": []}

        for interaction in contract["interactions"]:
            # 发送请求
            response = await self._client.request(
                method=interaction["request"]["method"],
                path=interaction["request"]["path"],
                body=interaction["request"].get("body"),
                headers=interaction["request"].get("headers"),
            )

            # 验证响应
            expected = interaction["response"]
            if response.status != expected["status"]:
                results["passed"] = False
                results["failures"].append({
                    "interaction": interaction["description"],
                    "expected_status": expected["status"],
                    "actual_status": response.status,
                })

            # 验证响应体结构
            if "body" in expected:
                schema_valid = self._validate_response_schema(
                    response.body, expected["body"]
                )
                if not schema_valid:
                    results["passed"] = False
                    results["failures"].append({
                        "interaction": interaction["description"],
                        "error": "Response body schema mismatch",
                    })

        return results

#5.2 契约快照测试

Python

class ContractSnapshotTest:
    """Snapshot testing for contract stability."""

    async def capture_snapshot(self, proto_path: str) -> dict:
        """Capture a snapshot of proto definitions."""
        return {
            "path": proto_path,
            "services": self._extract_services(proto_path),
            "messages": self._extract_messages(proto_path),
            "captured_at": datetime.utcnow().isoformat(),
        }

    async def compare_with_snapshot(
        self, current_proto: str, snapshot: dict
    ) -> dict:
        """Compare current proto with saved snapshot."""
        current = await self.capture_snapshot(current_proto)
        changes = {
            "added_services": [],
            "removed_services": [],
            "added_messages": [],
            "removed_messages": [],
            "modified_messages": [],
        }
        # ... 比较逻辑
        return changes

#六、自动代码生成

#6.1 从 Protobuf 生成多语言代码

Python

class MultiLanguageCodeGen:
    """Generate code from Protobuf definitions."""

    GENERATORS = {
        "python": "grpcio-tools",
        "java": "protoc-gen-grpc-java",
        "typescript": "protoc-gen-ts",
    }

    async def generate_all(self, proto_dir: str, output_dir: str) -> dict:
        """Generate client code for all supported languages."""
        results = {}
        for lang, tool in self.GENERATORS.items():
            result = await self._generate(proto_dir, output_dir, lang, tool)
            results[lang] = result
        return results

#Key Takeaways

接口先行：在写实现代码之前先定义 Protobuf/OpenAPI 契约
兼容性保证：CI/CD 管道自动检查契约的向后/向前兼容性
多层契约：gRPC 契约（Layer 间）、REST 契约（对外 API）、Schema 契约（Ontology 模型）
契约测试：消费者驱动的契约测试确保提供者满足消费者的期望
自动代码生成：从契约定义自动生成多语言的客户端和服务端代码
Schema 演进：Ontology Schema 的变更必须通过兼容性检查

#Next Article

S10-13: 自动供给模式：资源按需创建

#Tags

#设计模式 #契约优先 #ContractFirst #Protobuf #OpenAPI #gRPC #兼容性 #代码生成 #契约测试