Skip to content

fix: resolve ruff linting issues across codebase #41

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GuyZivRH merged 1 commit into from
Jun 24, 2026
Merged

fix: resolve ruff linting issues across codebase #41

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 10 additions & 8 deletions Docs/certification_and_checks.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,14 +42,16 @@ Levels are **hierarchical**: Certified requires Trusted, Trusted requires Founda

## Artifact Applicability Matrix

Different artifact types benefit from different checks:

| Artifact Type | Focus Area | Key Checks |
|---------------|------------|------------|
| **Skills** (Prompt + Code) | A/B gap testing, execution isolation | Functional Validation, Execution Validation |
| **MCP Servers** (Tool APIs) | API contracts, schema accuracy | Resilience/Chaos Testing, Operational Policy |
| **Plugins** (REST APIs) | OpenAPI validation, auth flows | Security Validation, Metadata Compliance |
| **Agent Evals** (Autonomous) | Reasoning trace, multi-turn | Advanced Agent Validation, Safety Guardrails |
Different artifact types benefit from different checks. Each artifact type has a dedicated certification profile (see `config/certification_profiles.yaml`) that defines which checks apply at each level.

| Artifact Type | Focus Area | Foundational | Trusted | Certified |
|---------------|------------|--------------|---------|-----------|
| **Skills** (Prompt + Code) | A/B gap testing, execution isolation | Structure, Security, Execution, Quality, Metadata | Eval Assets, Functional, Instruction Quality | Enterprise Structure, Advanced Agent |
| **MCP Servers** (Tool APIs) | API contracts, operational stability | Structure, Security, Metadata | Eval Assets, Functional | Enterprise Structure, Enterprise Security |
| **Plugins** (REST APIs) | OpenAPI validation, auth flows | Structure, Security, Metadata | Eval Assets, Advanced Security, Functional | Enterprise Structure, Enterprise Security |
| **Agent Evals** (Autonomous) | Reasoning, safety, multi-turn | Structure, Security, Execution, Metadata | Eval Assets, Functional, Instruction Quality | Enterprise Structure, Advanced Agent |

**Profile selection** is done at pipeline deployment level via `--certification-profile=<type>`. Default is `skill`.

---

Expand Down
56 changes: 20 additions & 36 deletions abevalflow/certification.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ def _load_profiles_yaml(profiles_path: Path | None = None) -> dict[str, Any]:
path = profiles_path or DEFAULT_PROFILES_PATH
if not path.exists():
raise FileNotFoundError(f"Certification profiles not found: {path}")

with open(path) as f:
return yaml.safe_load(f)

