# -*- coding: utf-8 -*-
"""开发规范 hard policy 守护（索引：development/POLICY_MATRIX.md）。"""
from __future__ import annotations

import os
import re
import unittest

from _support import get_skill_root

SKIP_DIR_NAMES = frozenset(
    {
        ".git",
        "__pycache__",
        ".pytest_cache",
        "rpa-artifacts",
        "videos",
    }
)

SKIP_REL_PREFIXES = ("tests/artifacts", "tests/diagnostics")

POLICY_IDS = (
    "POLICY-STRUCTURE-001",
    "POLICY-RUNTIME-001",
    "POLICY-RUNTIME-002",
    "POLICY-INSTALL-001",
    "POLICY-CONFIG-001",
    "POLICY-CONFIG-002",
    "POLICY-CONFIG-003",
    "POLICY-TESTING-001",
    "POLICY-TESTING-002",
    "POLICY-RPA-001",
    "POLICY-RPA-002",
    "POLICY-RPA-003",
    "POLICY-PACKAGING-001",
    "POLICY-PACKAGING-002",
    "POLICY-DOCS-001",
    "POLICY-LOGGING-001",
    "POLICY-LOGGING-002",
    "POLICY-LOGGING-003",
    "POLICY-LOGGING-004",
    "POLICY-LOGGING-005",
    "POLICY-CONTROL-001",
    "POLICY-CONTROL-002",
    "POLICY-CONTROL-003",
    "POLICY-DATA-PATH-001",
    "POLICY-SKILL-ACTION-001",
)

STRUCTURE_PATHS = (
    "scripts/main.py",
    "scripts/cli",
    "scripts/service",
    "scripts/db",
    "scripts/util",
)

CONFIG_ENV_BOUNDARY_FILES = frozenset(
    {
        "scripts/main.py",
        "scripts/util/config_bootstrap.py",
        "scripts/util/runtime_paths.py",
        "scripts/util/logging_config.py",
        "scripts/util/constants.py",
    }
)

INSTALL_FORBIDDEN_PATTERNS = (
    re.compile(r"\bpip\s+install\b", re.IGNORECASE),
    re.compile(r"\bpython\s+-m\s+pip\s+install\b", re.IGNORECASE),
    re.compile(r"\buv\s+pip\s+install\b", re.IGNORECASE),
    re.compile(r"\bplaywright\s+install\b", re.IGNORECASE),
)

CONFIG_ENV_FORBIDDEN_PATTERNS = (
    re.compile(r"\bos\.environ\b"),
    re.compile(r"\bos\.getenv\b"),
    re.compile(r"\bos\.putenv\b"),
    re.compile(r"\benviron\.get\b"),
)

TESTING_REAL_TARGET_ALLOWED_ROOT_TESTS = frozenset({"test_adapter_profile_policy.py"})

TESTING_001_FORBIDDEN_SUBSTRINGS = (
    "real_" + "api",
    "real_" + "rpa",
    "OPENCLAW_" + "TEST_TARGET",
)

TESTING_001_SKIP_ROOT_TESTS = frozenset({"test_development_policy_guard.py"})

RPA_001_FORBIDDEN_PATTERNS = (
    re.compile(r"\brpa_helpers\b"),
    re.compile(r"\binject_account_manager_scripts_path\b"),
    re.compile(r"\bget_account_credential\b"),
)

RPA_002_FFMPEG_PATTERNS = (
    re.compile(
        r"\bsubprocess\.(run|call|Popen|check_output|check_call)\s*\([^)]*ffmpeg",
        re.IGNORECASE | re.DOTALL,
    ),
    re.compile(
        r"\bsubprocess\.(run|call|Popen|check_output|check_call)\(\s*['\"][^'\"]*ffmpeg",
        re.IGNORECASE,
    ),
    re.compile(r"\bos\.system\s*\([^)]*ffmpeg", re.IGNORECASE | re.DOTALL),
    re.compile(r"\bos\.popen\s*\([^)]*ffmpeg", re.IGNORECASE | re.DOTALL),
    re.compile(
        r"\bsubprocess\.(run|call|Popen|check_output|check_call)\([^)]*shell\s*=\s*True[^)]*ffmpeg",
        re.IGNORECASE | re.DOTALL,
    ),
)

