feat: implement Operational Policy Compliance check (APPENG-5623)

abevalflow/certification.py

@@ -88,7 +88,7 @@ class CheckId(StrEnum):
     CheckId.FUNCTIONAL_VALIDATION,
     CheckId.INSTRUCTION_QUALITY,
     # CheckId.REGISTRY_GOVERNANCE,  # Not yet implemented
-    # CheckId.OPERATIONAL_POLICY_COMPLIANCE,  # Not yet implemented
+    CheckId.OPERATIONAL_POLICY_COMPLIANCE,
 ]
 
 CERTIFIED_CHECKS = [
@@ -510,6 +510,7 @@ def compute_certification(
     metadata_valid: bool = True,
     has_eval_assets: bool = True,
     policy: CertificationPolicy | None = None,
+    operational_policy_result: CheckResult | None = None,
 ) -> CertificationResult:
     """Compute certification levels from gate results.
 
@@ -581,16 +582,19 @@ def compute_certification(
         ),
     )
 
-    all_checks.setdefault(
-        CheckId.OPERATIONAL_POLICY_COMPLIANCE,
-        CheckResult(
-            check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE,
-            name="Operational Policy Compliance",
-            passed=False,
-            score=0.0,
-            message="Operational policy check not implemented",
-        ),
-    )
+    if operational_policy_result is not None:
+        all_checks[CheckId.OPERATIONAL_POLICY_COMPLIANCE] = operational_policy_result
+    else:
+        all_checks.setdefault(
+            CheckId.OPERATIONAL_POLICY_COMPLIANCE,
+            CheckResult(
+                check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE,
+                name="Operational Policy Compliance",
+                passed=False,
+                score=0.0,
+                message="Operational policy check not implemented",
+            ),
+        )
 
     all_checks.setdefault(
         CheckId.ENTERPRISE_BEHAVIORAL_TESTING,

abevalflow/operational_policy.py

@@ -0,0 +1,177 @@
+"""Operational policy compliance checks for skill submissions.
+
+Deterministic checks that validate submissions meet operational standards:
+resource limits, timeout compliance, error handling patterns, and logging
+suppression. No LLM calls required.
+
+These checks complement (not duplicate) Foundational-level schema validation.
+Schema validation (Pydantic) enforces field types and ``gt=0`` constraints;
+these checks enforce *policy limits* — e.g., ``cpus <= MAX_CPUS``.
+"""
+
+from __future__ import annotations
+
+import ast
+import logging
+import re
+from pathlib import Path
+
+import yaml
+from pydantic import ValidationError
+
+from abevalflow.certification import CheckId, CheckResult
+from abevalflow.schemas import OperationalLimits, SubmissionMetadata
+
+logger = logging.getLogger(__name__)
+
+LOGGING_SUPPRESSION_PATTERNS = [
+    re.compile(r"\bdo\s+not\s+log\b", re.IGNORECASE),
+    re.compile(r"\bdisable\s+logging\b", re.IGNORECASE),
+    re.compile(r"\bsuppress\s+output\b", re.IGNORECASE),
+    re.compile(r"\bno\s+logging\b", re.IGNORECASE),
+    re.compile(r"\bturn\s+off\s+log", re.IGNORECASE),
+]
+
+SECURITY_QUALIFIERS = re.compile(
+    r"\b(passwords?|credentials?|secrets?|pii|tokens?|sensitive|personal)\b",
+    re.IGNORECASE,
+)
+
+
+EXPECTED_RESOURCE_FIELDS = ["cpus", "memory_mb", "agent_timeout_sec"]
+
+
+def _check_resource_declaration(raw_data: dict) -> list[str]:
+    """Check that resource fields are explicitly declared, not just using defaults."""
+    missing = [f for f in EXPECTED_RESOURCE_FIELDS if f not in raw_data]
+    if missing:
+        return [f"Resource fields not explicitly declared: {', '.join(missing)}"]
+    return []
+
+
+def _check_resource_limits(metadata: SubmissionMetadata, limits: OperationalLimits) -> list[str]:
+    """Check that resource requests are within policy limits."""
+    issues = []
+    if metadata.cpus > limits.max_cpus:
+        issues.append(f"CPU request ({metadata.cpus}) exceeds max allowed ({limits.max_cpus})")
+    if metadata.memory_mb > limits.max_memory_mb:
+        issues.append(f"Memory request ({metadata.memory_mb}MB) exceeds max allowed ({limits.max_memory_mb}MB)")
+    return issues
+
+
+def _check_timeout_compliance(metadata: SubmissionMetadata, limits: OperationalLimits) -> list[str]:
+    """Check that timeouts are within reasonable bounds."""
+    issues = []
+    if metadata.agent_timeout_sec > limits.max_agent_timeout_sec:
+        issues.append(
+            f"Agent timeout ({metadata.agent_timeout_sec}s) exceeds max allowed ({limits.max_agent_timeout_sec}s)"
+        )
+    return issues
+
+
+def _check_error_handling(test_file: Path) -> list[str]:
+    """Check for bare except patterns that hide failures."""
+    issues = []
+    if not test_file.exists():
+        return issues
+
+    try:
+        tree = ast.parse(test_file.read_text())
+    except SyntaxError:
+        return issues
+
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.ExceptHandler):
+            continue
+        handler_body = node.body
+        if len(handler_body) != 1:
+            continue
+        stmt = handler_body[0]
+        is_pass = isinstance(stmt, ast.Pass)
+        is_ellipsis = isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and stmt.value.value is ...
+        if not (is_pass or is_ellipsis):
+            continue
+        if node.type is None:
+            issues.append(f"Bare 'except: pass' at line {node.lineno} — hides all errors")
+        elif isinstance(node.type, ast.Name) and node.type.id == "Exception":
+            issues.append(f"'except Exception: pass' at line {node.lineno} — hides all errors")
+
+    return issues
+
+
+def _check_logging_suppression(skill_path: Path) -> list[str]:
+    """Check SKILL.md for patterns that suppress agent logging.
+
+    Skips lines that contain security qualifiers (e.g., "Do not log passwords")
+    since those are legitimate security instructions, not logging suppression.
+    """
+    issues = []
+    if not skill_path.exists():
+        return issues
+
+    content = skill_path.read_text()
+    for line in content.splitlines():
+        for pattern in LOGGING_SUPPRESSION_PATTERNS:
+            match = pattern.search(line)
+            if match and not SECURITY_QUALIFIERS.search(line):
+                issues.append(f"Logging suppression pattern found: '{match.group()}'")
+
+    return issues
+
+
+def check_operational_policy(
+    submission_dir: Path,
+    limits: OperationalLimits | None = None,
+) -> CheckResult:
+    """Run all operational policy checks on a submission.
+
+    Args:
+        submission_dir: Path to the submission directory containing
+            metadata.yaml, skills/SKILL.md, and tests/test_outputs.py.
+        limits: Optional configurable limits from CertificationPolicy.
+            Uses OperationalLimits defaults if not provided.
+
+    Returns:
+        CheckResult with aggregated pass/fail, score, and message.
+    """
+    if limits is None:
+        limits = OperationalLimits()
+
+    all_issues: list[str] = []
+
+    metadata_path = submission_dir / "metadata.yaml"
+    if metadata_path.exists():
+        try:
+            data = yaml.safe_load(metadata_path.read_text()) or {}
+            all_issues.extend(_check_resource_declaration(data))
+            metadata = SubmissionMetadata(**data)
+            all_issues.extend(_check_resource_limits(metadata, limits))
+            all_issues.extend(_check_timeout_compliance(metadata, limits))
+        except (ValidationError, yaml.YAMLError) as e:
+            logger.warning("Failed to parse metadata.yaml for policy check: %s", e)
+    else:
+        logger.info("No metadata.yaml found, skipping resource/timeout checks")
+
+    test_file = submission_dir / "tests" / "test_outputs.py"
+    all_issues.extend(_check_error_handling(test_file))
+
+    skills_dir = submission_dir / "skills"
+    if skills_dir.exists():
+        for skill_file in skills_dir.glob("*.md"):
+            all_issues.extend(_check_logging_suppression(skill_file))
+
+    passed = len(all_issues) == 0
+    score = 1.0 if passed else max(0.0, 1.0 - (len(all_issues) * 0.25))
+
+    if passed:
+        message = "All operational policy checks passed"
+    else:
+        message = f"Operational policy issues ({len(all_issues)}): " + "; ".join(all_issues)
+
+    return CheckResult(
+        check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE,
+        name="Operational Policy Compliance",
+        passed=passed,
+        score=score,
+        message=message,
+    )

abevalflow/schemas.py

@@ -233,6 +233,18 @@ class ExperimentType(StrEnum):
     CUSTOM = "custom"
 
 
+class OperationalLimits(BaseModel):
+    """Configurable limits for operational policy compliance checks.
+
+    Defaults for operational policy checks. Can be overridden via
+    certification_policy.operational_limits in metadata.yaml.
+    """
+
+    max_cpus: int = Field(default=4, gt=0, description="Maximum allowed CPU cores")
+    max_memory_mb: int = Field(default=8192, gt=0, description="Maximum allowed memory in MB")
+    max_agent_timeout_sec: float = Field(default=3600.0, gt=0, description="Maximum allowed agent timeout in seconds")
+
+
 class CertificationLevelPolicy(BaseModel):
     """Policy configuration for a single certification level.
 
@@ -291,6 +303,9 @@ class CertificationPolicy(BaseModel):
 
     Example in metadata.yaml:
         certification_policy:
+          operational_limits:
+            max_cpus: 8
+            max_memory_mb: 16384
           foundational:
             checks:
               - valid_skill_structure
@@ -320,6 +335,14 @@ class CertificationPolicy(BaseModel):
         default=None,
         description="Policy for Certified certification level",
     )