Expand Down Expand Up @@ -156,7 +156,7 @@ def get_default_profile_name(profiles_path: Path | None = None) -> str:
def load_profile(
profile_name: str | None = None,
profiles_path: Path | None = None,
) -> "CertificationPolicy":
) -> CertificationPolicy:
"""Load a certification profile and return it as a CertificationPolicy.

Profiles define which checks apply to each certification level for different
Expand All @@ -183,41 +183,39 @@ def load_profile(

data = _load_profiles_yaml(profiles_path)
profiles = data.get("profiles", {})

name = profile_name or data.get("default_profile", "skill")

if name not in profiles:
available = list(profiles.keys())
raise ValueError(
f"Unknown certification profile '{name}'. Available: {available}"
)

raise ValueError(f"Unknown certification profile '{name}'. Available: {available}")

profile_data = profiles[name]
logger.info("Loading certification profile '%s': %s", name, profile_data.get("description", ""))

# Build CertificationPolicy from profile data
foundational = None
trusted = None
certified = None

if "foundational" in profile_data:
foundational = CertificationLevelPolicy(
checks=profile_data["foundational"].get("checks"),
thresholds=profile_data["foundational"].get("thresholds"),
)

if "trusted" in profile_data:
trusted = CertificationLevelPolicy(
checks=profile_data["trusted"].get("checks"),
thresholds=profile_data["trusted"].get("thresholds"),
)

if "certified" in profile_data:
certified = CertificationLevelPolicy(
checks=profile_data["certified"].get("checks"),
thresholds=profile_data["certified"].get("thresholds"),
)

return CertificationPolicy(
foundational=foundational,
trusted=trusted,
Expand Down Expand Up @@ -370,9 +368,7 @@ def _map_gate_to_checks(
details={"source_implementation": source_impl},
)
)
advanced_threshold = _get_threshold(
CheckId.ADVANCED_AGENT_VALIDATION, threshold_overrides
)
advanced_threshold = _get_threshold(CheckId.ADVANCED_AGENT_VALIDATION, threshold_overrides)
if gate.score >= advanced_threshold:
checks.append(
CheckResult(
Expand Down Expand Up @@ -413,9 +409,7 @@ def _map_gate_to_checks(
},
)
)
adv_security_threshold = _get_threshold(
CheckId.ADVANCED_SECURITY_VALIDATION, threshold_overrides
)
adv_security_threshold = _get_threshold(CheckId.ADVANCED_SECURITY_VALIDATION, threshold_overrides)
high_score = gate.score >= adv_security_threshold and gate.passed
checks.append(
CheckResult(
Expand Down Expand Up @@ -459,9 +453,7 @@ def _map_gate_to_checks(
details={"source_implementation": source_impl},
)
)
instruction_threshold = _get_threshold(
CheckId.INSTRUCTION_QUALITY, threshold_overrides
)
instruction_threshold = _get_threshold(CheckId.INSTRUCTION_QUALITY, threshold_overrides)
checks.append(
CheckResult(
check_id=CheckId.INSTRUCTION_QUALITY,
Expand Down Expand Up @@ -493,15 +485,13 @@ def _validate_check_ids(check_ids: list[str]) -> list[CheckId]:
result = []
for check_id in check_ids:
if check_id not in valid_ids:
raise ValueError(
f"Invalid check ID '{check_id}'. Valid IDs are: {sorted(valid_ids)}"
)
raise ValueError(f"Invalid check ID '{check_id}'. Valid IDs are: {sorted(valid_ids)}")
result.append(CheckId(check_id))
return result


def _collect_threshold_overrides(
policy: "CertificationPolicy | None",
policy: CertificationPolicy | None,
) -> dict[str, float]:
"""Collect all threshold overrides from a certification policy."""
if policy is None:
Expand All @@ -519,7 +509,7 @@ def compute_certification(
validation_passed: bool = True,
metadata_valid: bool = True,
has_eval_assets: bool = True,
policy: "CertificationPolicy | None" = None,
policy: CertificationPolicy | None = None,
) -> CertificationResult:
"""Compute certification levels from gate results.

Expand Down Expand Up @@ -664,15 +654,9 @@ def build_level(
return LevelResult(level=level, passed=all_passed, checks=level_checks)

# Build all levels
foundational_result = build_level(
CertificationLevel.FOUNDATIONAL, "foundational", FOUNDATIONAL_CHECKS
)
trusted_result = build_level(
CertificationLevel.TRUSTED, "trusted", TRUSTED_CHECKS
)
certified_result = build_level(
CertificationLevel.CERTIFIED, "certified", CERTIFIED_CHECKS
)
foundational_result = build_level(CertificationLevel.FOUNDATIONAL, "foundational", FOUNDATIONAL_CHECKS)
trusted_result = build_level(CertificationLevel.TRUSTED, "trusted", TRUSTED_CHECKS)
certified_result = build_level(CertificationLevel.CERTIFIED, "certified", CERTIFIED_CHECKS)

# Enforce hierarchy: lower levels must pass for higher levels to pass
# If foundational fails, trusted and certified cannot pass
Expand Down
52 changes: 27 additions & 25 deletions abevalflow/compass_facts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import re
import urllib.error
import urllib.request
from datetime import datetime, timezone
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel, Field
Expand Down Expand Up @@ -65,13 +65,13 @@ class FactPushResult(BaseModel):
success: bool = Field(description="Whether the push succeeded")
error: str | None = Field(default=None, description="Error message if push failed")
pushed_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
default_factory=lambda: datetime.now(UTC),
description="Timestamp of the push attempt",
)


