ln-643-api-contract-auditor

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

API Contract Auditor (L3 Worker)

API Contract Auditor (L3 Worker)

Specialized worker auditing API contracts, method signatures at service boundaries, and DTO usage patterns.
专注于审计API契约、服务边界处的方法签名以及DTO使用模式的工作器。

Purpose & Scope

目标与范围

  • Worker in ln-640 coordinator pipeline - invoked by ln-640-pattern-evolution-auditor
  • Audit API contracts at architecture level (service boundaries, layer separation)
  • Check layer leakage, DTO patterns, error contract consistency
  • Return structured analysis with 4 scores (compliance, completeness, quality, implementation)
  • ln-640协调器流水线中的工作器 - 由ln-640-pattern-evolution-auditor调用
  • 在架构层面(服务边界、分层隔离)审计API契约
  • 检查层泄漏、DTO模式、错误契约一致性
  • 通过4维度评分(合规性、完整性、质量、实现)返回结构化分析结果

Input (from ln-640 coordinator)

输入(来自ln-640协调器)

- pattern: "API Contracts"     # Pattern name
- locations: string[]          # Service/API directories
- adr_reference: string        # Path to related ADR
- bestPractices: object        # Best practices from MCP Ref/Context7
- pattern: "API Contracts"     # 模式名称
- locations: string[]          # 服务/API目录
- adr_reference: string        # 相关ADR的路径
- bestPractices: object        # 来自MCP Ref/Context7的最佳实践

Workflow

工作流程

Phase 1: Discover Service Boundaries

阶段1:发现服务边界

1. Find API layer: Glob("**/api/**/*.py", "**/routes/**/*.ts", "**/controllers/**/*.ts")
2. Find service layer: Glob("**/services/**/*.py", "**/services/**/*.ts")
3. Find domain layer: Glob("**/domain/**/*.py", "**/models/**/*.py")
4. Map: which services are called by which API endpoints
1. 查找API层:Glob("**/api/**/*.py", "**/routes/**/*.ts", "**/controllers/**/*.ts")
2. 查找服务层:Glob("**/services/**/*.py", "**/services/**/*.ts")
3. 查找领域层:Glob("**/domain/**/*.py", "**/models/**/*.py")
4. 映射:哪些服务被哪些API端点调用

Phase 2: Analyze Contracts

阶段2:分析契约

FOR EACH service file:
  Extract public method signatures (name, params, return type)
  Check each audit rule
  Collect findings
FOR EACH service file:
  Extract public method signatures (name, params, return type)
  Check each audit rule
  Collect findings

Phase 3: Calculate 4 Scores

阶段3:计算4维度评分

Compliance Score (0-100):
IF no layer leakage (HTTP types in service): +35
IF consistent error handling pattern: +25
IF follows project naming conventions: +20
IF no entity leakage to API: +20
Completeness Score (0-100):
IF all service methods have typed params: +30
IF all service methods have typed returns: +30
IF DTOs defined for complex data: +20
IF error types documented/typed: +20
Quality Score (0-100):
IF no boolean flag params in service methods: +25
IF no methods with >5 params without DTO: +25
IF consistent naming across module: +25
IF no redundant overloads: +25
Implementation Score (0-100):
IF DTOs/schemas exist and are used: +30
IF type annotations present (Python) or interfaces (TS): +25
IF validation at boundaries (Pydantic, Zod, etc.): +25
IF API response DTOs separate from domain models: +20
合规性评分(0-100):
IF no layer leakage (HTTP types in service): +35
IF consistent error handling pattern: +25
IF follows project naming conventions: +20
IF no entity leakage to API: +20
完整性评分(0-100):
IF all service methods have typed params: +30
IF all service methods have typed returns: +30
IF DTOs defined for complex data: +20
IF error types documented/typed: +20
质量评分(0-100):
IF no boolean flag params in service methods: +25
IF no methods with >5 params without DTO: +25
IF consistent naming across module: +25
IF no redundant overloads: +25
实现评分(0-100):
IF DTOs/schemas exist and are used: +30
IF type annotations present (Python) or interfaces (TS): +25
IF validation at boundaries (Pydantic, Zod, etc.): +25
IF API response DTOs separate from domain models: +20

