From a1ff01a14a4aca4198abdf0d71438734fc30275e Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 7 Jul 2026 13:56:46 +0300 Subject: [PATCH 1/7] feat: implement Operational Policy Compliance certification check (APPENG-5623) Add deterministic operational policy checks for skill submissions: - Resource limits: CPU and memory within configurable max thresholds - Timeout compliance: agent timeout within reasonable bounds - Error handling: detect bare except:pass patterns in test files - Logging suppression: detect patterns in SKILL.md that suppress agent logging Wire into certification system via compute_certification() parameter, following the existing boolean pattern. Uncomment in TRUSTED_CHECKS and in mcp_server/full certification profiles. 28 new tests in test_operational_policy.py, all passing. Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/certification.py | 26 +-- abevalflow/operational_policy.py | 151 ++++++++++++++++ config/certification_profiles.yaml | 4 +- scripts/aggregate_scorecard.py | 5 + tests/test_certification.py | 2 +- tests/test_operational_policy.py | 271 +++++++++++++++++++++++++++++ 6 files changed, 445 insertions(+), 14 deletions(-) create mode 100644 abevalflow/operational_policy.py create mode 100644 tests/test_operational_policy.py diff --git a/abevalflow/certification.py b/abevalflow/certification.py index 5d0bd11..429e7c4 100644 --- a/abevalflow/certification.py +++ b/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.setdefault(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, diff --git a/abevalflow/operational_policy.py b/abevalflow/operational_policy.py new file mode 100644 index 0000000..67ca2d7 --- /dev/null +++ b/abevalflow/operational_policy.py @@ -0,0 +1,151 @@ +"""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 SubmissionMetadata + +logger = logging.getLogger(__name__) + +MAX_CPUS = 4 +MAX_MEMORY_MB = 8192 +MAX_AGENT_TIMEOUT_SEC = 3600.0 + +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), +] + + +def _check_resource_limits(metadata: SubmissionMetadata) -> list[str]: + """Check that resource requests are within policy limits.""" + issues = [] + if metadata.cpus > MAX_CPUS: + issues.append(f"CPU request ({metadata.cpus}) exceeds max allowed ({MAX_CPUS})") + if metadata.memory_mb > MAX_MEMORY_MB: + issues.append(f"Memory request ({metadata.memory_mb}MB) exceeds max allowed ({MAX_MEMORY_MB}MB)") + return issues + + +def _check_timeout_compliance(metadata: SubmissionMetadata) -> list[str]: + """Check that timeouts are within reasonable bounds.""" + issues = [] + if metadata.agent_timeout_sec > MAX_AGENT_TIMEOUT_SEC: + issues.append( + f"Agent timeout ({metadata.agent_timeout_sec}s) exceeds max allowed ({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.""" + issues = [] + if not skill_path.exists(): + return issues + + content = skill_path.read_text() + for pattern in LOGGING_SUPPRESSION_PATTERNS: + match = pattern.search(content) + if match: + issues.append(f"Logging suppression pattern found: '{match.group()}'") + + return issues + + +def check_operational_policy(submission_dir: Path) -> 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. + + Returns: + CheckResult with aggregated pass/fail, score, and message. + """ + all_issues: list[str] = [] + + metadata_path = submission_dir / "metadata.yaml" + if metadata_path.exists(): + try: + data = yaml.safe_load(metadata_path.read_text()) + metadata = SubmissionMetadata(**(data or {})) + all_issues.extend(_check_resource_limits(metadata)) + all_issues.extend(_check_timeout_compliance(metadata)) + 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, + ) diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index c922bc8..9940db0 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -42,7 +42,7 @@ profiles: checks: - evaluation_assets - functional_validation - # - operational_policy_compliance # Not yet implemented + - operational_policy_compliance certified: checks: - enterprise_structure_validation @@ -105,7 +105,7 @@ profiles: - instruction_quality # Not yet implemented: # - registry_governance - # - operational_policy_compliance + - operational_policy_compliance certified: checks: - enterprise_structure_validation diff --git a/scripts/aggregate_scorecard.py b/scripts/aggregate_scorecard.py index 971ec10..be0c7c8 100644 --- a/scripts/aggregate_scorecard.py +++ b/scripts/aggregate_scorecard.py @@ -33,6 +33,7 @@ import yaml from abevalflow.certification import compute_certification, load_profile +from abevalflow.operational_policy import check_operational_policy from abevalflow.compass_facts import ( CertificationFactPushResult, FactPushResult, @@ -315,12 +316,16 @@ def aggregate_scorecard( has_eval_assets = (submission_dir / "evals" / "evals.json").exists() or (submission_dir / "tests").exists() + operational_policy_result = check_operational_policy(submission_dir) + 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)", diff --git a/tests/test_certification.py b/tests/test_certification.py index b0461af..a65a494 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -348,7 +348,7 @@ def test_trusted_has_expected_checks(self) -> None: assert CheckId.FUNCTIONAL_VALIDATION in TRUSTED_CHECKS assert CheckId.INSTRUCTION_QUALITY in TRUSTED_CHECKS # registry_governance and operational_policy_compliance not yet implemented - assert len(TRUSTED_CHECKS) == 4 + assert len(TRUSTED_CHECKS) == 5 def test_certified_has_expected_checks(self) -> None: assert CheckId.ENTERPRISE_STRUCTURE_VALIDATION in CERTIFIED_CHECKS diff --git a/tests/test_operational_policy.py b/tests/test_operational_policy.py new file mode 100644 index 0000000..6e1d6ea --- /dev/null +++ b/tests/test_operational_policy.py @@ -0,0 +1,271 @@ +"""Tests for operational policy compliance checks.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from abevalflow.certification import CheckId +from abevalflow.operational_policy import ( + MAX_AGENT_TIMEOUT_SEC, + MAX_CPUS, + MAX_MEMORY_MB, + _check_error_handling, + _check_logging_suppression, + _check_resource_limits, + _check_timeout_compliance, + check_operational_policy, +) +from abevalflow.schemas import SubmissionMetadata + + +def _make_metadata(**overrides) -> SubmissionMetadata: + base = {"name": "test-submission"} + base.update(overrides) + return SubmissionMetadata(**base) + + +def _write_submission(tmp_path: Path, metadata: dict | None = None, skill_content: str = "", test_content: str = ""): + if metadata is not None: + (tmp_path / "metadata.yaml").write_text(yaml.dump(metadata)) + skills_dir = tmp_path / "skills" + skills_dir.mkdir(exist_ok=True) + if skill_content: + (skills_dir / "SKILL.md").write_text(skill_content) + tests_dir = tmp_path / "tests" + tests_dir.mkdir(exist_ok=True) + if test_content: + (tests_dir / "test_outputs.py").write_text(test_content) + + +class TestResourceLimits: + def test_default_values_pass(self): + metadata = _make_metadata() + issues = _check_resource_limits(metadata) + assert issues == [] + + def test_within_limits_pass(self): + metadata = _make_metadata(cpus=MAX_CPUS, memory_mb=MAX_MEMORY_MB) + issues = _check_resource_limits(metadata) + assert issues == [] + + def test_cpu_exceeds_limit(self): + metadata = _make_metadata(cpus=MAX_CPUS + 1) + issues = _check_resource_limits(metadata) + assert len(issues) == 1 + assert "CPU" in issues[0] + + def test_memory_exceeds_limit(self): + metadata = _make_metadata(memory_mb=MAX_MEMORY_MB + 1) + issues = _check_resource_limits(metadata) + assert len(issues) == 1 + assert "Memory" in issues[0] + + def test_both_exceed(self): + metadata = _make_metadata(cpus=MAX_CPUS + 1, memory_mb=MAX_MEMORY_MB + 1) + issues = _check_resource_limits(metadata) + assert len(issues) == 2 + + +class TestTimeoutCompliance: + def test_default_timeout_passes(self): + metadata = _make_metadata() + issues = _check_timeout_compliance(metadata) + assert issues == [] + + def test_within_limit_passes(self): + metadata = _make_metadata(agent_timeout_sec=MAX_AGENT_TIMEOUT_SEC) + issues = _check_timeout_compliance(metadata) + assert issues == [] + + def test_exceeds_limit(self): + metadata = _make_metadata(agent_timeout_sec=MAX_AGENT_TIMEOUT_SEC + 1) + issues = _check_timeout_compliance(metadata) + assert len(issues) == 1 + assert "timeout" in issues[0].lower() + + +class TestErrorHandling: + def test_no_test_file(self, tmp_path: Path): + issues = _check_error_handling(tmp_path / "nonexistent.py") + assert issues == [] + + def test_clean_test_file(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except ValueError as e:\n" + " raise AssertionError(f'Failed: {e}')\n" + ) + issues = _check_error_handling(test_file) + assert issues == [] + + def test_bare_except_pass(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except:\n" + " pass\n" + ) + issues = _check_error_handling(test_file) + assert len(issues) == 1 + assert "Bare" in issues[0] + + def test_except_exception_pass(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except Exception:\n" + " pass\n" + ) + issues = _check_error_handling(test_file) + assert len(issues) == 1 + assert "Exception" in issues[0] + + def test_except_with_body_is_ok(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except Exception:\n" + " print('error')\n" + " raise\n" + ) + issues = _check_error_handling(test_file) + assert issues == [] + + def test_except_ellipsis(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except:\n" + " ...\n" + ) + issues = _check_error_handling(test_file) + assert len(issues) == 1 + + def test_syntax_error_is_ignored(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text("def broken(:\n") + issues = _check_error_handling(test_file) + assert issues == [] + + +class TestLoggingSuppression: + def test_no_skill_file(self, tmp_path: Path): + issues = _check_logging_suppression(tmp_path / "nonexistent.md") + assert issues == [] + + def test_clean_skill(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nThis skill helps with code review.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + def test_do_not_log(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDo not log any output.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_disable_logging(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nPlease disable logging for this task.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_suppress_output(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nSuppress output from the agent.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_case_insensitive(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDISABLE LOGGING immediately.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + +class TestCheckOperationalPolicy: + def test_all_pass(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "good-skill"}, + skill_content="# Good Skill\n\nHelps with testing.\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE + assert result.passed is True + assert result.score == 1.0 + + def test_resource_limit_failure(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "heavy-skill", "cpus": MAX_CPUS + 1}, + skill_content="# Skill\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert "CPU" in result.message + + def test_error_handling_failure(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "bad-tests"}, + skill_content="# Skill\n", + test_content="def test_bad():\n try:\n run()\n except:\n pass\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert "except" in result.message.lower() or "Bare" in result.message + + def test_logging_suppression_failure(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "noisy-skill"}, + skill_content="# Skill\n\nDo not log any output.\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert "log" in result.message.lower() + + def test_no_metadata_still_checks_files(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata=None, + skill_content="# Skill\n\nDisable logging please.\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + + def test_empty_submission(self, tmp_path: Path): + result = check_operational_policy(tmp_path) + assert result.passed is True + assert result.score == 1.0 + + def test_score_degrades_with_issues(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "bad", "cpus": MAX_CPUS + 1, "memory_mb": MAX_MEMORY_MB + 1}, + skill_content="# Skill\n\nDo not log output.\n", + test_content="def test_bad():\n try:\n x()\n except:\n pass\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert result.score < 1.0 From d41341637e84bc41c7a12825ca00a7980bfeaa22 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 7 Jul 2026 14:36:33 +0300 Subject: [PATCH 2/7] fix: lint issues in operational policy check - Fix import sort order in aggregate_scorecard.py - Split long logger.info line to stay within 120 char limit - Remove unused pytest import from test_operational_policy.py Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/aggregate_scorecard.py | 8 ++++++-- tests/test_operational_policy.py | 1 - 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/scripts/aggregate_scorecard.py b/scripts/aggregate_scorecard.py index be0c7c8..7e78a2b 100644 --- a/scripts/aggregate_scorecard.py +++ b/scripts/aggregate_scorecard.py @@ -33,7 +33,6 @@ import yaml from abevalflow.certification import compute_certification, load_profile -from abevalflow.operational_policy import check_operational_policy from abevalflow.compass_facts import ( CertificationFactPushResult, FactPushResult, @@ -46,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 @@ -317,7 +317,11 @@ def aggregate_scorecard( has_eval_assets = (submission_dir / "evals" / "evals.json").exists() or (submission_dir / "tests").exists() operational_policy_result = check_operational_policy(submission_dir) - logger.info("Operational policy: passed=%s, score=%.3f", operational_policy_result.passed, operational_policy_result.score) + logger.info( + "Operational policy: passed=%s, score=%.3f", + operational_policy_result.passed, + operational_policy_result.score, + ) certification = compute_certification( gates=gates, diff --git a/tests/test_operational_policy.py b/tests/test_operational_policy.py index 6e1d6ea..f49b0fd 100644 --- a/tests/test_operational_policy.py +++ b/tests/test_operational_policy.py @@ -4,7 +4,6 @@ from pathlib import Path -import pytest import yaml from abevalflow.certification import CheckId From dc91e26c17975c6a76598ab3db74e00b0854e3e7 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 7 Jul 2026 15:30:56 +0300 Subject: [PATCH 3/7] feat: address code review findings for operational policy check - Configurable thresholds via OperationalLimits model in schemas.py, overridable through CertificationPolicy in metadata.yaml - Detect missing resource declarations (cpus, memory_mb, agent_timeout_sec) by checking raw YAML before Pydantic defaults apply - Add operational_policy_compliance to skill and agent certification profiles - Fix logging suppression false positives: skip lines with security qualifiers (passwords, credentials, tokens, etc.) - Add 3 integration tests verifying certification wiring (pass, fail, None) - 105 tests passing, ruff clean Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/operational_policy.py | 68 +++++++++++++++------ abevalflow/schemas.py | 29 +++++++++ config/certification_profiles.yaml | 2 + scripts/aggregate_scorecard.py | 3 +- tests/test_certification.py | 95 ++++++++++++++++++++++++++++++ tests/test_operational_policy.py | 90 ++++++++++++++++++++++++---- 6 files changed, 257 insertions(+), 30 deletions(-) diff --git a/abevalflow/operational_policy.py b/abevalflow/operational_policy.py index 67ca2d7..0cdd35c 100644 --- a/abevalflow/operational_policy.py +++ b/abevalflow/operational_policy.py @@ -20,7 +20,7 @@ from pydantic import ValidationError from abevalflow.certification import CheckId, CheckResult -from abevalflow.schemas import SubmissionMetadata +from abevalflow.schemas import OperationalLimits, SubmissionMetadata logger = logging.getLogger(__name__) @@ -36,23 +36,39 @@ 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, +) -def _check_resource_limits(metadata: SubmissionMetadata) -> list[str]: + +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 > MAX_CPUS: - issues.append(f"CPU request ({metadata.cpus}) exceeds max allowed ({MAX_CPUS})") - if metadata.memory_mb > MAX_MEMORY_MB: - issues.append(f"Memory request ({metadata.memory_mb}MB) exceeds max allowed ({MAX_MEMORY_MB}MB)") + 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) -> list[str]: +def _check_timeout_compliance(metadata: SubmissionMetadata, limits: OperationalLimits) -> list[str]: """Check that timeouts are within reasonable bounds.""" issues = [] - if metadata.agent_timeout_sec > MAX_AGENT_TIMEOUT_SEC: + if metadata.agent_timeout_sec > limits.max_agent_timeout_sec: issues.append( - f"Agent timeout ({metadata.agent_timeout_sec}s) exceeds max allowed ({MAX_AGENT_TIMEOUT_SEC}s)" + f"Agent timeout ({metadata.agent_timeout_sec}s) exceeds max allowed ({limits.max_agent_timeout_sec}s)" ) return issues @@ -88,39 +104,53 @@ def _check_error_handling(test_file: Path) -> list[str]: def _check_logging_suppression(skill_path: Path) -> list[str]: - """Check SKILL.md for patterns that suppress agent logging.""" + """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 pattern in LOGGING_SUPPRESSION_PATTERNS: - match = pattern.search(content) - if match: - issues.append(f"Logging suppression pattern found: '{match.group()}'") + 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) -> CheckResult: +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()) - metadata = SubmissionMetadata(**(data or {})) - all_issues.extend(_check_resource_limits(metadata)) - all_issues.extend(_check_timeout_compliance(metadata)) + 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: diff --git a/abevalflow/schemas.py b/abevalflow/schemas.py index ad9d05a..e6a41d7 100644 --- a/abevalflow/schemas.py +++ b/abevalflow/schemas.py @@ -233,6 +233,18 @@ class ExperimentType(StrEnum): CUSTOM = "custom" +class OperationalLimits(BaseModel): + """Configurable limits for operational policy compliance checks. + + Overrides the hardcoded defaults in operational_policy.py when specified + in metadata.yaml under certification_policy.trusted.operational_limits. + """ + + 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. @@ -246,6 +258,10 @@ class CertificationLevelPolicy(BaseModel): - basic_security_validation thresholds: basic_execution_validation: 0.5 + trusted: + operational_limits: + max_cpus: 8 + max_memory_mb: 16384 """ checks: list[str] | None = Field( @@ -263,6 +279,13 @@ class CertificationLevelPolicy(BaseModel): "score thresholds (0.0-1.0). Overrides hardcoded thresholds." ), ) + operational_limits: OperationalLimits | None = Field( + default=None, + description=( + "Operational policy limits for resource, timeout, and other checks. " + "Overrides hardcoded defaults in operational_policy.py." + ), + ) @field_validator("thresholds") @classmethod @@ -343,6 +366,12 @@ 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 from the trusted level policy.""" + if self.trusted is not None: + return self.trusted.operational_limits + return None + class CopySpec(BaseModel): """A source directory and its destination path inside the container.""" diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index 9940db0..a639d4e 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -25,6 +25,7 @@ profiles: - evaluation_assets - functional_validation - instruction_quality + - operational_policy_compliance certified: checks: - enterprise_structure_validation @@ -81,6 +82,7 @@ profiles: - evaluation_assets - functional_validation - instruction_quality + - operational_policy_compliance certified: checks: - enterprise_structure_validation diff --git a/scripts/aggregate_scorecard.py b/scripts/aggregate_scorecard.py index 7e78a2b..a8eb9eb 100644 --- a/scripts/aggregate_scorecard.py +++ b/scripts/aggregate_scorecard.py @@ -316,7 +316,8 @@ def aggregate_scorecard( has_eval_assets = (submission_dir / "evals" / "evals.json").exists() or (submission_dir / "tests").exists() - operational_policy_result = check_operational_policy(submission_dir) + 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, diff --git a/tests/test_certification.py b/tests/test_certification.py index a65a494..8c02b3e 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -1116,3 +1116,98 @@ def test_profile_integrates_with_compute_certification(self) -> None: # Should use skill profile's check configuration assert result is not None assert result.foundational is not None + + +class TestOperationalPolicyCertificationIntegration: + def test_passing_check_enables_trusted(self) -> None: + passing_result = CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=True, + score=1.0, + message="All passed", + ) + engine_gate = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + message="Pass", + ) + security_gate = GateResult( + gate_name="security", + gate_type=GateType.SECURITY, + passed=True, + score=1.0, + mode=GateMode.BLOCK, + message="Pass", + findings=[], + ) + quality_gate = GateResult( + gate_name="quality", + gate_type=GateType.QUALITY, + passed=True, + score=0.8, + mode=GateMode.WARN, + message="Pass", + ) + result = compute_certification( + gates=[engine_gate, security_gate, quality_gate], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=passing_result, + ) + assert result.trusted.passed is True + op_check = next( + c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE + ) + assert op_check.passed is True + + def test_failing_check_blocks_trusted(self) -> None: + failing_result = CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=False, + score=0.25, + message="CPU exceeds limit", + ) + engine_gate = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + message="Pass", + ) + result = compute_certification( + gates=[engine_gate], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=failing_result, + ) + assert result.trusted.passed is False + + def test_none_result_uses_default_not_implemented(self) -> None: + engine_gate = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + message="Pass", + ) + result = compute_certification( + gates=[engine_gate], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=None, + ) + op_check = next( + c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE + ) + assert op_check.passed is False + assert "not implemented" in op_check.message diff --git a/tests/test_operational_policy.py b/tests/test_operational_policy.py index f49b0fd..1e0c90d 100644 --- a/tests/test_operational_policy.py +++ b/tests/test_operational_policy.py @@ -13,11 +13,12 @@ MAX_MEMORY_MB, _check_error_handling, _check_logging_suppression, + _check_resource_declaration, _check_resource_limits, _check_timeout_compliance, check_operational_policy, ) -from abevalflow.schemas import SubmissionMetadata +from abevalflow.schemas import OperationalLimits, SubmissionMetadata def _make_metadata(**overrides) -> SubmissionMetadata: @@ -39,49 +40,72 @@ def _write_submission(tmp_path: Path, metadata: dict | None = None, skill_conten (tests_dir / "test_outputs.py").write_text(test_content) +class TestResourceDeclaration: + def test_all_declared(self): + data = {"name": "test", "cpus": 2, "memory_mb": 4096, "agent_timeout_sec": 600} + issues = _check_resource_declaration(data) + assert issues == [] + + def test_none_declared(self): + data = {"name": "test"} + issues = _check_resource_declaration(data) + assert len(issues) == 1 + assert "cpus" in issues[0] + assert "memory_mb" in issues[0] + assert "agent_timeout_sec" in issues[0] + + def test_partial_declared(self): + data = {"name": "test", "cpus": 2} + issues = _check_resource_declaration(data) + assert len(issues) == 1 + assert "memory_mb" in issues[0] + assert "agent_timeout_sec" in issues[0] + assert "cpus" not in issues[0] + + class TestResourceLimits: def test_default_values_pass(self): metadata = _make_metadata() - issues = _check_resource_limits(metadata) + issues = _check_resource_limits(metadata, OperationalLimits()) assert issues == [] def test_within_limits_pass(self): metadata = _make_metadata(cpus=MAX_CPUS, memory_mb=MAX_MEMORY_MB) - issues = _check_resource_limits(metadata) + issues = _check_resource_limits(metadata, OperationalLimits()) assert issues == [] def test_cpu_exceeds_limit(self): metadata = _make_metadata(cpus=MAX_CPUS + 1) - issues = _check_resource_limits(metadata) + issues = _check_resource_limits(metadata, OperationalLimits()) assert len(issues) == 1 assert "CPU" in issues[0] def test_memory_exceeds_limit(self): metadata = _make_metadata(memory_mb=MAX_MEMORY_MB + 1) - issues = _check_resource_limits(metadata) + issues = _check_resource_limits(metadata, OperationalLimits()) assert len(issues) == 1 assert "Memory" in issues[0] def test_both_exceed(self): metadata = _make_metadata(cpus=MAX_CPUS + 1, memory_mb=MAX_MEMORY_MB + 1) - issues = _check_resource_limits(metadata) + issues = _check_resource_limits(metadata, OperationalLimits()) assert len(issues) == 2 class TestTimeoutCompliance: def test_default_timeout_passes(self): metadata = _make_metadata() - issues = _check_timeout_compliance(metadata) + issues = _check_timeout_compliance(metadata, OperationalLimits()) assert issues == [] def test_within_limit_passes(self): metadata = _make_metadata(agent_timeout_sec=MAX_AGENT_TIMEOUT_SEC) - issues = _check_timeout_compliance(metadata) + issues = _check_timeout_compliance(metadata, OperationalLimits()) assert issues == [] def test_exceeds_limit(self): metadata = _make_metadata(agent_timeout_sec=MAX_AGENT_TIMEOUT_SEC + 1) - issues = _check_timeout_compliance(metadata) + issues = _check_timeout_compliance(metadata, OperationalLimits()) assert len(issues) == 1 assert "timeout" in issues[0].lower() @@ -196,12 +220,30 @@ def test_case_insensitive(self, tmp_path: Path): issues = _check_logging_suppression(skill_file) assert len(issues) == 1 + def test_security_qualified_not_flagged(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDo not log passwords or PII.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + def test_security_qualified_credentials(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nNo logging of credentials.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + def test_security_qualified_tokens(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDo not log sensitive tokens.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + class TestCheckOperationalPolicy: def test_all_pass(self, tmp_path: Path): _write_submission( tmp_path, - metadata={"name": "good-skill"}, + metadata={"name": "good-skill", "cpus": 2, "memory_mb": 4096, "agent_timeout_sec": 600}, skill_content="# Good Skill\n\nHelps with testing.\n", test_content="def test_ok():\n assert True\n", ) @@ -268,3 +310,31 @@ def test_score_degrades_with_issues(self, tmp_path: Path): result = check_operational_policy(tmp_path) assert result.passed is False assert result.score < 1.0 + + +class TestCustomLimits: + _full_resources = {"cpus": 2, "memory_mb": 4096, "agent_timeout_sec": 600} + + def test_custom_limits_allow_higher_cpu(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "cpus": 8} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_cpus=10)) + assert result.passed is True + + def test_custom_limits_still_enforce(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "cpus": 8} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_cpus=4)) + assert result.passed is False + + def test_custom_memory_limit(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "memory_mb": 16384} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_memory_mb=32768)) + assert result.passed is True + + def test_custom_timeout_limit(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "agent_timeout_sec": 7200} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_agent_timeout_sec=10000)) + assert result.passed is True From 4f41c1b9b9943a3b015454c2186e414b761f59a0 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 7 Jul 2026 15:43:52 +0300 Subject: [PATCH 4/7] refactor: remove dual constants, add plugin profile coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove MAX_CPUS/MAX_MEMORY_MB/MAX_AGENT_TIMEOUT_SEC module constants from operational_policy.py — single source of truth is now OperationalLimits defaults in schemas.py - Tests derive boundary values from OperationalLimits() directly - Add operational_policy_compliance to plugin certification profile for consistency across all profiles Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/operational_policy.py | 4 ---- config/certification_profiles.yaml | 1 + tests/test_operational_policy.py | 8 +++++--- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/abevalflow/operational_policy.py b/abevalflow/operational_policy.py index 0cdd35c..96aff92 100644 --- a/abevalflow/operational_policy.py +++ b/abevalflow/operational_policy.py @@ -24,10 +24,6 @@ logger = logging.getLogger(__name__) -MAX_CPUS = 4 -MAX_MEMORY_MB = 8192 -MAX_AGENT_TIMEOUT_SEC = 3600.0 - LOGGING_SUPPRESSION_PATTERNS = [ re.compile(r"\bdo\s+not\s+log\b", re.IGNORECASE), re.compile(r"\bdisable\s+logging\b", re.IGNORECASE), diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index a639d4e..aeeb89b 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -63,6 +63,7 @@ profiles: - evaluation_assets - advanced_security_validation - functional_validation + - operational_policy_compliance certified: checks: - enterprise_structure_validation diff --git a/tests/test_operational_policy.py b/tests/test_operational_policy.py index 1e0c90d..443739b 100644 --- a/tests/test_operational_policy.py +++ b/tests/test_operational_policy.py @@ -8,9 +8,6 @@ from abevalflow.certification import CheckId from abevalflow.operational_policy import ( - MAX_AGENT_TIMEOUT_SEC, - MAX_CPUS, - MAX_MEMORY_MB, _check_error_handling, _check_logging_suppression, _check_resource_declaration, @@ -20,6 +17,11 @@ ) from abevalflow.schemas import OperationalLimits, SubmissionMetadata +_DEFAULTS = OperationalLimits() +MAX_CPUS = _DEFAULTS.max_cpus +MAX_MEMORY_MB = _DEFAULTS.max_memory_mb +MAX_AGENT_TIMEOUT_SEC = _DEFAULTS.max_agent_timeout_sec + def _make_metadata(**overrides) -> SubmissionMetadata: base = {"name": "test-submission"} From 730aa01fbedcc112f1fad8fc4f94ab33b1195437 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 9 Jul 2026 13:51:58 +0300 Subject: [PATCH 5/7] fix: ruff format for test files Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/test_certification.py | 8 ++------ tests/test_operational_policy.py | 22 +++------------------- 2 files changed, 5 insertions(+), 25 deletions(-) diff --git a/tests/test_certification.py b/tests/test_certification.py index 8c02b3e..308a423 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -1160,9 +1160,7 @@ def test_passing_check_enables_trusted(self) -> None: operational_policy_result=passing_result, ) assert result.trusted.passed is True - op_check = next( - c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE - ) + op_check = next(c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE) assert op_check.passed is True def test_failing_check_blocks_trusted(self) -> None: @@ -1206,8 +1204,6 @@ def test_none_result_uses_default_not_implemented(self) -> None: has_eval_assets=True, operational_policy_result=None, ) - op_check = next( - c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE - ) + op_check = next(c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE) assert op_check.passed is False assert "not implemented" in op_check.message diff --git a/tests/test_operational_policy.py b/tests/test_operational_policy.py index 443739b..8e86a0b 100644 --- a/tests/test_operational_policy.py +++ b/tests/test_operational_policy.py @@ -131,13 +131,7 @@ def test_clean_test_file(self, tmp_path: Path): def test_bare_except_pass(self, tmp_path: Path): test_file = tmp_path / "test_outputs.py" - test_file.write_text( - "def test_something():\n" - " try:\n" - " result = run()\n" - " except:\n" - " pass\n" - ) + test_file.write_text("def test_something():\n try:\n result = run()\n except:\n pass\n") issues = _check_error_handling(test_file) assert len(issues) == 1 assert "Bare" in issues[0] @@ -145,11 +139,7 @@ def test_bare_except_pass(self, tmp_path: Path): def test_except_exception_pass(self, tmp_path: Path): test_file = tmp_path / "test_outputs.py" test_file.write_text( - "def test_something():\n" - " try:\n" - " result = run()\n" - " except Exception:\n" - " pass\n" + "def test_something():\n try:\n result = run()\n except Exception:\n pass\n" ) issues = _check_error_handling(test_file) assert len(issues) == 1 @@ -170,13 +160,7 @@ def test_except_with_body_is_ok(self, tmp_path: Path): def test_except_ellipsis(self, tmp_path: Path): test_file = tmp_path / "test_outputs.py" - test_file.write_text( - "def test_something():\n" - " try:\n" - " result = run()\n" - " except:\n" - " ...\n" - ) + test_file.write_text("def test_something():\n try:\n result = run()\n except:\n ...\n") issues = _check_error_handling(test_file) assert len(issues) == 1 From d140a7c325fdf97107dbaac5fa68024dd0820b86 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 9 Jul 2026 14:15:41 +0300 Subject: [PATCH 6/7] refactor: move operational_limits to CertificationPolicy operational_limits was on CertificationLevelPolicy (per-level model) but only consumed from the trusted level. This broke the symmetry pattern of checks/thresholds and allowed silent misconfiguration. Moved it to the top-level CertificationPolicy where it belongs. Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/schemas.py | 32 +++++++++++++++----------------- tests/test_certification.py | 2 +- 2 files changed, 16 insertions(+), 18 deletions(-) diff --git a/abevalflow/schemas.py b/abevalflow/schemas.py index e6a41d7..1887f31 100644 --- a/abevalflow/schemas.py +++ b/abevalflow/schemas.py @@ -236,8 +236,8 @@ class ExperimentType(StrEnum): class OperationalLimits(BaseModel): """Configurable limits for operational policy compliance checks. - Overrides the hardcoded defaults in operational_policy.py when specified - in metadata.yaml under certification_policy.trusted.operational_limits. + 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") @@ -258,10 +258,6 @@ class CertificationLevelPolicy(BaseModel): - basic_security_validation thresholds: basic_execution_validation: 0.5 - trusted: - operational_limits: - max_cpus: 8 - max_memory_mb: 16384 """ checks: list[str] | None = Field( @@ -279,13 +275,6 @@ class CertificationLevelPolicy(BaseModel): "score thresholds (0.0-1.0). Overrides hardcoded thresholds." ), ) - operational_limits: OperationalLimits | None = Field( - default=None, - description=( - "Operational policy limits for resource, timeout, and other checks. " - "Overrides hardcoded defaults in operational_policy.py." - ), - ) @field_validator("thresholds") @classmethod @@ -314,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 @@ -343,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.""" @@ -367,10 +367,8 @@ def get_threshold(self, check_id: str) -> float | None: return result def get_operational_limits(self) -> OperationalLimits | None: - """Get operational limits from the trusted level policy.""" - if self.trusted is not None: - return self.trusted.operational_limits - return None + """Get operational limits for the operational policy compliance check.""" + return self.operational_limits class CopySpec(BaseModel): diff --git a/tests/test_certification.py b/tests/test_certification.py index 308a423..471cd76 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -347,7 +347,7 @@ def test_trusted_has_expected_checks(self) -> None: assert CheckId.ADVANCED_SECURITY_VALIDATION in TRUSTED_CHECKS assert CheckId.FUNCTIONAL_VALIDATION in TRUSTED_CHECKS assert CheckId.INSTRUCTION_QUALITY in TRUSTED_CHECKS - # registry_governance and operational_policy_compliance not yet implemented + # registry_governance not yet implemented assert len(TRUSTED_CHECKS) == 5 def test_certified_has_expected_checks(self) -> None: From 8b6cfe5097b0df4c9358134631eb7b5a1d6cb2a9 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 9 Jul 2026 15:09:23 +0300 Subject: [PATCH 7/7] fix: use direct assignment instead of setdefault for operational policy result setdefault would silently drop the explicitly-passed result if a gate ever maps to OPERATIONAL_POLICY_COMPLIANCE. Direct assignment makes the intent clear: this is the authoritative result for this check. Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/certification.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/abevalflow/certification.py b/abevalflow/certification.py index 429e7c4..d7c53f0 100644 --- a/abevalflow/certification.py +++ b/abevalflow/certification.py @@ -583,7 +583,7 @@ def compute_certification( ) if operational_policy_result is not None: - all_checks.setdefault(CheckId.OPERATIONAL_POLICY_COMPLIANCE, operational_policy_result) + all_checks[CheckId.OPERATIONAL_POLICY_COMPLIANCE] = operational_policy_result else: all_checks.setdefault( CheckId.OPERATIONAL_POLICY_COMPLIANCE,