def _build_fact_payload(
gate_result: "GateResult",
gate_result: GateResult,
entity_ref: str,
fact_ref: str,
) -> dict[str, Any]:
Expand Down Expand Up @@ -100,7 +100,7 @@ def _build_fact_payload(
"evaluated_at": (
gate_result.evaluated_at.isoformat()
if gate_result.evaluated_at
else datetime.now(timezone.utc).isoformat()
else datetime.now(UTC).isoformat()
),
},
}
Expand All @@ -109,7 +109,7 @@ def _build_fact_payload(


def push_gate_fact(
gate_result: "GateResult",
gate_result: GateResult,
endpoint: str,
entity_ref: str,
fact_ref: str,
Expand Down Expand Up @@ -139,7 +139,7 @@ def push_gate_fact(
}
if bearer_token:
headers["Authorization"] = f"Bearer {bearer_token}"

request = urllib.request.Request(
endpoint,
data=data,
Expand Down Expand Up @@ -210,8 +210,8 @@ def push_gate_fact(


def push_gate_fact_from_config(
gate_result: "GateResult",
push_facts_config: "PushFactsConfig",
gate_result: GateResult,
push_facts_config: PushFactsConfig,
) -> FactPushResult:
"""Push a gate result using PushFactsConfig settings.

Expand Down Expand Up @@ -250,7 +250,7 @@ def push_gate_fact_from_config(


def validate_push_facts_config(
push_facts_config: "PushFactsConfig | None",
push_facts_config: PushFactsConfig | None,
gates_with_push_fact: list[str],
) -> None:
"""Validate push_facts configuration and warn if misconfigured.
Expand Down Expand Up @@ -280,13 +280,13 @@ class CertificationFactPushResult(BaseModel):
success: bool = Field(description="Whether the push succeeded")
error: str | None = Field(default=None, description="Error message if push failed")
pushed_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
default_factory=lambda: datetime.now(UTC),
description="Timestamp of the push attempt",
)


def _build_certification_level_payload(
level_result: "LevelResult",
level_result: LevelResult,
entity_ref: str,
fact_ref: str,
) -> dict[str, Any]:
Expand All @@ -302,14 +302,16 @@ def _build_certification_level_payload(
"""
checks_data = []
for check in level_result.checks:
checks_data.append({
"check_id": check.check_id.value,
"name": check.name,
"passed": check.passed,
"score": check.score,
"message": check.message,
"source_gate": check.source_gate,
})
checks_data.append(
{
"check_id": check.check_id.value,
"name": check.name,
"passed": check.passed,
"score": check.score,
"message": check.message,
"source_gate": check.source_gate,
}
)

# Determine if this level failed due to hierarchy (all checks passed but level failed)
all_checks_passed = level_result.checks_total > 0 and level_result.checks_passed == level_result.checks_total
Expand All @@ -322,7 +324,7 @@ def _build_certification_level_payload(
"checks_total": level_result.checks_total,
"overall_score": level_result.overall_score,
"checks": checks_data,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
"evaluated_at": datetime.now(UTC).isoformat(),
}

if hierarchy_forced_failure:
Expand All @@ -340,7 +342,7 @@ def _build_certification_level_payload(


def _build_certification_summary_payload(
certification_result: "CertificationResult",
certification_result: CertificationResult,
entity_ref: str,
fact_ref: str,
) -> dict[str, Any]:
Expand Down Expand Up @@ -371,15 +373,15 @@ def _build_certification_summary_payload(
"trusted_score": certification_result.trusted.overall_score,
"certified_passed": certification_result.certified.passed,
"certified_score": certification_result.certified.overall_score,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
"evaluated_at": datetime.now(UTC).isoformat(),
},
}
]
}


def push_certification_level_fact(
level_result: "LevelResult",
level_result: LevelResult,
endpoint: str,
entity_ref: str,
fact_ref: str,
Expand Down Expand Up @@ -534,8 +536,8 @@ def _push_raw_fact(


def push_certification_facts(
certification_result: "CertificationResult",
push_facts_config: "PushFactsConfig",
certification_result: CertificationResult,
push_facts_config: PushFactsConfig,
) -> list[CertificationFactPushResult]:
"""Push all certification level facts to Compass.

Expand Down
2 changes: 1 addition & 1 deletion abevalflow/db/engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@
import os

from sqlalchemy import Engine, create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session, sessionmaker
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

from abevalflow.db.models import Base
Expand Down
Loading
Loading