Phase 4: Calculate Overall Score

阶段4:计算整体评分

overall_score = average(compliance, completeness, quality, implementation) / 10
Example: (65 + 70 + 55 + 80) / 4 / 10 = 6.75
overall_score = average(compliance, completeness, quality, implementation) / 10
Example: (65 + 70 + 55 + 80) / 4 / 10 = 6.75
整体评分 = (合规性 + 完整性 + 质量 + 实现)的平均值 / 10 示例:(65 + 70 + 55 + 80) / 4 / 10 = 6.75

Phase 5: Return Result

阶段5:返回结果

json
{
  "pattern": "API Contracts",
  "overall_score": 6.75,
  "scores": {
    "compliance": 65,
    "completeness": 70,
    "quality": 55,
    "implementation": 80
  },
  "checks": [
    {"id": "layer_leakage", "name": "Layer Leakage", "status": "failed", "details": "Service accepts parsed_body: dict in 3 methods"},
    {"id": "missing_dto", "name": "Missing DTO", "status": "warning", "details": "4 params repeated in 2 methods without grouping DTO"},
    {"id": "entity_leakage", "name": "Entity Leakage", "status": "passed", "details": "All API endpoints use response DTOs"},
    {"id": "error_contracts", "name": "Error Contracts", "status": "warning", "details": "Mixed patterns: raise + return None in UserService"},
    {"id": "redundant_overloads", "name": "Redundant Overloads", "status": "passed", "details": "No redundant method pairs found"}
  ],
  "codeReferences": ["app/services/translation/", "app/api/v1/"],
  "issues": [...],
  "gaps": {...},
  "recommendations": [...]
}
json
{
  "pattern": "API Contracts",
  "overall_score": 6.75,
  "scores": {
    "compliance": 65,
    "completeness": 70,
    "quality": 55,
    "implementation": 80
  },
  "checks": [
    {"id": "layer_leakage", "name": "Layer Leakage", "status": "failed", "details": "Service accepts parsed_body: dict in 3 methods"},
    {"id": "missing_dto", "name": "Missing DTO", "status": "warning", "details": "4 params repeated in 2 methods without grouping DTO"},
    {"id": "entity_leakage", "name": "Entity Leakage", "status": "passed", "details": "All API endpoints use response DTOs"},
    {"id": "error_contracts", "name": "Error Contracts", "status": "warning", "details": "Mixed patterns: raise + return None in UserService"},
    {"id": "redundant_overloads", "name": "Redundant Overloads", "status": "passed", "details": "No redundant method pairs found"}
  ],
  "codeReferences": ["app/services/translation/", "app/api/v1/"],
  "issues": [...],
  "gaps": {...},
  "recommendations": [...]
}

Audit Rules

审计规则

1. Layer Leakage in Signatures

1. 签名中的层泄漏

What: Service/domain method accepts HTTP-layer types (Request, parsed body dict, headers)
Detection:
  • Grep for service/domain methods accepting
    parsed_body: dict
    ,
    request: Request
    ,
    form_data: dict
  • Pattern:
    def translate(self, parsed_body: dict)
    in service layer
  • Pattern:
    from fastapi import Request
    imported in service/domain files
  • Check: service methods should accept domain-typed params, not HTTP artifacts
Severity:
  • HIGH: Service method accepts raw HTTP dict (tight coupling to transport layer)
  • MEDIUM: Service method accepts
    Request
    object (should receive extracted fields)
Recommendation: Create domain DTO (dataclass/Pydantic model) accepted by service; API layer extracts and maps
Effort: M (create DTO, update service signature, update callers)
内容: 服务/领域方法接收HTTP层类型(Request、解析后的body字典、headers)
检测方式:
  • 查找服务/领域方法中接收
    parsed_body: dict
    request: Request
    form_data: dict
    的情况
  • 示例模式:服务层中的
    def translate(self, parsed_body: dict)
  • 示例模式:服务/领域文件中导入
    from fastapi import Request
  • 检查点:服务方法应接收领域类型的参数,而非HTTP相关对象