RPA_002_EXEMPT_REL = "scripts/service/task_run_support.py"

RPA_003_SOURCE = (
    "development/RPA.md §5.3; development/TESTING.md §12; development/POLICY_MATRIX.md"
)

LOGGING_001_SOURCE = "development/LOGGING.md; development/RUNTIME.md"
LOGGING_002_SOURCE = "development/LOGGING.md; development/DEVELOPMENT.md"
LOGGING_003_SOURCE = "development/LOGGING.md"
LOGGING_004_SOURCE = "development/LOGGING.md; development/RPA.md"
LOGGING_005_SOURCE = "development/LOGGING.md; development/CONFIG.md"

LOGGING_CALL_PATTERN = re.compile(
    r"\b(log|logger|get_skill_logger\s*\(\s*\))\s*\.\s*(info|warning|error|debug|exception|critical)\s*\(",
    re.IGNORECASE,
)
LOGGING_CALL_GET_LOGGER_PATTERN = re.compile(
    r"get_skill_logger\s*\(\s*\)\s*\.\s*(info|warning|error|debug|exception|critical)\s*\(",
    re.IGNORECASE,
)
SENSITIVE_LOG_ASSIGNMENT_PATTERN = re.compile(
    r"(password|token|cookie|secret|api_key|authorization)\s*=",
    re.IGNORECASE,
)


def _policy_msg(policy_id: str, source: str, detail: str) -> str:
    return f"{policy_id} [{source}]: {detail}"


def _rel(skill_root: str, path: str) -> str:
    return os.path.relpath(path, skill_root).replace("\\", "/")


def _should_skip_dirname(dirname: str) -> bool:
    return dirname in SKIP_DIR_NAMES


def _should_skip_rel_path(skill_root: str, abs_path: str) -> bool:
    rel = _rel(skill_root, abs_path).replace("\\", "/")
    for prefix in SKIP_REL_PREFIXES:
        normalized = prefix.rstrip("/")
        if rel == normalized or rel.startswith(normalized + "/"):
            return True
    return False


def _walk_files(skill_root: str, rel_root: str, *, suffix: str | None = None) -> list[str]:
    base = os.path.join(skill_root, rel_root)
    if not os.path.exists(base):
        return []
    found: list[str] = []
    for root, dirnames, filenames in os.walk(base):
        kept: list[str] = []
        for d in dirnames:
            if _should_skip_dirname(d):
                continue
            sub = os.path.join(root, d)
            if _should_skip_rel_path(skill_root, sub):
                continue
            kept.append(d)
        dirnames[:] = kept
        for name in filenames:
            full = os.path.join(root, name)
            if _should_skip_rel_path(skill_root, full):
                continue
            if suffix and not name.endswith(suffix):
                continue
            found.append(full)
    return sorted(found)


def _read_text(path: str) -> str:
    with open(path, encoding="utf-8") as f:
        return f.read()


def _requirement_dependency_lines(req_text: str) -> list[str]:
    return [
        line.strip()
        for line in req_text.splitlines()
        if line.strip() and not line.strip().startswith("#")
    ]


def _scan_install_delivery_files(skill_root: str) -> list[str]:
    offenders: list[str] = []
    scan_specs: list[tuple[str, str | None]] = [
        ("scripts", ".py"),
        (".", ".ps1"),
        (".", ".sh"),
        (".", None),
        (".github/workflows", ".yml"),
        (".github/workflows", ".yaml"),
    ]
    seen: set[str] = set()
    for rel_root, suffix in scan_specs:
        if rel_root == "." and suffix is None:
            req = os.path.join(skill_root, "requirements.txt")
            if os.path.isfile(req):
                seen.add(req)
                text = _read_text(req)
                for pattern in INSTALL_FORBIDDEN_PATTERNS:
                    if pattern.search(text):
                        offenders.append(
                            _policy_msg(
                                "POLICY-INSTALL-001",
                                "development/RUNTIME.md; development/RPA.md §1.4; development/DEVELOPMENT.md §3.1",
                                f"{_rel(skill_root, req)} matches {pattern.pattern}",
                            )
                        )
            continue
        for path in _walk_files(skill_root, rel_root, suffix=suffix):
            if path in seen:
                continue
            seen.add(path)
            if suffix == ".ps1" and not path.endswith(".ps1"):
                continue
            if suffix == ".sh" and not path.endswith(".sh"):
                continue
            text = _read_text(path)
            for pattern in INSTALL_FORBIDDEN_PATTERNS:
                if pattern.search(text):
                    offenders.append(
                        _policy_msg(
                            "POLICY-INSTALL-001",
                            "development/RUNTIME.md; development/RPA.md §1.4; development/DEVELOPMENT.md §3.1",
                            f"{_rel(skill_root, path)} matches {pattern.pattern}",
                        )
                    )
    return offenders