+    operational_limits: OperationalLimits | None = Field(
+        default=None,
+        description=(
+            "Operational policy limits for resource, timeout, and other checks. "
+            "Used by the Trusted-level operational_policy_compliance check. "
+            "Overrides OperationalLimits defaults."
+        ),
+    )
 
     def get_checks_for_level(self, level: str) -> list[str] | None:
         """Get custom check list for a level, or None to use defaults."""
@@ -343,6 +366,10 @@ def get_threshold(self, check_id: str) -> float | None:
                     result = level_policy.thresholds[check_id]
         return result
 
+    def get_operational_limits(self) -> OperationalLimits | None:
+        """Get operational limits for the operational policy compliance check."""
+        return self.operational_limits
+
 
 class CopySpec(BaseModel):
     """A source directory and its destination path inside the container."""

config/certification_profiles.yaml

@@ -25,6 +25,7 @@ profiles:
         - evaluation_assets
         - functional_validation
         - instruction_quality
+        - operational_policy_compliance
     certified:
       checks:
         - enterprise_structure_validation
@@ -42,7 +43,7 @@ profiles:
       checks:
         - evaluation_assets
         - functional_validation
-        # - operational_policy_compliance  # Not yet implemented
+        - operational_policy_compliance
     certified:
       checks:
         - enterprise_structure_validation