严重程度:
  • 高: 服务方法接收原始HTTP字典(与传输层紧耦合)
  • 中: 服务方法接收
    Request
    对象(应接收提取后的字段)
建议: 创建领域DTO(数据类/Pydantic模型)供服务接收;API层负责提取并映射参数
工作量: 中等(创建DTO、更新服务签名、更新调用方)

2. Missing DTO for Grouped Parameters

2. 分组参数缺失DTO

What: >=4 related parameters always passed together across multiple methods
Detection:
  • Find parameter groups: same 4+ params appear in >=2 method signatures
  • Example:
    (account_id, engine, source_lang, target_lang, ...)
    repeated across methods
  • Pattern: params that logically form a "context" or "request" but are passed individually
Severity:
  • MEDIUM: 4-6 related params in >=2 methods (DTO recommended)
  • LOW: 4-6 related params in single method (borderline)
Recommendation: Create dataclass/NamedTuple grouping related params:
TranslationContext(engine, source_lang, target_lang, ...)
Effort: M (create DTO, refactor signatures, update callers)
内容: 多个方法中始终一起传递的相关参数数量≥4个
检测方式:
  • 查找参数组:相同的4个以上参数出现在≥2个方法签名中
  • 示例:
    (account_id, engine, source_lang, target_lang, ...)
    在多个方法中重复出现
  • 示例模式:在逻辑上构成"上下文"或"请求"的参数被单独传递
严重程度:
  • 中: 4-6个相关参数出现在≥2个方法中(推荐使用DTO)
  • 低: 4-6个相关参数仅出现在单个方法中(临界情况)
建议: 创建数据类/NamedTuple对相关参数进行分组:
TranslationContext(engine, source_lang, target_lang, ...)
工作量: 中等(创建DTO、重构签名、更新调用方)

3. Entity Leakage to API

3. 实体向API泄漏

What: ORM/domain entity returned directly from API endpoint without response DTO
Detection:
  • API endpoint returns ORM model object directly
  • Pattern:
    return user
    in endpoint where
    user
    is SQLAlchemy model
  • Pattern:
    return {"user": user.__dict__}
    (manual serialization of ORM entity)
  • Check: look for Pydantic
    response_model
    or explicit serialization schema
Severity:
  • HIGH: ORM entity returned with all fields (exposes internal structure, password hashes, etc.)
  • MEDIUM: ORM entity returned with manual field selection (fragile, no schema)
Recommendation: Create response DTO (Pydantic BaseModel) mapping only needed fields; use
response_model
in FastAPI
Effort: M (create response DTO, update endpoint)
内容: ORM/领域实体直接从API端点返回,未使用响应DTO
检测方式:
  • API端点直接返回ORM模型对象
  • 示例模式:端点中
    return user
    ,其中
    user
    是SQLAlchemy模型
  • 示例模式:
    return {"user": user.__dict__}
    (手动序列化ORM实体)
  • 检查点:查找是否存在Pydantic
    response_model
    或显式的序列化schema
严重程度:
  • 高: 返回包含所有字段的ORM实体(暴露内部结构、密码哈希等)
  • 中: 手动选择字段返回ORM实体(脆弱、无schema约束)
建议: 创建仅包含所需字段的响应DTO(Pydantic BaseModel);在FastAPI中使用
response_model
工作量: 中等(创建响应DTO、更新端点)

4. Inconsistent Error Contracts

4. 不一致的错误契约

What: Mixed error handling patterns within same service (some raise, some return None, some return Result)
Detection:
  • Analyze all public methods in a service class
  • Check error handling:
    raise Exception
    ,
    return None
    ,
    return {"error": ...}
    ,
    return Result
  • Flag if >1 pattern used within same service
Severity:
  • MEDIUM: Mixed patterns within service (caller must guess error handling)
  • LOW: Inconsistency across different services (less critical if each is internally consistent)
Recommendation: Standardize: either raise domain exceptions (recommended for Python) or use Result type throughout
Effort: M (standardize error pattern across service methods)
内容: 同一服务中混合使用多种错误处理模式(部分抛出异常、部分返回None、部分返回Result)
检测方式:
  • 分析服务类中的所有公共方法
  • 检查错误处理方式:
    raise Exception
    return None
    return {"error": ...}
    return Result
  • 若同一服务中使用超过1种模式则标记