def _gitignore_has_pattern(lines: list[str], pattern: str) -> bool:
    normalized = [ln.strip() for ln in lines if ln.strip() and not ln.strip().startswith("#")]
    return pattern in normalized


class TestPolicyStructure001(unittest.TestCase):
    def test_standard_scripts_layout_exists(self) -> None:
        skill_root = get_skill_root()
        missing: list[str] = []
        for rel in STRUCTURE_PATHS:
            path = os.path.join(skill_root, *rel.split("/"))
            if not os.path.exists(path):
                missing.append(rel)
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-STRUCTURE-001",
                "development/DEVELOPMENT.md §2; development/RUNTIME.md §目录结构",
                "missing: " + ", ".join(missing),
            ),
        )


class TestPolicyRuntime001(unittest.TestCase):
    def test_no_vendored_jiangchang_skill_core(self) -> None:
        skill_root = get_skill_root()
        vendored = os.path.join(skill_root, "scripts", "jiangchang_skill_core")
        self.assertFalse(
            os.path.isdir(vendored),
            msg=_policy_msg(
                "POLICY-RUNTIME-001",
                "development/RUNTIME.md §共享 Python Runtime",
                _rel(skill_root, vendored),
            ),
        )


class TestPolicyRuntime002(unittest.TestCase):
    def test_requirements_txt_excludes_shared_runtime_deps(self) -> None:
        skill_root = get_skill_root()
        req_path = os.path.join(skill_root, "requirements.txt")
        self.assertTrue(os.path.isfile(req_path), msg="requirements.txt missing")
        text = _read_text(req_path)
        dep_lines = _requirement_dependency_lines(text)
        offenders: list[str] = []
        for line in dep_lines:
            lower = line.lower()
            if "jiangchang-platform-kit" in lower:
                offenders.append(line)
            if re.search(r"(?<![\w.-])playwright(?![\w.-])", lower):
                offenders.append(line)
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-RUNTIME-002",
                "development/RUNTIME.md; development/DEVELOPMENT.md §3.1",
                "forbidden dependency lines: " + "; ".join(offenders),
            ),
        )


class TestPolicyInstall001(unittest.TestCase):
    def test_delivery_files_have_no_auto_install_commands(self) -> None:
        offenders = _scan_install_delivery_files(get_skill_root())
        self.assertEqual(offenders, [], msg="\n".join(offenders))


class TestPolicyConfig001(unittest.TestCase):
    def test_env_example_exists(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, ".env.example")
        self.assertTrue(
            os.path.isfile(path),
            msg=_policy_msg(
                "POLICY-CONFIG-001",
                "development/CONFIG.md",
                ".env.example missing at repo root",
            ),
        )


class TestPolicyConfig002(unittest.TestCase):
    def test_gitignore_ignores_env_files(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, ".gitignore")
        self.assertTrue(os.path.isfile(path), msg=".gitignore missing")
        lines = _read_text(path).splitlines()
        missing: list[str] = []
        if not _gitignore_has_pattern(lines, ".env"):
            missing.append(".env")
        if not _gitignore_has_pattern(lines, "*.env.local"):
            missing.append("*.env.local")
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-CONFIG-002",
                "development/CONFIG.md §红线：敏感信息不进 .env",
                "missing gitignore entries: " + ", ".join(missing),
            ),
        )


