PMFlow/scripts/semantic_lint.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import sys
from dataclasses import dataclass
from pathlib import Path
import re
from typing import Iterable, Sequence


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SCAN_ROOTS = (
    Path("backend/app/services/ai_pmo"),
    Path("backend/app/services/integrations"),
    Path("backend/app/services/signal"),
    Path("backend/app/api/ai_pmo_api"),
)
ALLOWLIST_PREFIXES = (
    Path("backend/app/config/ai_prompt_registry.py"),
    Path("backend/app/config/ai_prompts.py"),
    Path("backend/app/config/ai_prompts_catalog"),
    Path("backend/app/config/operation_catalog_core"),
    Path("backend/app/services/ai_pmo/ai_pmo_kernel_world_model_core/mixin_project.py"),
    Path("backend/app/services/signal/signal_source_orchestration/mixin_planning_locator_roles.py"),
    Path("backend/app/services/signal/signal_source_orchestration/mixin_planning_onboard.py"),
    Path("backend/app/services/signal/signal_capture_service/mixin_analysis_normalize.py"),
    Path("backend/app/services/signal/symbol_index/shared.py"),
    Path("backend/app/services/integrations/feishu_integration_service/mixin_message_payload.py"),
)
DEFAULT_BASELINE = Path("scripts/semantic_lint_baseline.txt")


@dataclass(frozen=True)
class Problem:
    path: str
    line: int
    rule_id: str
    excerpt: str

    def key(self) -> str:
        return f"{self.path}:{self.line}:{self.rule_id}"


@dataclass(frozen=True)
class Rule:
    rule_id: str
    pattern: re.Pattern[str]
    message: str


RULES: tuple[Rule, ...] = (
    Rule(
        rule_id="semantic_hint_constant",
        pattern=re.compile(r"\b[A-Z][A-Z0-9_]*(?:HINTS|KEYWORDS)\b\s*="),
        message="semantic hint/keyword table should live in workflow_semantic_planner or prompt/catalog config",
    ),
    Rule(
        rule_id="semantic_regex_constant",
        pattern=re.compile(
            r"\b[A-Z][A-Z0-9_]*(?:PROJECT|TASK|SIGNAL|FOLLOWUP|COMMAND|QUERY|SOURCE|INTENT|EXTERNAL|CORRECTION)[A-Z0-9_]*_RE\b\s*=\s*re\.compile"
        ),
        message="business routing regex should live in workflow_semantic_planner",
    ),
    Rule(
        rule_id="looks_like_helper",
        pattern=re.compile(
            r"def\s+_?looks_like_(?:project|task|signal|followup|query|command|external|contextual|write|source|intent)[A-Za-z0-9_]*\s*\("
        ),
        message="behavior-routing looks_like helper should live in workflow_semantic_planner",
    ),
    Rule(
        rule_id="question_routing_helper",
        pattern=re.compile(r"def\s+_is_(?:query_like_question|obvious_question)\s*\("),
        message="question routing helper should live in workflow_semantic_planner",
    ),
)


def _normalize_relpath(root: Path, path: Path) -> str:
    return path.resolve().relative_to(root.resolve()).as_posix()


def _is_allowlisted(relpath: str) -> bool:
    relative = Path(relpath)
    for prefix in ALLOWLIST_PREFIXES:
        if relative == prefix or prefix in relative.parents:
            return True
    return False


def _iter_files(root: Path, scan_roots: Sequence[Path]) -> Iterable[Path]:
    for rel_root in scan_roots:
        candidate = (root / rel_root).resolve()
        if not candidate.exists():
            continue
        if candidate.is_file():
            yield candidate
            continue
        for path in candidate.rglob("*"):
            if not path.is_file():
                continue
            if path.suffix not in {".py", ".ts", ".tsx"}:
                continue
            yield path


def _load_baseline(root: Path, baseline_path: Path) -> set[str]:
    candidate = (root / baseline_path).resolve()
    if not candidate.exists():
        return set()
    rows: set[str] = set()
    for line in candidate.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        rows.add(stripped)
    return rows


def collect_problems(
    *,
    root: Path,
    scan_roots: Sequence[Path],
    baseline_path: Path,
) -> list[Problem]:
    baseline = _load_baseline(root, baseline_path)
    problems: list[Problem] = []
    for path in _iter_files(root, scan_roots):
        relpath = _normalize_relpath(root, path)
        if _is_allowlisted(relpath):
            continue
        lines = path.read_text(encoding="utf-8").splitlines()
        for idx, raw_line in enumerate(lines, start=1):
            line = raw_line.rstrip("\n")
            if "semantic-lint: ignore" in line:
                continue
            for rule in RULES:
                if not rule.pattern.search(line):
                    continue
                problem = Problem(
                    path=relpath,
                    line=idx,
                    rule_id=rule.rule_id,
                    excerpt=line.strip()[:240],
                )
                if problem.key() in baseline:
                    continue
                problems.append(problem)
                break
    return problems


def _parse_args(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Guard PMFlow business semantics from leaking outside workflow_semantic_planner.")
    parser.add_argument(
        "--root",
        default=str(ROOT),
        help="Repository root to scan.",
    )
    parser.add_argument(
        "--baseline",
        default=str(DEFAULT_BASELINE),
        help="Baseline file with approved compatibility exceptions.",
    )
    parser.add_argument(
        "scan_roots",
        nargs="*",
        help="Optional relative roots/files to scan. Defaults to backend/app and frontend/src.",
    )
    return parser.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = _parse_args(argv or sys.argv[1:])
    root = Path(args.root).resolve()
    scan_roots = tuple(Path(item) for item in (args.scan_roots or DEFAULT_SCAN_ROOTS))
    problems = collect_problems(
        root=root,
        scan_roots=scan_roots,
        baseline_path=Path(args.baseline),
    )
    if not problems:
        print("semantic-lint: ok")
        return 0

    print("semantic-lint: problems found")
    for item in problems:
        print(f"- {item.path}:{item.line} [{item.rule_id}] {item.excerpt}")
    return 1


if __name__ == "__main__":
    raise SystemExit(main())