Release v1.0.60: harden business logging policies (LOGGING-006/007/008)

tests/test_development_policy_guard.py

@@ -46,6 +46,9 @@ POLICY_IDS = (
     "POLICY-LOGGING-003",
     "POLICY-LOGGING-004",
     "POLICY-LOGGING-005",
+    "POLICY-LOGGING-006",
+    "POLICY-LOGGING-007",
+    "POLICY-LOGGING-008",
     "POLICY-CONTROL-001",
     "POLICY-CONTROL-002",
     "POLICY-CONTROL-003",
@@ -201,6 +204,26 @@ 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_006_SOURCE = "development/LOGGING.md §0、§3"
+LOGGING_007_SOURCE = "development/LOGGING.md §3.1–§3.2"
+LOGGING_008_SOURCE = "development/LOGGING.md §0 铁律 3"
+
+LOGGING_006_EXEMPT_BASENAMES = frozenset(
+    {
+        "__init__.py",
+        "task_run_support.py",
+        "entitlement_service.py",
+        "sibling_bridge.py",
+    }
+)
+
+LOGGING_RPA_MARKERS = (
+    "RpaVideoSession",
+    "uiautomation",
+    "playwright",
+    "@rpa_step",
+    "rpa_step(",
+)
 
 LOGGING_CALL_PATTERN = re.compile(
     r"\b(log|logger|get_skill_logger\s*\(\s*\))\s*\.\s*(info|warning|error|debug|exception|critical)\s*\(",
@@ -911,6 +934,121 @@ class TestPolicyLogging005(unittest.TestCase):
         )
 
 
+def _iter_service_py_files(skill_root: str) -> list[str]:
+    service_dir = os.path.join(skill_root, "scripts", "service")
+    if not os.path.isdir(service_dir):
+        return []
+    return list(_walk_files(skill_root, os.path.join("scripts", "service"), suffix=".py"))
+
+
+def _service_layer_combined_text(skill_root: str) -> str:
+    parts: list[str] = []
+    for path in _iter_service_py_files(skill_root):
+        parts.append(_read_text(path))
+    return "\n".join(parts)
+
+
+_LOGGING_DOMAIN_CALL_RE = re.compile(
+    r"\b(log|logger|_logger)\s*\.\s*(info|warning|error|exception)\s*\(",
+    re.IGNORECASE,
+)
+_PRINT_ERROR_RE = re.compile(r"\bprint\s*\(.*ERROR:", re.IGNORECASE | re.DOTALL)
+_FUNC_DEF_RE = re.compile(r"(?m)^(async\s+)?def\s+\w+\s*\(")
+
+
+def _scan_print_error_without_logger(skill_root: str, path: str, text: str) -> list[str]:
+    """同一函数体内 print(...ERROR:) 须伴随 warning/error/exception 日志。"""
+    offenders: list[str] = []
+    matches = list(_FUNC_DEF_RE.finditer(text))
+    if not matches:
+        return offenders
+    for i, m in enumerate(matches):
+        start = m.start()
+        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
+        body = text[start:end]
+        if not _PRINT_ERROR_RE.search(body):
+            continue
+        if re.search(
+            r"\b(log|logger|_logger)\s*\.\s*(warning|error|exception)\s*\(",
+            body,
+            re.IGNORECASE,
+        ):
+            continue
+        lineno = text[:start].count("\n") + 1
+        offenders.append(
+            f"{_rel(skill_root, path)}:{lineno}: print(ERROR:) without log.warning|error|exception in same function"
+        )
+    return offenders
+
+
+class TestPolicyLogging006(unittest.TestCase):
+    def test_service_domain_modules_have_file_logging(self) -> None:
+        skill_root = get_skill_root()
+        offenders: list[str] = []
+        for path in _iter_service_py_files(skill_root):
+            base = os.path.basename(path)
+            if base in LOGGING_006_EXEMPT_BASENAMES:
+                continue
+            text = _read_text(path)
+            if not _LOGGING_DOMAIN_CALL_RE.search(text):
+                offenders.append(_rel(skill_root, path))
+        self.assertEqual(
+            offenders,
+            [],
+            msg=_policy_msg(
+                "POLICY-LOGGING-006",
+                LOGGING_006_SOURCE,
+                "service domain modules missing log/logger/_logger.(info|warning|error|exception):\n"
+                + "\n".join(offenders),
+            ),
+        )
+
+
+class TestPolicyLogging007(unittest.TestCase):
+    def test_rpa_service_layer_has_event_tokens(self) -> None:
+        skill_root = get_skill_root()
+        combined = _service_layer_combined_text(skill_root)
+        if not any(marker in combined for marker in LOGGING_RPA_MARKERS):
+            self.skipTest("no RPA markers in scripts/service; LOGGING-007 N/A")
+        missing: list[str] = []
+        if "task_start" not in combined:
+            missing.append("task_start")
+        if "task_failed" not in combined and "rpa_failed" not in combined:
+            missing.append("task_failed|rpa_failed")
+        if "stage=" not in combined:
+            missing.append("stage=")
+        # entitlement tokens strongly recommended; require at least one
+        if "entitlement_ok" not in combined and "entitlement_failed" not in combined:
+            missing.append("entitlement_ok|entitlement_failed")
+        self.assertEqual(
+            missing,
+            [],
+            msg=_policy_msg(
+                "POLICY-LOGGING-007",
+                LOGGING_007_SOURCE,
+                "RPA service layer missing event tokens: " + ", ".join(missing),
+            ),
+        )
+
+
+class TestPolicyLogging008(unittest.TestCase):
+    def test_print_error_paired_with_logger(self) -> None:
+        skill_root = get_skill_root()
+        offenders: list[str] = []
+        for path in _iter_service_py_files(skill_root):
+            text = _read_text(path)
+            offenders.extend(_scan_print_error_without_logger(skill_root, path, text))
+        self.assertEqual(
+            offenders,
+            [],
+            msg=_policy_msg(
+                "POLICY-LOGGING-008",
+                LOGGING_008_SOURCE,
+                "\n".join(offenders) or "print(ERROR:) must share function with log.warning|error|exception",
+            ),
+        )
+
+
 class TestPolicyControl001(unittest.TestCase):
     def test_no_util_progress_module(self) -> None:
         skill_root = get_skill_root()