class TestPolicyConfig003(unittest.TestCase):
    def test_business_scripts_do_not_read_os_environ_directly(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts", suffix=".py"):
            rel = _rel(skill_root, path)
            if rel.replace("\\", "/") in CONFIG_ENV_BOUNDARY_FILES:
                continue
            text = _read_text(path)
            for lineno, line in enumerate(text.splitlines(), 1):
                if line.strip().startswith("#"):
                    continue
                for pattern in CONFIG_ENV_FORBIDDEN_PATTERNS:
                    if pattern.search(line):
                        offenders.append(
                            f"{rel}:{lineno}: {line.strip()} — use jiangchang_skill_core.config.get*"
                        )
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-CONFIG-003",
                "development/CONFIG.md; development/ADAPTER.md",
                "\n".join(offenders),
            ),
        )


class TestPolicyTesting001(unittest.TestCase):
    def test_root_tests_do_not_hardcode_real_targets(self) -> None:
        skill_root = get_skill_root()
        tests_dir = os.path.join(skill_root, "tests")
        offenders: list[str] = []
        for name in sorted(os.listdir(tests_dir)):
            if not (name.startswith("test_") and name.endswith(".py")):
                continue
            if name in TESTING_001_SKIP_ROOT_TESTS:
                continue
            if name in TESTING_REAL_TARGET_ALLOWED_ROOT_TESTS:
                continue
            path = os.path.join(tests_dir, name)
            text = _read_text(path)
            for lineno, line in enumerate(text.splitlines(), 1):
                stripped = line.strip()
                if stripped.startswith("#"):
                    continue
                for token in TESTING_001_FORBIDDEN_SUBSTRINGS:
                    if token in line:
                        offenders.append(f"{name}:{lineno}: {stripped}")
                        break
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-TESTING-001",
                "development/TESTING.md §默认必跑测试要做什么",
                "\n".join(offenders),
            ),
        )


class TestPolicyTesting002(unittest.TestCase):
    def test_integration_python_tests_are_sample_only(self) -> None:
        skill_root = get_skill_root()
        integration_dir = os.path.join(skill_root, "tests", "integration")
        if not os.path.isdir(integration_dir):
            return
        offenders: list[str] = []
        for name in sorted(os.listdir(integration_dir)):
            if not name.endswith(".py"):
                continue
            if name.endswith(".sample"):
                continue
            offenders.append(f"tests/integration/{name}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-TESTING-002",
                "development/TESTING.md §7 真实联调测试的安全约束",
                "non-sample integration tests: " + ", ".join(offenders),
            ),
        )


class TestPolicyRpa001(unittest.TestCase):
    def test_scripts_have_no_account_manager_internal_imports(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts", suffix=".py"):
            rel = _rel(skill_root, path)
            text = _read_text(path)
            for pattern in RPA_001_FORBIDDEN_PATTERNS:
                if pattern.search(text):
                    offenders.append(f"{rel}: {pattern.pattern}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-RPA-001",
                "development/ADAPTER.md §兄弟依赖; development/RPA.md §1.7",
                "\n".join(offenders),
            ),
        )


class TestPolicyRpa002(unittest.TestCase):
    def test_scripts_do_not_invoke_ffmpeg_directly(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts", suffix=".py"):
            rel = _rel(skill_root, path)
            if rel.replace("\\", "/") == RPA_002_EXEMPT_REL:
                continue
            text = _read_text(path)
            for pattern in RPA_002_FFMPEG_PATTERNS:
                if pattern.search(text):
                    offenders.append(f"{rel}: {pattern.pattern}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-RPA-002",
                "development/RPA.md §5.3 录屏成片标准",
                "\n".join(offenders),
            ),
        )


class TestPolicyRpa003(unittest.TestCase):
    def test_service_layer_retains_video_session_integration(self) -> None:
        skill_root = get_skill_root()
        service_files = _walk_files(skill_root, "scripts/service", suffix=".py")
        self.assertTrue(service_files, msg="scripts/service/*.py not found")
        combined = "\n".join(_read_text(path) for path in service_files)
        missing: list[str] = []
        if "RpaVideoSession" not in combined:
            missing.append("RpaVideoSession")
        if "video.add_step" not in combined and ".add_step(" not in combined:
            missing.append("video.add_step or .add_step(")
        if "merge_video_into_result_summary" not in combined:
            missing.append("merge_video_into_result_summary")
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-RPA-003",
                RPA_003_SOURCE,
                "scripts/service/*.py missing: " + ", ".join(missing),
            ),
        )