@@ -62,6 +63,7 @@ profiles:
         - evaluation_assets
         - advanced_security_validation
         - functional_validation
+        - operational_policy_compliance
     certified:
       checks:
         - enterprise_structure_validation
@@ -81,6 +83,7 @@ profiles:
         - evaluation_assets
         - functional_validation
         - instruction_quality
+        - operational_policy_compliance
     certified:
       checks:
         - enterprise_structure_validation
@@ -105,7 +108,7 @@ profiles:
         - instruction_quality
         # Not yet implemented:
         # - registry_governance
-        # - operational_policy_compliance
+        - operational_policy_compliance
     certified:
       checks:
         - enterprise_structure_validation

scripts/aggregate_scorecard.py

@@ -45,6 +45,7 @@
 from abevalflow.gates.base import GateResult
 from abevalflow.gates.quality import get_all_quality_gates
 from abevalflow.gates.security import get_all_security_gates
+from abevalflow.operational_policy import check_operational_policy
 from abevalflow.schemas import CertificationPolicy, GatePolicy, SubmissionMetadata
 from abevalflow.scorecard import Scorecard, apply_combination_logic
 
@@ -315,12 +316,21 @@ def aggregate_scorecard(
 
     has_eval_assets = (submission_dir / "evals" / "evals.json").exists() or (submission_dir / "tests").exists()
 
+    operational_limits = certification_policy.get_operational_limits() if certification_policy else None
+    operational_policy_result = check_operational_policy(submission_dir, limits=operational_limits)
+    logger.info(
+        "Operational policy: passed=%s, score=%.3f",
+        operational_policy_result.passed,
+        operational_policy_result.score,
+    )
+
     certification = compute_certification(
         gates=gates,
         validation_passed=validation_passed,
         metadata_valid=metadata_valid,
         has_eval_assets=has_eval_assets,
         policy=certification_policy,
+        operational_policy_result=operational_policy_result,
     )
     logger.info(
         "Certification levels: foundational=%s, trusted=%s, certified=%s (highest=%s)",