PMFlow/scripts/openclaw_routing_lint.py

#!/usr/bin/env python3
"""OpenClaw 路由规则静态校验（PMFlow 三引擎版）。"""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any, Optional

VALID_PROVIDERS = {"glm", "openai", "deepseek", "ollama"}
VALID_TIERS = {"low", "high", "safety"}

LOW_REQUIRED = {
    "feishu_level_classification",
    "feishu_signal_extraction",
    "rag_intent_classification",
    "classify_intent",
    "feishu_humanized_reply",
}

HIGH_REQUIRED = {
    "adaptive_rag_answer",
    "feishu_project_prefill",
    "feishu_task_orchestration",
    "generate_task_breakdown",
    "generate_charter_from_signal",
    "decompose_tasks_from_charter",
    "recommend_assignees_for_task",
    "evaluate_model_requirement",
    "calculate_alignment_score",
    "detect_cognitive_conflicts",
    "rag_db_project_semantic_retrieval",
    "workload_optimize_prompt",
    "analyze_cost_roi",
}


def _rule_matches_function(rule: dict[str, Any], function_name: str) -> bool:
    if bool(rule.get("match_all", False)):
        return True
    fn = str(function_name or "").strip().lower()
    if not fn:
        return False
    exact = str(rule.get("function_exact") or "").strip().lower()
    if exact and fn == exact:
        return True
    prefix = str(rule.get("function_prefix") or "").strip().lower()
    if prefix in {"*", "all"}:
        return True
    if prefix and fn.startswith(prefix):
        return True
    match_any = rule.get("match_any")
    if isinstance(match_any, list):
        for hint in match_any:
            token = str(hint or "").strip().lower()
            if token and token in fn:
                return True
    return False


def _resolve_primary_tier(function_name: str, rules: list[dict[str, Any]]) -> Optional[str]:
    scored: list[tuple[int, int, str]] = []
    for idx, row in enumerate(rules):
        if not bool(row.get("enabled", True)):
            continue
        tier = str(row.get("tier") or "").strip().lower()
        if tier not in {"low", "high"}:
            continue
        if not _rule_matches_function(row, function_name):
            continue
        scored.append((int(row.get("priority") or 0), -idx, tier))
    if not scored:
        return None
    scored.sort(reverse=True)
    return scored[0][2]


def _lint(path: Path) -> tuple[bool, dict[str, Any]]:
    if not path.exists():
        return False, {"error": f"rules file not found: {path}"}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        return False, {"error": f"invalid json: {exc}"}
    if not isinstance(payload, dict):
        return False, {"error": "root must be object"}

    rules_raw = payload.get("rules")
    if not isinstance(rules_raw, list):
        return False, {"error": "rules must be list"}

    rules: list[dict[str, Any]] = []
    rule_names: list[str] = []
    problems: list[str] = []
    for idx, row in enumerate(rules_raw, start=1):
        if not isinstance(row, dict):
            problems.append(f"rule#{idx}: must be object")
            continue
        name = str(row.get("name") or "").strip()
        if not name:
            problems.append(f"rule#{idx}: missing name")
            continue
        tier = str(row.get("tier") or "").strip().lower()
        if tier and tier not in VALID_TIERS:
            problems.append(f"rule#{idx}:{name} unsupported tier={tier}")
        provider = str(row.get("provider") or "").strip().lower()
        if provider and provider not in VALID_PROVIDERS:
            problems.append(f"rule#{idx}:{name} unsupported provider={provider}")
        if row.get("match_any") is not None and not isinstance(row.get("match_any"), list):
            problems.append(f"rule#{idx}:{name} match_any must be list")
        rules.append(row)
        rule_names.append(name)

    for fn in sorted(LOW_REQUIRED):
        actual_tier = _resolve_primary_tier(fn, rules)
        if actual_tier != "low":
            problems.append(f"function {fn} expected tier=low, actual={actual_tier}")
    for fn in sorted(HIGH_REQUIRED):
        actual_tier = _resolve_primary_tier(fn, rules)
        if actual_tier != "high":
            problems.append(f"function {fn} expected tier=high, actual={actual_tier}")

    safety_rules = [
        row
        for row in rules
        if bool(row.get("enabled", True)) and str(row.get("tier") or "").strip().lower() == "safety"
    ]
    if not safety_rules:
        problems.append("missing safety tier rule")
    else:
        for fn in sorted(LOW_REQUIRED | HIGH_REQUIRED):
            if not any(_rule_matches_function(row, fn) for row in safety_rules):
                problems.append(f"function {fn} not covered by safety tier rule")
                break

    model_hints = {
        "low": "zai/glm-4-flash",
        "high": "minimax/MiniMax-M2.5",
        "safety": "deepseek/deepseek-chat",
    }
    for tier, expected_model in model_hints.items():
        tier_rows = [
            row for row in rules
            if bool(row.get("enabled", True)) and str(row.get("tier") or "").strip().lower() == tier
        ]
        if tier_rows and not any(str(row.get("model") or "").strip() == expected_model for row in tier_rows):
            problems.append(f"tier={tier} missing expected model hint: {expected_model}")

    return len(problems) == 0, {
        "path": str(path),
        "version": payload.get("version"),
        "rule_count": len(rules),
        "rule_names": sorted(rule_names),
        "required_low_count": len(LOW_REQUIRED),
        "required_high_count": len(HIGH_REQUIRED),
        "problems": problems,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="OpenClaw routing lint")
    parser.add_argument(
        "--input",
        default=".openclaw/pmflow-routing-rules.json",
        help="Rules file path",
    )
    args = parser.parse_args()

    ok, report = _lint(Path(args.input))
    print(json.dumps(report, ensure_ascii=False, indent=2))
    raise SystemExit(0 if ok else 2)


if __name__ == "__main__":
    main()