def _scan_sensitive_logging_assignments(skill_root: str) -> list[str]:
    offenders: list[str] = []
    for path in _walk_files(skill_root, "scripts", suffix=".py"):
        rel = _rel(skill_root, path)
        for lineno, line in enumerate(_read_text(path).splitlines(), 1):
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue
            if not LOGGING_CALL_PATTERN.search(line):
                continue
            if SENSITIVE_LOG_ASSIGNMENT_PATTERN.search(line):
                offenders.append(f"{rel}:{lineno}: {stripped}")
    return offenders


class TestPolicyLogging001(unittest.TestCase):
    def test_cli_app_sets_up_unified_logging(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "cli", "app.py")
        text = _read_text(path)
        rel = _rel(skill_root, path)
        self.assertIn(
            "setup_skill_logging",
            text,
            msg=_policy_msg(
                "POLICY-LOGGING-001",
                LOGGING_001_SOURCE,
                f"{rel} missing setup_skill_logging",
            ),
        )
        has_logger_info = bool(
            LOGGING_CALL_GET_LOGGER_PATTERN.search(text)
            or re.search(r"\bget_skill_logger\s*\(\s*\)\s*\.\s*info\s*\(", text)
        )
        self.assertTrue(
            has_logger_info,
            msg=_policy_msg(
                "POLICY-LOGGING-001",
                LOGGING_001_SOURCE,
                f"{rel} missing get_skill_logger().info (or equivalent)",
            ),
        )


class TestPolicyLogging002(unittest.TestCase):
    def test_task_service_uses_get_skill_logger(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "service", "task_service.py")
        text = _read_text(path)
        rel = _rel(skill_root, path)
        self.assertRegex(
            text,
            r"get_skill_logger",
            msg=_policy_msg(
                "POLICY-LOGGING-002",
                LOGGING_002_SOURCE,
                f"{rel} must import/use get_skill_logger",
            ),
        )


class TestPolicyLogging003(unittest.TestCase):
    def test_task_service_logs_start_failure_and_task_logs(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "service", "task_service.py")
        text = _read_text(path)
        rel = _rel(skill_root, path)
        missing: list[str] = []
        if not re.search(r"\b(log|logger)\.info\s*\(", text):
            missing.append("log.info or logger.info")
        if not re.search(r"\b(log|logger)\.exception\s*\(", text):
            missing.append("log.exception or logger.exception")
        if "save_task_log" not in text:
            missing.append("save_task_log")
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-LOGGING-003",
                LOGGING_003_SOURCE,
                f"{rel} missing: " + ", ".join(missing),
            ),
        )


class TestPolicyLogging004(unittest.TestCase):
    def test_task_service_emits_progress_for_long_tasks(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "service", "task_service.py")
        text = _read_text(path)
        rel = _rel(skill_root, path)
        has_progress = bool(
            re.search(r"\bemit\s*\(", text)
            or re.search(r"\bactivity\.emit\s*\(", text)
            or re.search(r"\bvideo\.add_step\s*\(", text)
        )
        self.assertTrue(
            has_progress,
            msg=_policy_msg(
                "POLICY-LOGGING-004",
                LOGGING_004_SOURCE,
                f"{rel} must use emit( / activity.emit / video.add_step",
            ),
        )


class TestPolicyLogging005(unittest.TestCase):
    def test_scripts_do_not_log_sensitive_assignments(self) -> None:
        offenders = _scan_sensitive_logging_assignments(get_skill_root())
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-LOGGING-005",
                LOGGING_005_SOURCE,
                "\n".join(offenders),
            ),
        )


class TestPolicyControl001(unittest.TestCase):
    def test_no_util_progress_module(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "util", "progress.py")
        self.assertFalse(
            os.path.isfile(path),
            msg=_policy_msg(
                "POLICY-CONTROL-001",
                "development/LOGGING.md §2.5; development/RPA.md §0.1",
                "scripts/util/progress.py must not exist",
            ),
        )