严重程度:
  • 中: 服务内部混合使用多种模式(调用方需猜测错误处理方式)
  • 低: 不同服务之间存在不一致性(若每个服务内部一致则影响较小)
建议: 标准化处理:要么抛出领域异常(Python推荐方式),要么全程使用Result类型
工作量: 中等(在服务方法中标准化错误模式)

5. Redundant Method Overloads

5. 冗余的方法重载

What: Two methods differ only in 1-2 parameters that could be optional
Detection:
  • Find pairs:
    get_user(id)
    +
    get_user_with_profile(id)
    → could be
    get_user(id, include_profile=False)
  • Find pairs:
    translate(text)
    +
    translate_with_qe(text)
    → could be
    translate(text, enable_qe=False)
  • Pattern: method names with
    _with_
    ,
    _and_
    ,
    _full
    suffix duplicating base method
Severity:
  • LOW: 1-2 redundant overload pairs (minor DRY violation)
  • MEDIUM: >3 redundant overload pairs in same service (maintenance burden)
Recommendation: Merge into single method with optional parameters or strategy/config object
Effort: S-M (merge methods, update callers)
内容: 两个方法仅在1-2个参数上存在差异,而这些参数可设为可选参数
检测方式:
  • 查找方法对:
    get_user(id)
    +
    get_user_with_profile(id)
    → 可合并为
    get_user(id, include_profile=False)
  • 查找方法对:
    translate(text)
    +
    translate_with_qe(text)
    → 可合并为
    translate(text, enable_qe=False)
  • 示例模式:方法名带有
    _with_
    _and_
    _full
    后缀,重复基础方法的功能
严重程度:
  • 低: 1-2对冗余重载(轻微违反DRY原则)
  • 中: 同一服务中存在超过3对冗余重载(增加维护负担)
建议: 合并为单个带可选参数的方法,或使用策略/配置对象
工作量: 小-中等(合并方法、更新调用方)

Critical Rules

关键规则

  • Architecture-level only: Focus on service boundaries, not internal implementation
  • Read before score: Never score without reading actual service code
  • Best practices comparison: Use bestPractices from coordinator for framework-specific patterns
  • Code references: Include file paths for all findings
  • One pattern only: Analyze "API Contracts" pattern as a whole
  • 仅关注架构层面: 聚焦服务边界,而非内部实现细节
  • 先阅读代码再评分: 未阅读实际服务代码绝不评分
  • 对比最佳实践: 使用协调器提供的bestPractices来适配框架特定模式
  • 代码引用: 所有审计结果都需包含文件路径
  • 仅分析单一模式: 将"API Contracts"模式作为一个整体进行分析

Definition of Done

完成标准

  • Service boundaries discovered (API, service, domain layers)
  • Method signatures extracted and analyzed
  • All 5 checks completed:
    • layer leakage, missing DTOs, entity leakage, error contracts, redundant overloads
  • 4 scores calculated with justification
  • Issues identified with severity, category, suggestion, effort
  • Gaps documented
  • Structured result returned to coordinator
  • 已发现服务边界(API层、服务层、领域层)
  • 已提取并分析方法签名
  • 已完成全部5项检查:
    • 层泄漏、缺失DTO、实体泄漏、错误契约、冗余重载
  • 已计算4维度评分并提供依据
  • 已识别问题并标注严重程度、类别、建议、工作量
  • 已记录差距
  • 已向协调器返回结构化结果

Reference Files

参考文件

  • Scoring rules:
    ../ln-640-pattern-evolution-auditor/references/scoring_rules.md
  • Common patterns:
    ../ln-640-pattern-evolution-auditor/references/common_patterns.md

Version: 1.0.0 Last Updated: 2026-02-04
  • 评分规则:
    ../ln-640-pattern-evolution-auditor/references/scoring_rules.md
  • 常见模式:
    ../ln-640-pattern-evolution-auditor/references/common_patterns.md

版本: 1.0.0 最后更新: 2026-02-04