class TestPolicyControl002(unittest.TestCase):
    def test_task_service_uses_job_context_and_finish(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "service", "task_service.py")
        text = _read_text(path)
        rel = _rel(skill_root, path)
        missing: list[str] = []
        if "job_context(" not in text:
            missing.append("job_context(")
        if "finish(" not in text:
            missing.append("finish(")
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-CONTROL-002",
                "development/LOGGING.md §2.5、§7",
                f"{rel} missing: " + ", ".join(missing),
            ),
        )


class TestPolicyControl003(unittest.TestCase):
    def test_service_layer_has_no_bare_asyncio_sleep(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts/service", suffix=".py"):
            rel = _rel(skill_root, path)
            for lineno, line in enumerate(_read_text(path).splitlines(), 1):
                stripped = line.strip()
                if not stripped or stripped.startswith("#"):
                    continue
                if "asyncio.sleep" in stripped:
                    offenders.append(f"{rel}:{lineno}: {stripped}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-CONTROL-003",
                "development/RPA.md §0.1",
                "use interruptible_sleep instead:\n" + "\n".join(offenders),
            ),
        )


class TestPolicyDataPath001(unittest.TestCase):
    def test_env_example_has_no_cwd_relative_paths(self) -> None:
        path = os.path.join(get_skill_root(), ".env.example")
        offenders: list[str] = []
        with open(path, encoding="utf-8") as f:
            for lineno, line in enumerate(f, 1):
                stripped = line.strip()
                if not stripped or stripped.startswith("#"):
                    continue
                active = stripped.split("#", 1)[0].strip()
                if re.search(r"=\s*\./", active):
                    offenders.append(f".env.example:{lineno}: {active}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-DATA-PATH-001",
                "development/DATA_PATHS.md",
                "\n".join(offenders) or "CWD-relative ./ paths in .env.example",
            ),
        )

    def test_runtime_paths_exposes_resolve_data_path(self) -> None:
        import util.runtime_paths as rp

        self.assertTrue(callable(getattr(rp, "resolve_data_path", None)))
        self.assertTrue(callable(getattr(rp, "list_resolved_data_paths", None)))

    def test_business_scripts_do_not_abspath_config_get(self) -> None:
        skill_root = get_skill_root()
        pattern = re.compile(r"os\.path\.abspath\s*\(\s*config\.get\b")
        offenders: list[str] = []
        for rel in _walk_files(skill_root, "scripts", suffix=".py"):
            if rel.replace("\\", "/") == "scripts/util/runtime_paths.py":
                continue
            text = _read_text(os.path.join(skill_root, rel))
            for lineno, line in enumerate(text.splitlines(), 1):
                if line.strip().startswith("#"):
                    continue
                if pattern.search(line):
                    offenders.append(f"{rel}:{lineno}: {line.strip()}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-DATA-PATH-001",
                "development/DATA_PATHS.md",
                "\n".join(offenders),
            ),
        )


_TEMPLATE_PLACEHOLDER_SLUG = "your-skill-slug"
_RPA_SESSION_TOKENS = ("RpaVideoSession",)


def _service_has_rpa_video_session(skill_root: str) -> bool:
    service_dir = os.path.join(skill_root, "scripts", "service")
    if not os.path.isdir(service_dir):
        return False
    for rel in _walk_files(skill_root, "scripts/service", suffix=".py"):
        text = _read_text(os.path.join(skill_root, rel))
        if any(token in text for token in _RPA_SESSION_TOKENS):
            return True
    return False


def _load_actions_json(skill_root: str) -> dict | None:
    path = os.path.join(skill_root, "assets", "actions.json")
    if not os.path.isfile(path):
        return None
    import json

    with open(path, encoding="utf-8") as f:
        return json.load(f)


class TestPolicySkillAction001(unittest.TestCase):
    def test_skill_action_runtime_doc_exists(self) -> None:
        path = os.path.join(get_skill_root(), "development", "SKILL_ACTION_RUNTIME.md")
        self.assertTrue(
            os.path.isfile(path),
            msg=_policy_msg(
                "POLICY-SKILL-ACTION-001",
                "development/SKILL_ACTION_RUNTIME.md",
                "file missing",
            ),
        )
        text = _read_text(path)
        for marker in ("executionProfile", "async", "任务中心", "run_skill_action"):
            self.assertIn(
                marker,
                text,
                msg=_policy_msg(
                    "POLICY-SKILL-ACTION-001",
                    "development/SKILL_ACTION_RUNTIME.md",
                    f"missing marker {marker!r}",
                ),
            )

    def test_schema_defines_execution_profile(self) -> None:
        import json

        path = os.path.join(get_skill_root(), "assets", "schemas", "skill-actions.schema.json")
        self.assertTrue(os.path.isfile(path))
        with open(path, encoding="utf-8") as f:
            schema = json.load(f)
        action_props = schema.get("$defs", {}).get("action", {}).get("properties", {})
        profile = action_props.get("executionProfile")
        self.assertIsInstance(
            profile,
            dict,
            msg=_policy_msg(
                "POLICY-SKILL-ACTION-001",
                "assets/schemas/skill-actions.schema.json",
                "action.properties.executionProfile missing",
            ),
        )
        self.assertEqual(set(profile.get("enum") or []), {"sync", "async"})

    def test_actions_md_documents_execution_profile(self) -> None:
        path = os.path.join(get_skill_root(), "references", "ACTIONS.md")
        text = _read_text(path)
        for marker in ("executionProfile", "async", "任务中心", "SKILL_ACTION_RUNTIME"):
            self.assertIn(
                marker,
                text,
                msg=_policy_msg(
                    "POLICY-SKILL-ACTION-001",
                    "references/ACTIONS.md",
                    f"missing marker {marker!r}",
                ),
            )

    def test_actions_json_actions_declare_execution_profile(self) -> None:
        skill_root = get_skill_root()
        manifest = _load_actions_json(skill_root)
        if manifest is None:
            return
        offenders: list[str] = []
        for action in manifest.get("actions") or []:
            action_id = action.get("id", "?")
            profile = action.get("executionProfile")
            if profile not in ("sync", "async"):
                offenders.append(f"{action_id}: executionProfile={profile!r}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-SKILL-ACTION-001",
                "assets/actions.json",
                "every action must set executionProfile to sync|async:\n" + "\n".join(offenders),
            ),
        )

    def test_business_rpa_skill_has_async_agent_action(self) -> None:
        from util.constants import SKILL_SLUG

        skill_root = get_skill_root()
        if SKILL_SLUG == _TEMPLATE_PLACEHOLDER_SLUG:
            self.skipTest("template placeholder slug skips business RPA action check")
        if not _service_has_rpa_video_session(skill_root):
            self.skipTest("no RpaVideoSession in scripts/service")
        manifest = _load_actions_json(skill_root)
        self.assertIsNotNone(
            manifest,
            msg=_policy_msg(
                "POLICY-SKILL-ACTION-001",
                "assets/actions.json",
                "RPA skill must provide assets/actions.json",
            ),
        )
        found = False
        for action in manifest.get("actions") or []:
            placements = set(action.get("placements") or [])
            if action.get("executionProfile") == "async" and "agent" in placements:
                found = True
                break
        self.assertTrue(
            found,
            msg=_policy_msg(
                "POLICY-SKILL-ACTION-001",
                "assets/actions.json",
                "RPA business skill needs at least one action with executionProfile=async and placements containing agent",
            ),
        )


class TestPolicyDocs001(unittest.TestCase):
    def test_policy_matrix_exists_and_lists_policy_ids(self) -> None:
        skill_root = get_skill_root()
        matrix_path = os.path.join(skill_root, "development", "POLICY_MATRIX.md")
        self.assertTrue(
            os.path.isfile(matrix_path),
            msg=_policy_msg(
                "POLICY-DOCS-001",
                "development/POLICY_MATRIX.md",
                "file missing",
            ),
        )
        text = _read_text(matrix_path)
        missing = [pid for pid in POLICY_IDS if pid not in text]
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-DOCS-001",
                "development/POLICY_MATRIX.md",
                "missing policy_id entries: " + ", ".join(missing),
            ),
        )


if __name__ == "__main__":
    unittest.main()