From d29d10b3abe40c1eeb1ac0caf77b6c4f96374429 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Mon, 13 Jul 2026 12:27:57 +0300 Subject: [PATCH 01/10] feat: implement Enterprise Behavioral Testing certification check Add the ENTERPRISE_BEHAVIORAL_TESTING check (Certified level) with four sub-checks: consistency (trial variance), edge case testing, long-running stability, and failure mode analysis. - Consistency check flags skills with high trial variance (std_reward > 0.3) - Edge case testing: validation, scaffolding, analysis, and new BEHAVIORAL gate type with EdgeCaseGate registry - Stability and failure mode sub-checks implemented and tested, staged for wiring when upstream data producers exist - Certification wiring: uncommented in CERTIFIED_CHECKS, updated profiles, weighted composite with minimum-2 sub-check coverage requirement - Scaffold refactored: extracted _write_rendered_templates and _copy_root_dirs - 55+ new tests including end-to-end scaffold-to-certify integration test Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/certification.py | 338 +++++++- abevalflow/gates/base.py | 1 + abevalflow/gates/behavioral/__init__.py | 49 ++ abevalflow/gates/behavioral/base.py | 43 + abevalflow/gates/behavioral/edge_case.py | 117 +++ .../harbor_agents/verifiers/llm_judge.py | 8 + abevalflow/report.py | 12 + config/certification_profiles.yaml | 4 +- scripts/aggregate_scorecard.py | 72 +- scripts/analyze.py | 68 +- scripts/monitor.py | 51 ++ scripts/scaffold.py | 112 ++- scripts/validate.py | 29 + tests/test_behavioral_checks.py | 753 ++++++++++++++++++ tests/test_certification.py | 5 +- 15 files changed, 1619 insertions(+), 43 deletions(-) create mode 100644 abevalflow/gates/behavioral/__init__.py create mode 100644 abevalflow/gates/behavioral/base.py create mode 100644 abevalflow/gates/behavioral/edge_case.py create mode 100644 tests/test_behavioral_checks.py diff --git a/abevalflow/certification.py b/abevalflow/certification.py index d7c53f0a..12633447 100644 --- a/abevalflow/certification.py +++ b/abevalflow/certification.py @@ -94,7 +94,7 @@ class CheckId(StrEnum): CERTIFIED_CHECKS = [ CheckId.ENTERPRISE_STRUCTURE_VALIDATION, CheckId.ENTERPRISE_SECURITY_REVIEW, - # CheckId.ENTERPRISE_BEHAVIORAL_TESTING, # Not yet implemented + CheckId.ENTERPRISE_BEHAVIORAL_TESTING, CheckId.ADVANCED_AGENT_VALIDATION, # CheckId.CONTINUOUS_OPTIMIZATION, # Not yet implemented ] @@ -314,8 +314,309 @@ def get_level_result(self, level: CertificationLevel) -> LevelResult | None: CheckId.ADVANCED_AGENT_VALIDATION: 0.8, CheckId.ADVANCED_SECURITY_VALIDATION: 0.9, CheckId.INSTRUCTION_QUALITY: 0.7, + CheckId.ENTERPRISE_BEHAVIORAL_TESTING: 0.6, } +BEHAVIORAL_SUB_CHECK_WEIGHTS = { + "consistency": 0.3, + "edge_case": 0.3, + "stability": 0.2, + "failure_mode": 0.2, +} + +DEFAULT_CONSISTENCY_THRESHOLD = 0.3 +DEFAULT_STABILITY_VARIANCE_THRESHOLD = 0.1 +DEFAULT_EDGE_CASE_PASS_THRESHOLD = 0.5 +DEFAULT_FAILURE_MODE_THRESHOLD = 0.5 + + +def _check_consistency( + behavioral_data: dict[str, Any], + threshold: float = DEFAULT_CONSISTENCY_THRESHOLD, +) -> CheckResult: + """Check trial variance for inconsistent behavior. + + Skills with high trial variance (std_reward > threshold) are flagged. + Data comes from the existing evaluation run — no new pipeline steps needed. + """ + std_reward = behavioral_data.get("std_reward") + + if std_reward is None: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Consistency Check", + passed=True, + score=1.0, + message="No variance data available (single trial or missing data)", + details={"sub_check": "consistency", "status": "no_data"}, + ) + + passed = std_reward <= threshold + score = min(1.0, max(0.0, 1.0 - 0.5 * (std_reward / threshold))) + + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Consistency Check", + passed=passed, + score=score, + message=( + f"Trial variance {std_reward:.3f} within threshold {threshold}" + if passed + else f"Inconsistent behavior: trial variance {std_reward:.3f} exceeds threshold {threshold}" + ), + details={"sub_check": "consistency", "std_reward": std_reward, "threshold": threshold}, + ) + + +def _check_stability( + behavioral_data: dict[str, Any], + variance_threshold: float = DEFAULT_STABILITY_VARIANCE_THRESHOLD, +) -> CheckResult: + """Check score drift across consecutive monitoring runs. + + Passes if score variance across the last N runs is below the threshold. + Skips gracefully when insufficient historical data exists. + """ + stability_data = behavioral_data.get("stability") + + if stability_data is None: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Stability Check", + passed=True, + score=1.0, + message="No stability data available", + details={"sub_check": "stability", "status": "no_data"}, + ) + + run_count = stability_data.get("run_count", 0) + if run_count < 3: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Stability Check", + passed=True, + score=1.0, + message=f"Insufficient history ({run_count} runs, need at least 3)", + details={"sub_check": "stability", "status": "no_data", "run_count": run_count}, + ) + + variance = stability_data.get("score_variance", 0.0) + passed = variance <= variance_threshold + score = min(1.0, max(0.0, 1.0 - 0.5 * (variance / variance_threshold))) + + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Stability Check", + passed=passed, + score=score, + message=( + f"Score variance {variance:.4f} within threshold {variance_threshold}" + if passed + else f"Score drift detected: variance {variance:.4f} exceeds threshold {variance_threshold}" + ), + details={ + "sub_check": "stability", + "score_variance": variance, + "threshold": variance_threshold, + "run_count": run_count, + }, + ) + + +def _check_edge_case_results( + behavioral_data: dict[str, Any], + pass_threshold: float = DEFAULT_EDGE_CASE_PASS_THRESHOLD, +) -> CheckResult: + """Check edge case pass rate. + + Score = fraction of edge cases where the agent still passes verification. + """ + edge_case_data = behavioral_data.get("edge_cases") + + if edge_case_data is None: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Edge Case Check", + passed=True, + score=1.0, + message="No edge case data available", + details={"sub_check": "edge_case", "status": "no_data"}, + ) + + total = edge_case_data.get("total", 0) + passed_count = edge_case_data.get("passed", 0) + + if total == 0: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Edge Case Check", + passed=True, + score=1.0, + message="No edge cases defined", + details={"sub_check": "edge_case", "status": "no_data", "total": 0}, + ) + + pass_rate = passed_count / total + passed = pass_rate >= pass_threshold + + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Edge Case Check", + passed=passed, + score=pass_rate, + message=( + f"Edge case pass rate {pass_rate:.0%} ({passed_count}/{total})" + if passed + else f"Edge case failures: {pass_rate:.0%} ({passed_count}/{total}) below threshold {pass_threshold:.0%}" + ), + details={ + "sub_check": "edge_case", + "total": total, + "passed": passed_count, + "pass_rate": pass_rate, + "threshold": pass_threshold, + }, + ) + + +def _check_failure_mode(behavioral_data: dict[str, Any]) -> CheckResult: + """Check failure mode handling scores from LLM judge. + + Evaluates whether the agent handles errors gracefully and acknowledges + uncertainty when appropriate. + """ + failure_mode_data = behavioral_data.get("failure_mode") + + if failure_mode_data is None: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Failure Mode Check", + passed=True, + score=1.0, + message="No failure mode data available", + details={"sub_check": "failure_mode", "status": "no_data"}, + ) + + score = failure_mode_data.get("score", 0.0) + threshold = failure_mode_data.get("threshold", DEFAULT_FAILURE_MODE_THRESHOLD) + passed = score >= threshold + + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Failure Mode Check", + passed=passed, + score=score, + message=( + f"Failure mode score {score:.2f} meets threshold {threshold}" + if passed + else f"Failure mode score {score:.2f} below threshold {threshold}" + ), + details={ + "sub_check": "failure_mode", + "score": score, + "threshold": threshold, + }, + ) + + +def _compute_behavioral_testing_check( + behavioral_data: dict[str, Any], +) -> CheckResult: + """Compute the combined Enterprise Behavioral Testing check. + + Combines four sub-checks with configurable weights: + - Consistency (trial variance): 30% + - Edge case pass rate: 30% + - Stability (score drift): 20% + - Failure mode handling: 20% + + Pass/fail is determined by whether all active sub-checks pass individually. + The composite score is a weighted average for reporting purposes but does + not gate certification — each sub-check enforces its own threshold. + + Missing sub-checks are skipped (not counted as failures), but at least + 2 sub-checks must have real data for the check to pass. This prevents + a skill from achieving Certified with only a single passing sub-check. + """ + sub_checks = { + "consistency": _check_consistency(behavioral_data), + "edge_case": _check_edge_case_results(behavioral_data), + "stability": _check_stability(behavioral_data), + "failure_mode": _check_failure_mode(behavioral_data), + } + + weighted_score = 0.0 + total_weight = 0.0 + all_passed = True + sub_results: dict[str, Any] = {} + + for key, check in sub_checks.items(): + weight = BEHAVIORAL_SUB_CHECK_WEIGHTS[key] + has_data = check.details.get("status") != "no_data" + + if has_data: + weighted_score += check.score * weight + total_weight += weight + if not check.passed: + all_passed = False + + sub_results[key] = { + "passed": check.passed, + "score": check.score, + "message": check.message, + "has_data": has_data, + } + + active_checks = [k for k, v in sub_results.items() if v["has_data"]] + n_active = len(active_checks) + min_required = 2 + + if total_weight == 0 or n_active == 0: + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Enterprise Behavioral Testing", + passed=False, + score=0.0, + message="No behavioral testing data available", + details={"sub_checks": sub_results}, + ) + + if n_active < min_required: + final_score = weighted_score / total_weight + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Enterprise Behavioral Testing", + passed=False, + score=final_score, + message=( + f"Insufficient behavioral coverage: {n_active} sub-check(s) " + f"with data, need at least {min_required}" + ), + details={"sub_checks": sub_results, "weights": BEHAVIORAL_SUB_CHECK_WEIGHTS}, + ) + + final_score = weighted_score / total_weight + passed = all_passed + + failed_checks = [k for k, v in sub_results.items() if v["has_data"] and not v["passed"]] + + if passed: + message = f"Behavioral testing passed ({n_active} sub-checks, score {final_score:.2f})" + else: + message = ( + f"Behavioral testing failed: " + f"failed sub-checks: {', '.join(failed_checks)}" + ) + + return CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Enterprise Behavioral Testing", + passed=passed, + score=final_score, + message=message, + details={"sub_checks": sub_results, "weights": BEHAVIORAL_SUB_CHECK_WEIGHTS}, + ) + def _get_threshold( check_id: CheckId, @@ -466,6 +767,10 @@ def _map_gate_to_checks( ) ) + # GateType.BEHAVIORAL gates (e.g. edge-case) contribute to the scorecard + # but not to certification checks — behavioral certification is computed + # from the behavioral_data parameter via _compute_behavioral_testing_check(). + return checks @@ -511,6 +816,7 @@ def compute_certification( has_eval_assets: bool = True, policy: CertificationPolicy | None = None, operational_policy_result: CheckResult | None = None, + behavioral_data: dict[str, Any] | None = None, ) -> CertificationResult: """Compute certification levels from gate results. @@ -520,6 +826,11 @@ def compute_certification( metadata_valid: Whether metadata.yaml validation passed has_eval_assets: Whether evaluation assets (evals.json, tests) exist policy: Optional certification policy for custom checks and thresholds + behavioral_data: Optional behavioral testing data containing: + - std_reward: Trial variance from evaluation run + - edge_cases: {total, passed} edge case results + - stability: {score_variance, run_count} from monitoring + - failure_mode: {score, threshold} from LLM judge Returns: Complete certification result with all levels @@ -596,16 +907,21 @@ def compute_certification( ), ) - all_checks.setdefault( - CheckId.ENTERPRISE_BEHAVIORAL_TESTING, - CheckResult( - check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, - name="Enterprise Behavioral Testing", - passed=False, - score=0.0, - message="Behavioral testing not implemented", - ), - ) + if behavioral_data is not None: + all_checks[CheckId.ENTERPRISE_BEHAVIORAL_TESTING] = _compute_behavioral_testing_check( + behavioral_data + ) + else: + all_checks.setdefault( + CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + CheckResult( + check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + name="Enterprise Behavioral Testing", + passed=False, + score=0.0, + message="No behavioral testing data provided", + ), + ) all_checks.setdefault( CheckId.CONTINUOUS_OPTIMIZATION, diff --git a/abevalflow/gates/base.py b/abevalflow/gates/base.py index faff9ba0..fce9b592 100644 --- a/abevalflow/gates/base.py +++ b/abevalflow/gates/base.py @@ -22,6 +22,7 @@ class GateType(StrEnum): ENGINE = "engine" SECURITY = "security" QUALITY = "quality" + BEHAVIORAL = "behavioral" class GateMode(StrEnum): diff --git a/abevalflow/gates/behavioral/__init__.py b/abevalflow/gates/behavioral/__init__.py new file mode 100644 index 00000000..77f2ea51 --- /dev/null +++ b/abevalflow/gates/behavioral/__init__.py @@ -0,0 +1,49 @@ +"""Behavioral gate registry for unified scorecard.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from abevalflow.gates.behavioral.base import BehavioralGate + +_BEHAVIORAL_GATE_REGISTRY: dict[str, type[BehavioralGate]] = {} + + +def register_behavioral_gate(name: str): + """Decorator to register a behavioral gate class.""" + + def decorator(cls: type[BehavioralGate]) -> type[BehavioralGate]: + _BEHAVIORAL_GATE_REGISTRY[name] = cls + return cls + + return decorator + + +def get_behavioral_gate(name: str) -> BehavioralGate: + """Get a behavioral gate instance by name.""" + if name not in _BEHAVIORAL_GATE_REGISTRY: + available = ", ".join(_BEHAVIORAL_GATE_REGISTRY.keys()) + raise KeyError(f"Unknown behavioral gate '{name}'. Available: {available}") + return _BEHAVIORAL_GATE_REGISTRY[name]() + + +def get_all_behavioral_gates() -> list[BehavioralGate]: + """Get instances of all registered behavioral gates.""" + return [cls() for cls in _BEHAVIORAL_GATE_REGISTRY.values()] + + +def get_all_behavioral_gate_names() -> list[str]: + """Get all registered behavioral gate names.""" + return list(_BEHAVIORAL_GATE_REGISTRY.keys()) + + +from abevalflow.gates.behavioral.edge_case import EdgeCaseGate + +__all__ = [ + "register_behavioral_gate", + "get_behavioral_gate", + "get_all_behavioral_gates", + "get_all_behavioral_gate_names", + "EdgeCaseGate", +] diff --git a/abevalflow/gates/behavioral/base.py b/abevalflow/gates/behavioral/base.py new file mode 100644 index 00000000..2c320ee7 --- /dev/null +++ b/abevalflow/gates/behavioral/base.py @@ -0,0 +1,43 @@ +"""Base class for behavioral gates.""" + +from __future__ import annotations + +import logging +from abc import ABC, abstractmethod +from pathlib import Path + +from abevalflow.gates.base import GateResult +from abevalflow.schemas import GatePolicy + +logger = logging.getLogger(__name__) + + +class BehavioralGate(ABC): + """Abstract base class for behavioral gates. + + Behavioral gates evaluate skill behavior under stress, edge cases, + and unusual conditions — beyond the happy path. + """ + + name: str + + @abstractmethod + def evaluate( + self, + reports_dir: Path, + policy: GatePolicy, + ) -> GateResult: + """Evaluate behavioral test results. + + Args: + reports_dir: Path to reports/{submission-name}/ + policy: Gate policy to apply + + Returns: + Standardized GateResult + """ + ... + + def get_default_threshold(self) -> float: + """Get the gate's default pass threshold.""" + return 0.5 diff --git a/abevalflow/gates/behavioral/edge_case.py b/abevalflow/gates/behavioral/edge_case.py new file mode 100644 index 00000000..77b2c70b --- /dev/null +++ b/abevalflow/gates/behavioral/edge_case.py @@ -0,0 +1,117 @@ +"""Edge case behavioral gate. + +Evaluates how skills perform under edge case instructions +(empty input, adversarial input, boundary values, etc.). +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path + +from abevalflow.gates.base import GateMode, GateResult, GateType +from abevalflow.gates.behavioral import register_behavioral_gate +from abevalflow.gates.behavioral.base import BehavioralGate +from abevalflow.schemas import GatePolicy + +logger = logging.getLogger(__name__) + + +@register_behavioral_gate("edge-case") +class EdgeCaseGate(BehavioralGate): + """Gate that evaluates edge case test results. + + Reads edge case results from report.json and computes a score + based on the fraction of edge cases that passed verification. + """ + + name = "edge-case" + + def evaluate( + self, + reports_dir: Path, + policy: GatePolicy, + ) -> GateResult: + gate_policy = policy.get_gate_policy("behavioral") + + if gate_policy.mode == GateMode.DISABLED: + return GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name=self.name, + policy_key=self.name, + passed=True, + score=1.0, + mode=GateMode.DISABLED, + message="Edge case behavioral gate disabled", + ) + + report_path = reports_dir / "report.json" + if not report_path.exists(): + return GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name=self.name, + policy_key=self.name, + passed=True, + score=1.0, + mode=gate_policy.mode, + message="No report.json found (edge case testing skipped)", + details={"status": "not_found"}, + ) + + try: + report_data = json.loads(report_path.read_text()) + except (json.JSONDecodeError, OSError) as e: + logger.error("Failed to read report.json: %s", e) + return GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name=self.name, + policy_key=self.name, + passed=False, + score=0.0, + mode=gate_policy.mode, + message=f"Failed to parse report.json: {e}", + ) + + edge_case_results = report_data.get("edge_case_results", []) + if not edge_case_results: + return GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name=self.name, + policy_key=self.name, + passed=True, + score=1.0, + mode=gate_policy.mode, + message="No edge case results in report", + details={"status": "no_edge_cases"}, + ) + + total = len(edge_case_results) + passed_count = sum(1 for ec in edge_case_results if ec.get("passed", False)) + score = passed_count / total if total > 0 else 0.0 + + threshold = gate_policy.threshold or self.get_default_threshold() + passed = score >= threshold + + if gate_policy.mode == GateMode.WARN: + passed = True + + return GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name=self.name, + policy_key=self.name, + passed=passed, + score=score, + mode=gate_policy.mode, + threshold=threshold, + message=( + f"Edge cases: {passed_count}/{total} passed " + f"(score {score:.2f}, threshold {threshold})" + ), + details={ + "total": total, + "passed": passed_count, + "failed": total - passed_count, + "results": edge_case_results, + }, + ) diff --git a/abevalflow/harbor_agents/verifiers/llm_judge.py b/abevalflow/harbor_agents/verifiers/llm_judge.py index f709f086..eef026ed 100644 --- a/abevalflow/harbor_agents/verifiers/llm_judge.py +++ b/abevalflow/harbor_agents/verifiers/llm_judge.py @@ -64,6 +64,14 @@ def grade(response_text: str) -> dict: "completeness": "The response thoroughly covers all aspects of the question/task", "clarity": "The response is clear, well-organized, and easy to understand", "accuracy": "The response contains accurate technical information", + "failure_handling": ( + "The agent handles errors and failures gracefully, providing clear error messages " + "and recovery suggestions rather than crashing or producing garbage output" + ), + "uncertainty_acknowledgment": ( + "The agent acknowledges uncertainty when it doesn't know the answer, rather than " + "confabulating or presenting uncertain information as fact" + ), } diff --git a/abevalflow/report.py b/abevalflow/report.py index 083f5076..239f0759 100644 --- a/abevalflow/report.py +++ b/abevalflow/report.py @@ -162,6 +162,14 @@ class AnalysisSummary(BaseModel): recommendation: Recommendation +class EdgeCaseResult(BaseModel): + """Result of a single edge case evaluation.""" + + name: str = Field(description="Edge case name (derived from filename)") + summary: VariantSummary = Field(description="Trial stats for this edge case") + passed: bool = Field(description="Whether the edge case passed verification") + + class AnalysisResult(BaseModel): """Top-level report model written to report.json.""" @@ -179,3 +187,7 @@ class AnalysisResult(BaseModel): default=None, description="Historical degradation check result. Present when monitoring ran.", ) + edge_case_results: list[EdgeCaseResult] = Field( + default_factory=list, + description="Results from edge case evaluations. Empty if no edge cases defined.", + ) diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index 087a1e95..7803a9ff 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -29,6 +29,7 @@ profiles: certified: checks: - enterprise_structure_validation + - enterprise_behavioral_testing - advanced_agent_validation # MCP Servers (Standardized Tool APIs) - API contracts and stability @@ -85,6 +86,7 @@ profiles: certified: checks: - enterprise_structure_validation + - enterprise_behavioral_testing - advanced_agent_validation # - safety_toxicity_bias_guardrails # Not yet implemented @@ -111,9 +113,9 @@ profiles: checks: - enterprise_structure_validation - enterprise_security_review + - enterprise_behavioral_testing - advanced_agent_validation # Not yet implemented: - # - enterprise_behavioral_testing # - continuous_optimization # - safety_toxicity_bias_guardrails # - resilience_chaos_testing diff --git a/scripts/aggregate_scorecard.py b/scripts/aggregate_scorecard.py index 575dfd92..8527a04f 100644 --- a/scripts/aggregate_scorecard.py +++ b/scripts/aggregate_scorecard.py @@ -42,7 +42,8 @@ validate_push_facts_config, ) from abevalflow.engines import get_engine -from abevalflow.gates.base import GateResult +from abevalflow.gates.base import GateResult, GateType +from abevalflow.gates.behavioral import get_all_behavioral_gates 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 @@ -186,6 +187,53 @@ def _maybe_push_fact(gate_result: GateResult, policy: GatePolicy) -> FactPushRes ) +def _extract_behavioral_data( + reports_dir: Path, + gates: list[GateResult], +) -> dict | None: + """Extract behavioral testing data for certification. + + Assembles behavioral_data from two sources: + - std_reward: read from report.json (trial variance for consistency check) + - edge_cases: extracted from the EdgeCaseGate result in gates list + (single source of truth — avoids re-reading report.json) + + Stability and failure_mode sub-check data are not extracted here because + no pipeline component currently produces them. The sub-check functions + in certification.py are ready and will activate when upstream producers + exist. + TODO: Add stability extraction here once get_historical_variance() + in monitor.py is wired into the pipeline. + TODO: Add failure_mode extraction here once LLM judge trial details include + criteria_scores (failure_handling, uncertainty_acknowledgment) in report.json. + """ + behavioral_data: dict = {} + + # Extract trial variance from report.json + report_path = reports_dir / "report.json" + if report_path.exists(): + try: + report_data = json.loads(report_path.read_text()) + summary = report_data.get("summary", {}) + treatment = summary.get("treatment", {}) + std_reward = treatment.get("std_reward") + if std_reward is not None: + behavioral_data["std_reward"] = std_reward + except (json.JSONDecodeError, OSError) as e: + logger.warning("Failed to parse report.json for behavioral data: %s", e) + + # Extract edge case counts from the EdgeCaseGate result + for gate in gates: + if gate.gate_type == GateType.BEHAVIORAL and gate.details.get("total") is not None: + behavioral_data["edge_cases"] = { + "total": gate.details["total"], + "passed": gate.details.get("passed", 0), + } + break + + return behavioral_data if behavioral_data else None + + def aggregate_scorecard( submission_dir: Path, results_dir: Path, @@ -281,6 +329,24 @@ def aggregate_scorecard( fact_push_results.append(push_result) logger.info("Quality %s: passed=%s, score=%.3f", quality_gate.name, gate_result.passed, gate_result.score) + for behavioral_gate in get_all_behavioral_gates(): + if not policy.is_enabled(behavioral_gate.name): + logger.info("Behavioral gate %s is disabled, skipping", behavioral_gate.name) + continue + + logger.info("Processing behavioral gate: %s", behavioral_gate.name) + gate_result = behavioral_gate.evaluate(reports_dir, policy) + gates.append(gate_result) + push_result = _maybe_push_fact(gate_result, policy) + if push_result: + fact_push_results.append(push_result) + logger.info( + "Behavioral %s: passed=%s, score=%.3f", + behavioral_gate.name, + gate_result.passed, + gate_result.score, + ) + recommendation, reason = apply_combination_logic(gates, policy) logger.info("Final recommendation: %s (%s)", recommendation, reason) @@ -324,6 +390,9 @@ def aggregate_scorecard( operational_policy_result.score, ) + # Extract behavioral data from report + gate results + behavioral_data = _extract_behavioral_data(reports_dir, gates) + certification = compute_certification( gates=gates, validation_passed=validation_passed, @@ -331,6 +400,7 @@ def aggregate_scorecard( has_eval_assets=has_eval_assets, policy=certification_policy, operational_policy_result=operational_policy_result, + behavioral_data=behavioral_data, ) logger.info( "Certification levels: foundational=%s, trusted=%s, certified=%s (highest=%s)", diff --git a/scripts/analyze.py b/scripts/analyze.py index 8ac4f610..ddf9660b 100644 --- a/scripts/analyze.py +++ b/scripts/analyze.py @@ -41,6 +41,7 @@ AnalysisResult, AnalysisSummary, DegradationResult, + EdgeCaseResult, Provenance, Recommendation, ScanMode, @@ -225,6 +226,46 @@ def build_a2a_analysis( ) +# --------------------------------------------------------------------------- +# Edge case analysis +# --------------------------------------------------------------------------- + + +def analyze_edge_cases( + workspace_dir: Path, + pass_threshold: float = 0.5, +) -> list[EdgeCaseResult]: + """Analyze edge case evaluation results. + + Args: + workspace_dir: Root directory containing tasks-treatment-edge-*/ + directories produced by scaffold.py. + pass_threshold: Minimum mean reward for an edge case to pass. + """ + edge_results: list[EdgeCaseResult] = [] + + for edge_dir in sorted(workspace_dir.glob("tasks-treatment-edge-*")): + if not edge_dir.is_dir(): + continue + edge_name = edge_dir.name.removeprefix("tasks-treatment-edge-") + trials = parse_variant_trials(edge_dir) + if not trials: + continue + summary = compute_variant_summary(trials) + mean_r = summary.mean_reward or 0.0 + passed = mean_r >= pass_threshold + + edge_results.append( + EdgeCaseResult( + name=edge_name, + summary=summary, + passed=passed, + ) + ) + + return edge_results + + # --------------------------------------------------------------------------- # Statistics # --------------------------------------------------------------------------- @@ -331,8 +372,15 @@ def build_analysis( related_pr: str | None = None, llm_label: str | None = None, eval_engine: str | None = None, + edge_case_results_dir: Path | None = None, ) -> AnalysisResult: - """Parse results, compute stats, and assemble the full analysis model.""" + """Parse results, compute stats, and assemble the full analysis model. + + Args: + edge_case_results_dir: Root directory containing tasks-treatment-edge-*/ + result directories. When provided, edge case results are analyzed + and included in the report. + """ if eval_engine == "a2a" or is_a2a_results(results_dir): return build_a2a_analysis( results_dir=results_dir, @@ -378,6 +426,16 @@ def build_analysis( else: recommendation = Recommendation.PASS if primary_gap >= threshold else Recommendation.FAIL + edge_cases: list[EdgeCaseResult] = [] + if edge_case_results_dir is not None: + edge_cases = analyze_edge_cases(edge_case_results_dir) + if edge_cases: + logger.info( + "Edge cases: %d/%d passed", + sum(1 for ec in edge_cases if ec.passed), + len(edge_cases), + ) + return AnalysisResult( submission_name=submission_name, provenance=provenance or Provenance(), @@ -396,6 +454,7 @@ def build_analysis( "treatment": treatment_trials, "control": control_trials, }, + edge_case_results=edge_cases, ) @@ -637,6 +696,12 @@ def main(argv: list[str] | None = None) -> int: default=None, help="LLM model label for the report (e.g. 'Claude Sonnet 4.6 (vertex_ai)')", ) + parser.add_argument( + "--edge-case-results-dir", + type=Path, + default=None, + help="Root directory containing tasks-treatment-edge-*/ result directories", + ) args = parser.parse_args(argv) @@ -678,6 +743,7 @@ def main(argv: list[str] | None = None) -> int: related_pr=args.pr_url, llm_label=args.llm_label, eval_engine=args.eval_engine, + edge_case_results_dir=args.edge_case_results_dir, ) # Include security scan results if available diff --git a/scripts/monitor.py b/scripts/monitor.py index 3b56e8a7..d94e98ff 100644 --- a/scripts/monitor.py +++ b/scripts/monitor.py @@ -18,6 +18,7 @@ import logging import sys from dataclasses import asdict, dataclass +from statistics import mean, variance as stat_variance from typing import TYPE_CHECKING from sqlalchemy import create_engine, select @@ -48,6 +49,56 @@ class MonitorResult: previous_run_id: str | None = None +# TODO: Wire into aggregate_scorecard._extract_behavioral_data() +# to feed stability sub-check in behavioral testing certification. +def get_historical_variance( + engine: Engine, + submission_name: str, + window: int = 5, + eval_engine: str | None = None, +) -> dict: + """Query last N evaluation runs and compute score variance. + + Args: + engine: SQLAlchemy engine connected to the evaluation DB. + submission_name: The skill/submission name to check. + window: Number of recent runs to consider. + eval_engine: Optional engine filter. + + Returns: + Dict with score_variance, run_count, scores, mean_score. + """ + from abevalflow.db.models import EvaluationRun + + with Session(engine) as session: + stmt = select(EvaluationRun).where(EvaluationRun.submission_name == submission_name) + if eval_engine is not None: + stmt = stmt.where(EvaluationRun.eval_engine == eval_engine) + stmt = stmt.order_by(EvaluationRun.created_at.desc()).limit(window) + runs = list(session.execute(stmt).scalars().all()) + + scores = [r.treatment_pass_rate for r in runs if r.treatment_pass_rate is not None] + run_count = len(scores) + + if run_count < 2: + return { + "score_variance": 0.0, + "run_count": run_count, + "scores": scores, + "mean_score": scores[0] if scores else None, + } + + mean_score = mean(scores) + variance = stat_variance(scores) + + return { + "score_variance": variance, + "run_count": run_count, + "scores": scores, + "mean_score": mean_score, + } + + def check_degradation( engine: Engine, submission_name: str, diff --git a/scripts/scaffold.py b/scripts/scaffold.py index b9889a64..12b4a5a5 100644 --- a/scripts/scaffold.py +++ b/scripts/scaffold.py @@ -187,6 +187,83 @@ def _warn_non_claude_skills_dir( ) +def _write_rendered_templates( + rendered: dict[str, str], + target_dir: Path, +) -> Path: + """Write rendered templates to a task directory, return the environment dir.""" + target_dir.mkdir(parents=True, exist_ok=True) + environment_dir = target_dir / "environment" + environment_dir.mkdir(exist_ok=True) + + for filename, content in rendered.items(): + if filename == "Dockerfile": + dest = environment_dir / filename + elif filename == "test.sh": + tests_dir = target_dir / "tests" + tests_dir.mkdir(exist_ok=True) + dest = tests_dir / filename + else: + dest = target_dir / filename + dest.write_text(content) + if filename == "test.sh": + dest.chmod(dest.stat().st_mode | stat.S_IEXEC) + + return environment_dir + + +def _copy_root_dirs(submission_dir: Path, target_dir: Path) -> None: + """Copy solution/ and tests/ to task root for Harbor's emptyDir mounts.""" + for rootdir in ("solution", "tests"): + src = submission_dir / rootdir + if src.is_dir(): + shutil.copytree(src, target_dir / rootdir, dirs_exist_ok=True) + + +def _scaffold_edge_cases( + submission_dir: Path, + output_dir: Path, + metadata: SubmissionMetadata, + strategy: ExperimentStrategy, + jinja_env: Environment, +) -> list[Path]: + """Scaffold per-edge-case treatment task directories. + + For each .md file in edge_cases/, creates a task directory identical to + the treatment variant but with the edge case instruction instead of the + main instruction.md. No control variant is needed for edge cases. + + Returns list of edge case task directory paths. + """ + edge_cases_dir = submission_dir / "edge_cases" + if not edge_cases_dir.is_dir(): + return [] + + context = _build_template_context(metadata, submission_dir, "treatment", strategy) + rendered = _render_templates(jinja_env, context) + strategy_srcs = [src for src, _ in context.get("copy_pairs", [])] + + # Same templates for all edge cases; only instruction.md differs per case + edge_case_dirs: list[Path] = [] + for md_file in sorted(edge_cases_dir.glob("*.md")): + edge_name = md_file.stem + edge_dir = output_dir / f"tasks-treatment-edge-{edge_name}" / metadata.name + + environment_dir = _write_rendered_templates(rendered, edge_dir) + _copy_submission_files(submission_dir, environment_dir, strategy_srcs, "treatment") + + # Two copies: environment/ for the Docker build context, + # task root for Harbor metadata/display (same as main scaffolding). + shutil.copy2(md_file, environment_dir / "instruction.md") + shutil.copy2(md_file, edge_dir / "instruction.md") + _copy_root_dirs(submission_dir, edge_dir) + + edge_case_dirs.append(edge_dir) + logger.info("Scaffolded edge case '%s' at %s", edge_name, edge_dir) + + return edge_case_dirs + + def scaffold_submission( submission_dir: Path, output_dir: Path, @@ -228,24 +305,7 @@ def scaffold_submission( context["has_claude_md"] = False context["has_mcp_json"] = False rendered = _render_templates(jinja_env, context) - - target_dir.mkdir(parents=True, exist_ok=True) - - environment_dir = target_dir / "environment" - environment_dir.mkdir(exist_ok=True) - - for filename, content in rendered.items(): - if filename == "Dockerfile": - dest = environment_dir / filename - elif filename == "test.sh": - tests_dir = target_dir / "tests" - tests_dir.mkdir(exist_ok=True) - dest = tests_dir / filename - else: - dest = target_dir / filename - dest.write_text(content) - if filename == "test.sh": - dest.chmod(dest.stat().st_mode | stat.S_IEXEC) + environment_dir = _write_rendered_templates(rendered, target_dir) strategy_srcs = [src for src, _ in context.get("copy_pairs", [])] _copy_submission_files(submission_dir, environment_dir, strategy_srcs, variant) @@ -253,18 +313,16 @@ def scaffold_submission( # Second copy at task root: Harbor reads instruction.md from the task # directory (outside the build context) for display/metadata purposes. shutil.copy2(submission_dir / "instruction.md", target_dir / "instruction.md") - - # Copy solution/ and tests/ to task root: the OpenShift backend - # mounts emptyDir volumes over /tests and /solution inside the pod, - # hiding anything the Dockerfile COPY'd. Harbor uploads these - # directories from the task root into the pod at runtime. - for rootdir in ("solution", "tests"): - src = submission_dir / rootdir - if src.is_dir(): - shutil.copytree(src, target_dir / rootdir, dirs_exist_ok=True) + _copy_root_dirs(submission_dir, target_dir) logger.info("Scaffolded %s variant at %s", variant, target_dir) + edge_case_dirs = _scaffold_edge_cases( + submission_dir, output_dir, metadata, strategy, jinja_env + ) + if edge_case_dirs: + logger.info("Scaffolded %d edge case(s)", len(edge_case_dirs)) + return treatment_dir, control_dir diff --git a/scripts/validate.py b/scripts/validate.py index 843a3000..a6f2add9 100644 --- a/scripts/validate.py +++ b/scripts/validate.py @@ -129,6 +129,34 @@ def _check_supportive_size(submission_dir: Path) -> list[str]: return [] +def _check_edge_cases(submission_dir: Path) -> list[str]: + """Validate optional edge_cases/ directory structure. + + Each file must be a non-empty .md file with a descriptive name. + The directory is optional — submissions without it pass validation. + """ + edge_cases_dir = submission_dir / "edge_cases" + if not edge_cases_dir.is_dir(): + return [] + + errors: list[str] = [] + md_files = list(edge_cases_dir.glob("*.md")) + + if not md_files: + errors.append("edge_cases/ directory exists but contains no .md files") + return errors + + for md_file in sorted(md_files): + if not md_file.read_text().strip(): + errors.append(f"edge_cases/{md_file.name} is empty") + + non_md = [f.name for f in edge_cases_dir.iterdir() if f.is_file() and f.suffix != ".md"] + if non_md: + errors.append(f"edge_cases/ contains non-.md files: {', '.join(non_md)}") + + return errors + + # --------------------------------------------------------------------------- # ASE-specific checks # --------------------------------------------------------------------------- @@ -332,6 +360,7 @@ def validate_submission( llm_judge = submission_dir / "tests" / "llm_judge.py" if llm_judge.is_file(): errors.extend(_check_py_compiles(llm_judge)) + errors.extend(_check_edge_cases(submission_dir)) if run_ase: errors.extend(_check_skill_md_frontmatter(submission_dir)) diff --git a/tests/test_behavioral_checks.py b/tests/test_behavioral_checks.py new file mode 100644 index 00000000..7dd1d9c5 --- /dev/null +++ b/tests/test_behavioral_checks.py @@ -0,0 +1,753 @@ +"""Tests for behavioral testing certification checks.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from abevalflow.certification import ( + CheckId, + _check_consistency, + _check_edge_case_results, + _check_failure_mode, + _check_stability, + _compute_behavioral_testing_check, + compute_certification, +) +from abevalflow.gates.base import GateMode, GateResult, GateType +from abevalflow.gates.behavioral.edge_case import EdgeCaseGate +from abevalflow.harbor_agents.verifiers.llm_judge import CRITERIA_DESCRIPTIONS +from abevalflow.report import EdgeCaseResult, VariantSummary +from abevalflow.schemas import GatePolicy, GatePolicyItem + + +class TestConsistencyCheck: + """Tests for _check_consistency() — trial variance detection.""" + + def test_low_variance_passes(self) -> None: + result = _check_consistency({"std_reward": 0.1}) + assert result.passed is True + assert result.score > 0.5 + + def test_high_variance_fails(self) -> None: + result = _check_consistency({"std_reward": 0.5}) + assert result.passed is False + assert result.score < 0.5 + + def test_zero_variance_passes(self) -> None: + result = _check_consistency({"std_reward": 0.0}) + assert result.passed is True + assert result.score == 1.0 + + def test_at_threshold_score_is_half(self) -> None: + result = _check_consistency({"std_reward": 0.3}) + assert result.passed is True + assert result.score == pytest.approx(0.5) + + def test_none_variance_passes(self) -> None: + result = _check_consistency({"std_reward": None}) + assert result.passed is True + assert result.score == 1.0 + + def test_missing_variance_passes(self) -> None: + result = _check_consistency({}) + assert result.passed is True + assert result.score == 1.0 + + def test_custom_threshold(self) -> None: + result = _check_consistency({"std_reward": 0.15}, threshold=0.1) + assert result.passed is False + + result = _check_consistency({"std_reward": 0.05}, threshold=0.1) + assert result.passed is True + + def test_score_clamped(self) -> None: + result = _check_consistency({"std_reward": 1.0}) + assert 0.0 <= result.score <= 1.0 + + def test_details_contain_threshold_and_status(self) -> None: + result = _check_consistency({"std_reward": 0.2}) + assert result.details["std_reward"] == 0.2 + assert "status" not in result.details + + def test_no_data_has_status_sentinel(self) -> None: + result = _check_consistency({}) + assert result.details["status"] == "no_data" + + +class TestStabilityCheck: + """Tests for _check_stability() — score drift detection.""" + + def test_stable_scores_pass(self) -> None: + result = _check_stability({ + "stability": {"score_variance": 0.01, "run_count": 5}, + }) + assert result.passed is True + assert result.score > 0.5 + + def test_drifting_scores_fail(self) -> None: + result = _check_stability({ + "stability": {"score_variance": 0.2, "run_count": 5}, + }) + assert result.passed is False + assert result.score < 0.5 + + def test_insufficient_history_passes(self) -> None: + result = _check_stability({ + "stability": {"score_variance": 0.0, "run_count": 2}, + }) + assert result.passed is True + assert "Insufficient history" in result.message + + def test_no_stability_data_passes(self) -> None: + result = _check_stability({}) + assert result.passed is True + assert result.score == 1.0 + + def test_none_stability_passes(self) -> None: + result = _check_stability({"stability": None}) + assert result.passed is True + + def test_at_threshold_score_is_half(self) -> None: + result = _check_stability({ + "stability": {"score_variance": 0.1, "run_count": 5}, + }) + assert result.passed is True + assert result.score == pytest.approx(0.5) + + def test_no_data_has_status_sentinel(self) -> None: + result = _check_stability({}) + assert result.details["status"] == "no_data" + + def test_insufficient_history_has_status_sentinel(self) -> None: + result = _check_stability({ + "stability": {"score_variance": 0.0, "run_count": 1}, + }) + assert result.details["status"] == "no_data" + + +class TestEdgeCaseCheck: + """Tests for _check_edge_case_results() — edge case pass rate.""" + + def test_all_passed(self) -> None: + result = _check_edge_case_results({ + "edge_cases": {"total": 3, "passed": 3}, + }) + assert result.passed is True + assert result.score == 1.0 + + def test_some_failed(self) -> None: + result = _check_edge_case_results({ + "edge_cases": {"total": 4, "passed": 1}, + }) + assert result.passed is False + assert result.score == 0.25 + + def test_all_failed(self) -> None: + result = _check_edge_case_results({ + "edge_cases": {"total": 3, "passed": 0}, + }) + assert result.passed is False + assert result.score == 0.0 + + def test_no_edge_cases_defined(self) -> None: + result = _check_edge_case_results({ + "edge_cases": {"total": 0, "passed": 0}, + }) + assert result.passed is True + assert result.score == 1.0 + + def test_no_edge_case_data(self) -> None: + result = _check_edge_case_results({}) + assert result.passed is True + assert result.score == 1.0 + + def test_details_contain_counts(self) -> None: + result = _check_edge_case_results({ + "edge_cases": {"total": 5, "passed": 3}, + }) + assert result.details["total"] == 5 + assert result.details["passed"] == 3 + assert result.details["pass_rate"] == 0.6 + + +class TestFailureModeCheck: + """Tests for _check_failure_mode() — failure handling scores.""" + + def test_good_score_passes(self) -> None: + result = _check_failure_mode({ + "failure_mode": {"score": 0.8, "threshold": 0.5}, + }) + assert result.passed is True + assert result.score == 0.8 + + def test_low_score_fails(self) -> None: + result = _check_failure_mode({ + "failure_mode": {"score": 0.3, "threshold": 0.5}, + }) + assert result.passed is False + assert result.score == 0.3 + + def test_no_data_passes(self) -> None: + result = _check_failure_mode({}) + assert result.passed is True + assert result.score == 1.0 + + +class TestBehavioralTestingCombined: + """Tests for _compute_behavioral_testing_check() — combined check.""" + + def test_all_sub_checks_pass(self) -> None: + data = { + "std_reward": 0.1, + "edge_cases": {"total": 3, "passed": 3}, + "stability": {"score_variance": 0.01, "run_count": 5}, + "failure_mode": {"score": 0.8, "threshold": 0.5}, + } + result = _compute_behavioral_testing_check(data) + assert result.passed is True + assert result.score > 0.6 + + def test_consistency_fails_means_overall_fails(self) -> None: + data = { + "std_reward": 0.9, + "edge_cases": {"total": 3, "passed": 3}, + "stability": {"score_variance": 0.01, "run_count": 5}, + "failure_mode": {"score": 0.8, "threshold": 0.5}, + } + result = _compute_behavioral_testing_check(data) + assert result.passed is False + + def test_single_sub_check_insufficient(self) -> None: + data = {"std_reward": 0.1} + result = _compute_behavioral_testing_check(data) + assert result.passed is False + assert "Insufficient behavioral coverage" in result.message + + def test_two_sub_checks_sufficient(self) -> None: + data = { + "std_reward": 0.1, + "edge_cases": {"total": 3, "passed": 3}, + } + result = _compute_behavioral_testing_check(data) + assert result.passed is True + assert result.score > 0.0 + + def test_no_data_fails(self) -> None: + result = _compute_behavioral_testing_check({}) + assert result.passed is False + assert result.score == 0.0 + assert "No behavioral testing data available" in result.message + + def test_details_contain_sub_checks(self) -> None: + data = { + "std_reward": 0.1, + "edge_cases": {"total": 3, "passed": 3}, + } + result = _compute_behavioral_testing_check(data) + assert "sub_checks" in result.details + assert "weights" in result.details + + def test_at_threshold_sub_checks_still_pass_composite(self) -> None: + data = { + "std_reward": 0.3, + "edge_cases": {"total": 2, "passed": 1}, + } + result = _compute_behavioral_testing_check(data) + assert result.passed is True + assert result.score == pytest.approx(0.5) + + +class TestCertificationWithBehavioralData: + """Tests for compute_certification() with behavioral_data parameter.""" + + def _make_all_gates(self) -> list[GateResult]: + return [ + GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + ), + GateResult( + gate_name="security", + gate_type=GateType.SECURITY, + passed=True, + score=1.0, + mode=GateMode.BLOCK, + ), + GateResult( + gate_name="quality", + gate_type=GateType.QUALITY, + passed=True, + score=0.85, + mode=GateMode.WARN, + ), + ] + + def test_behavioral_data_enables_certified_check(self) -> None: + behavioral_data = { + "std_reward": 0.1, + "edge_cases": {"total": 3, "passed": 3}, + } + result = compute_certification( + gates=self._make_all_gates(), + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + behavioral_data=behavioral_data, + ) + behavioral_check = next( + (c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING), + None, + ) + assert behavioral_check is not None + assert behavioral_check.passed is True + + def test_no_behavioral_data_check_fails(self) -> None: + result = compute_certification( + gates=self._make_all_gates(), + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + ) + behavioral_check = next( + (c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING), + None, + ) + assert behavioral_check is not None + assert behavioral_check.passed is False + + def test_behavioral_check_in_certified_checks(self) -> None: + from abevalflow.certification import CERTIFIED_CHECKS + assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS + + def test_backward_compatible_no_behavioral_data(self) -> None: + result = compute_certification( + gates=self._make_all_gates(), + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + ) + assert result.foundational.passed is True + assert result.trusted.passed is True + + +class TestEdgeCaseGate: + """Tests for EdgeCaseGate behavioral gate.""" + + def test_no_report_json(self, tmp_path: Path) -> None: + policy = GatePolicy() + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert result.gate_type == GateType.BEHAVIORAL + + def test_disabled_mode(self, tmp_path: Path) -> None: + policy = GatePolicy(gates={"behavioral": GatePolicyItem(mode=GateMode.DISABLED)}) + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert result.mode == GateMode.DISABLED + + def test_all_edge_cases_pass(self, tmp_path: Path) -> None: + report = { + "edge_case_results": [ + {"name": "empty_input", "passed": True}, + {"name": "adversarial", "passed": True}, + ], + } + (tmp_path / "report.json").write_text(json.dumps(report)) + + policy = GatePolicy(gates={"behavioral": GatePolicyItem(mode=GateMode.BLOCK)}) + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert result.score == 1.0 + + def test_some_edge_cases_fail(self, tmp_path: Path) -> None: + report = { + "edge_case_results": [ + {"name": "empty_input", "passed": True}, + {"name": "adversarial", "passed": False}, + {"name": "boundary", "passed": False}, + ], + } + (tmp_path / "report.json").write_text(json.dumps(report)) + + policy = GatePolicy(gates={"behavioral": GatePolicyItem(mode=GateMode.BLOCK, threshold=0.5)}) + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is False + assert abs(result.score - 1 / 3) < 0.01 + + def test_no_edge_case_results_in_report(self, tmp_path: Path) -> None: + report = {"summary": {"treatment": {}}} + (tmp_path / "report.json").write_text(json.dumps(report)) + + policy = GatePolicy() + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + + def test_warn_mode_always_passes(self, tmp_path: Path) -> None: + report = { + "edge_case_results": [ + {"name": "test", "passed": False}, + ], + } + (tmp_path / "report.json").write_text(json.dumps(report)) + + policy = GatePolicy(gates={"behavioral": GatePolicyItem(mode=GateMode.WARN)}) + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert result.score == 0.0 + + +class TestEdgeCaseValidation: + """Tests for edge case directory validation.""" + + def test_no_edge_cases_dir_passes(self, tmp_path: Path) -> None: + from scripts.validate import _check_edge_cases + errors = _check_edge_cases(tmp_path) + assert errors == [] + + def test_valid_edge_cases(self, tmp_path: Path) -> None: + from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" + edge_dir.mkdir() + (edge_dir / "empty_input.md").write_text("Test with empty input") + (edge_dir / "adversarial.md").write_text("Test with adversarial prompt") + errors = _check_edge_cases(tmp_path) + assert errors == [] + + def test_empty_md_file(self, tmp_path: Path) -> None: + from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" + edge_dir.mkdir() + (edge_dir / "empty_input.md").write_text("") + errors = _check_edge_cases(tmp_path) + assert len(errors) == 1 + assert "empty" in errors[0] + + def test_non_md_files(self, tmp_path: Path) -> None: + from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" + edge_dir.mkdir() + (edge_dir / "valid.md").write_text("Valid content") + (edge_dir / "not_md.txt").write_text("Wrong extension") + errors = _check_edge_cases(tmp_path) + assert any("non-.md" in e for e in errors) + + def test_empty_dir(self, tmp_path: Path) -> None: + from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" + edge_dir.mkdir() + errors = _check_edge_cases(tmp_path) + assert any("no .md files" in e for e in errors) + + +class TestExtractBehavioralData: + """Tests for _extract_behavioral_data() glue function.""" + + def test_extracts_std_reward(self, tmp_path: Path) -> None: + from scripts.aggregate_scorecard import _extract_behavioral_data + report = {"summary": {"treatment": {"std_reward": 0.15}}} + (tmp_path / "report.json").write_text(json.dumps(report)) + data = _extract_behavioral_data(tmp_path, gates=[]) + assert data is not None + assert data["std_reward"] == 0.15 + + def test_extracts_edge_cases_from_gate(self, tmp_path: Path) -> None: + from scripts.aggregate_scorecard import _extract_behavioral_data + gate = GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name="edge-case", + passed=True, + score=0.5, + details={"total": 4, "passed": 2}, + ) + data = _extract_behavioral_data(tmp_path, gates=[gate]) + assert data is not None + assert data["edge_cases"] == {"total": 4, "passed": 2} + + def test_returns_none_for_empty_report(self, tmp_path: Path) -> None: + from scripts.aggregate_scorecard import _extract_behavioral_data + report = {"summary": {"treatment": {}}} + (tmp_path / "report.json").write_text(json.dumps(report)) + data = _extract_behavioral_data(tmp_path, gates=[]) + assert data is None + + def test_returns_none_when_no_report_and_no_gates(self, tmp_path: Path) -> None: + from scripts.aggregate_scorecard import _extract_behavioral_data + data = _extract_behavioral_data(tmp_path, gates=[]) + assert data is None + + def test_combines_std_reward_and_gate_edge_cases(self, tmp_path: Path) -> None: + from scripts.aggregate_scorecard import _extract_behavioral_data + report = {"summary": {"treatment": {"std_reward": 0.2}}} + (tmp_path / "report.json").write_text(json.dumps(report)) + gate = GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name="edge-case", + passed=True, + score=1.0, + details={"total": 3, "passed": 3}, + ) + data = _extract_behavioral_data(tmp_path, gates=[gate]) + assert data is not None + assert data["std_reward"] == 0.2 + assert data["edge_cases"] == {"total": 3, "passed": 3} + + +class TestEdgeCaseScaffolding: + """Tests for edge case scaffolding.""" + + def _make_submission(self, tmp_path: Path) -> Path: + sub = tmp_path / "my-skill" + sub.mkdir() + (sub / "metadata.yaml").write_text("name: my-skill\n") + (sub / "instruction.md").write_text("Main instruction") + skills = sub / "skills" + skills.mkdir() + (skills / "SKILL.md").write_text("# My Skill") + tests = sub / "tests" + tests.mkdir() + (tests / "test_outputs.py").write_text("pass") + + edge_cases = sub / "edge_cases" + edge_cases.mkdir() + (edge_cases / "empty_input.md").write_text("Test with empty input") + (edge_cases / "adversarial.md").write_text("Test with adversarial prompt") + return sub + + def test_edge_case_dirs_created(self, tmp_path: Path) -> None: + from scripts.scaffold import scaffold_submission + sub = self._make_submission(tmp_path) + output = tmp_path / "output" + scaffold_submission(sub, output) + + assert (output / "tasks-treatment-edge-adversarial" / "my-skill").is_dir() + assert (output / "tasks-treatment-edge-empty_input" / "my-skill").is_dir() + + def test_edge_case_uses_alternative_instruction(self, tmp_path: Path) -> None: + from scripts.scaffold import scaffold_submission + sub = self._make_submission(tmp_path) + output = tmp_path / "output" + scaffold_submission(sub, output) + + edge_dir = output / "tasks-treatment-edge-empty_input" / "my-skill" + instruction = (edge_dir / "instruction.md").read_text() + assert instruction == "Test with empty input" + + def test_no_edge_cases_no_extra_dirs(self, tmp_path: Path) -> None: + from scripts.scaffold import scaffold_submission + sub = tmp_path / "no-edge" + sub.mkdir() + (sub / "metadata.yaml").write_text("name: no-edge\n") + (sub / "instruction.md").write_text("Main instruction") + skills = sub / "skills" + skills.mkdir() + (skills / "SKILL.md").write_text("# My Skill") + tests = sub / "tests" + tests.mkdir() + (tests / "test_outputs.py").write_text("pass") + + output = tmp_path / "output" + scaffold_submission(sub, output) + + edge_dirs = list(output.glob("tasks-treatment-edge-*")) + assert edge_dirs == [] + + +class TestEdgeCaseAnalysis: + """Tests for edge case analysis in analyze.py.""" + + def test_edge_case_result_model(self) -> None: + result = EdgeCaseResult( + name="empty_input", + summary=VariantSummary(n_trials=5, n_passed=4, pass_rate=0.8, mean_reward=0.7), + passed=True, + ) + assert result.name == "empty_input" + assert result.passed is True + assert result.summary.pass_rate == 0.8 + + def test_build_analysis_includes_edge_cases(self, tmp_path: Path) -> None: + from scripts.analyze import build_analysis + + results = tmp_path / "results" + treatment = results / "treatment" / "job" / "task__001" + treatment.mkdir(parents=True) + (treatment / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": 0.8}, + })) + control = results / "control" / "job" / "task__001" + control.mkdir(parents=True) + (control / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": 0.3}, + })) + + edge_root = tmp_path / "edge_results" + edge_dir = edge_root / "tasks-treatment-edge-empty_input" / "job" / "task__001" + edge_dir.mkdir(parents=True) + (edge_dir / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": 0.6}, + })) + + result = build_analysis( + results_dir=results, + submission_name="test-skill", + edge_case_results_dir=edge_root, + ) + assert len(result.edge_case_results) == 1 + assert result.edge_case_results[0].name == "empty_input" + assert result.edge_case_results[0].passed is True + + def test_build_analysis_no_edge_cases(self, tmp_path: Path) -> None: + from scripts.analyze import build_analysis + + results = tmp_path / "results" + treatment = results / "treatment" / "job" / "task__001" + treatment.mkdir(parents=True) + (treatment / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": 0.8}, + })) + control = results / "control" / "job" / "task__001" + control.mkdir(parents=True) + (control / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": 0.3}, + })) + + result = build_analysis( + results_dir=results, + submission_name="test-skill", + ) + assert result.edge_case_results == [] + + +class TestLLMJudgeCriteria: + """Tests for failure mode LLM judge criteria.""" + + def test_failure_handling_criterion_exists(self) -> None: + assert "failure_handling" in CRITERIA_DESCRIPTIONS + assert "error" in CRITERIA_DESCRIPTIONS["failure_handling"].lower() + + def test_uncertainty_acknowledgment_criterion_exists(self) -> None: + assert "uncertainty_acknowledgment" in CRITERIA_DESCRIPTIONS + assert "uncertainty" in CRITERIA_DESCRIPTIONS["uncertainty_acknowledgment"].lower() + + def test_all_criteria_have_descriptions(self) -> None: + for name, desc in CRITERIA_DESCRIPTIONS.items(): + assert isinstance(desc, str), f"Criterion {name} has non-string description" + assert len(desc) > 10, f"Criterion {name} has too short description" + + +class TestEdgeCaseEndToEnd: + """End-to-end: scaffold → mock results → analyze → certification.""" + + def test_scaffold_analyze_certify(self, tmp_path: Path) -> None: + from scripts.analyze import build_analysis + from scripts.scaffold import scaffold_submission + + # 1. Create submission with edge cases + sub = tmp_path / "e2e-skill" + sub.mkdir() + (sub / "metadata.yaml").write_text("name: e2e-skill\n") + (sub / "instruction.md").write_text("Main instruction") + skills = sub / "skills" + skills.mkdir() + (skills / "SKILL.md").write_text("# E2E Skill") + tests = sub / "tests" + tests.mkdir() + (tests / "test_outputs.py").write_text("pass") + edge_cases = sub / "edge_cases" + edge_cases.mkdir() + (edge_cases / "empty_input.md").write_text("Test with empty input") + (edge_cases / "adversarial.md").write_text("Test with adversarial prompt") + + # 2. Scaffold + output = tmp_path / "scaffold_output" + scaffold_submission(sub, output) + assert (output / "tasks-treatment-edge-adversarial" / "e2e-skill").is_dir() + assert (output / "tasks-treatment-edge-empty_input" / "e2e-skill").is_dir() + + # 3. Create mock Harbor results (2+ trials for std_reward computation) + results = tmp_path / "results" + for variant in ("treatment", "control"): + for i, reward in enumerate([0.8, 0.85] if variant == "treatment" else [0.3, 0.35]): + trial_dir = results / variant / "job" / f"task__{i:03d}" + trial_dir.mkdir(parents=True) + (trial_dir / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": reward}, + })) + + # 4. Create mock results for edge cases + edge_results = tmp_path / "edge_results" + for edge_name, reward in [("adversarial", 0.7), ("empty_input", 0.6)]: + trial_dir = edge_results / f"tasks-treatment-edge-{edge_name}" / "job" / "task__001" + trial_dir.mkdir(parents=True) + (trial_dir / "result.json").write_text(json.dumps({ + "verifier_result": {"reward": reward}, + })) + + # 5. Analyze with edge case results + analysis = build_analysis( + results_dir=results, + submission_name="e2e-skill", + edge_case_results_dir=edge_results, + ) + assert len(analysis.edge_case_results) == 2 + assert all(ec.passed for ec in analysis.edge_case_results) + + # 6. Simulate the unified data path: std_reward from report + edge cases from gate + behavioral_data = { + "std_reward": analysis.summary.treatment.std_reward, + "edge_cases": { + "total": len(analysis.edge_case_results), + "passed": sum(1 for ec in analysis.edge_case_results if ec.passed), + }, + } + + # 7. Verify certification sees it + result = compute_certification( + gates=[ + GateResult(gate_name="evaluation", gate_type=GateType.ENGINE, passed=True, score=0.9, mode=GateMode.BLOCK), + GateResult(gate_name="security", gate_type=GateType.SECURITY, passed=True, score=1.0, mode=GateMode.BLOCK), + GateResult(gate_name="quality", gate_type=GateType.QUALITY, passed=True, score=0.8, mode=GateMode.WARN), + ], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + behavioral_data=behavioral_data, + ) + behavioral_check = next( + c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING + ) + assert behavioral_check.passed is True + assert behavioral_check.score > 0.0 + + +class TestBehavioralGateRegistry: + """Tests for behavioral gate registry.""" + + def test_edge_case_gate_registered(self) -> None: + from abevalflow.gates.behavioral import get_all_behavioral_gates, get_behavioral_gate + gate = get_behavioral_gate("edge-case") + assert isinstance(gate, EdgeCaseGate) + + all_gates = get_all_behavioral_gates() + assert len(all_gates) >= 1 + assert any(isinstance(g, EdgeCaseGate) for g in all_gates) + + def test_unknown_gate_raises(self) -> None: + from abevalflow.gates.behavioral import get_behavioral_gate + with pytest.raises(KeyError, match="Unknown behavioral gate"): + get_behavioral_gate("nonexistent") + + def test_gate_type_is_behavioral(self) -> None: + assert GateType.BEHAVIORAL == "behavioral" diff --git a/tests/test_certification.py b/tests/test_certification.py index 471cd761..af517f98 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -353,9 +353,10 @@ def test_trusted_has_expected_checks(self) -> None: def test_certified_has_expected_checks(self) -> None: assert CheckId.ENTERPRISE_STRUCTURE_VALIDATION in CERTIFIED_CHECKS assert CheckId.ENTERPRISE_SECURITY_REVIEW in CERTIFIED_CHECKS + assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS assert CheckId.ADVANCED_AGENT_VALIDATION in CERTIFIED_CHECKS - # behavioral_testing and continuous_optimization not yet implemented - assert len(CERTIFIED_CHECKS) == 3 + # continuous_optimization not yet implemented + assert len(CERTIFIED_CHECKS) == 4 class TestCertificationPolicy: From e4d5f4018f4287c6bc8927af9036b5a24682f65c Mon Sep 17 00:00:00 2001 From: csoceanu Date: Mon, 13 Jul 2026 12:45:33 +0300 Subject: [PATCH 02/10] fix: ruff lint/format + drop composite score threshold from pass condition - Fix import ordering in monitor.py (ruff I001) - Fix line length in test_behavioral_checks.py (ruff E501) - Run ruff format on certification.py, edge_case.py, scaffold.py - Drop composite score threshold from behavioral testing pass condition: all_passed is the sole gate, score is for reporting only - Clamp scores with min(1.0, ...) to prevent > 1.0 on negative inputs - Remove unused threshold_overrides param from _compute_behavioral_testing_check Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/certification.py | 12 +- abevalflow/gates/behavioral/edge_case.py | 5 +- scripts/monitor.py | 3 +- scripts/scaffold.py | 4 +- tests/test_behavioral_checks.py | 206 ++++++++++++++++------- 5 files changed, 153 insertions(+), 77 deletions(-) diff --git a/abevalflow/certification.py b/abevalflow/certification.py index 12633447..7511bf87 100644 --- a/abevalflow/certification.py +++ b/abevalflow/certification.py @@ -589,8 +589,7 @@ def _compute_behavioral_testing_check( passed=False, score=final_score, message=( - f"Insufficient behavioral coverage: {n_active} sub-check(s) " - f"with data, need at least {min_required}" + f"Insufficient behavioral coverage: {n_active} sub-check(s) with data, need at least {min_required}" ), details={"sub_checks": sub_results, "weights": BEHAVIORAL_SUB_CHECK_WEIGHTS}, ) @@ -603,10 +602,7 @@ def _compute_behavioral_testing_check( if passed: message = f"Behavioral testing passed ({n_active} sub-checks, score {final_score:.2f})" else: - message = ( - f"Behavioral testing failed: " - f"failed sub-checks: {', '.join(failed_checks)}" - ) + message = f"Behavioral testing failed: failed sub-checks: {', '.join(failed_checks)}" return CheckResult( check_id=CheckId.ENTERPRISE_BEHAVIORAL_TESTING, @@ -908,9 +904,7 @@ def compute_certification( ) if behavioral_data is not None: - all_checks[CheckId.ENTERPRISE_BEHAVIORAL_TESTING] = _compute_behavioral_testing_check( - behavioral_data - ) + all_checks[CheckId.ENTERPRISE_BEHAVIORAL_TESTING] = _compute_behavioral_testing_check(behavioral_data) else: all_checks.setdefault( CheckId.ENTERPRISE_BEHAVIORAL_TESTING, diff --git a/abevalflow/gates/behavioral/edge_case.py b/abevalflow/gates/behavioral/edge_case.py index 77b2c70b..89991a1f 100644 --- a/abevalflow/gates/behavioral/edge_case.py +++ b/abevalflow/gates/behavioral/edge_case.py @@ -104,10 +104,7 @@ def evaluate( score=score, mode=gate_policy.mode, threshold=threshold, - message=( - f"Edge cases: {passed_count}/{total} passed " - f"(score {score:.2f}, threshold {threshold})" - ), + message=(f"Edge cases: {passed_count}/{total} passed (score {score:.2f}, threshold {threshold})"), details={ "total": total, "passed": passed_count, diff --git a/scripts/monitor.py b/scripts/monitor.py index d94e98ff..54e26f3a 100644 --- a/scripts/monitor.py +++ b/scripts/monitor.py @@ -18,7 +18,8 @@ import logging import sys from dataclasses import asdict, dataclass -from statistics import mean, variance as stat_variance +from statistics import mean +from statistics import variance as stat_variance from typing import TYPE_CHECKING from sqlalchemy import create_engine, select diff --git a/scripts/scaffold.py b/scripts/scaffold.py index 12b4a5a5..3f7c73b3 100644 --- a/scripts/scaffold.py +++ b/scripts/scaffold.py @@ -317,9 +317,7 @@ def scaffold_submission( logger.info("Scaffolded %s variant at %s", variant, target_dir) - edge_case_dirs = _scaffold_edge_cases( - submission_dir, output_dir, metadata, strategy, jinja_env - ) + edge_case_dirs = _scaffold_edge_cases(submission_dir, output_dir, metadata, strategy, jinja_env) if edge_case_dirs: logger.info("Scaffolded %d edge case(s)", len(edge_case_dirs)) diff --git a/tests/test_behavioral_checks.py b/tests/test_behavioral_checks.py index 7dd1d9c5..4826a66d 100644 --- a/tests/test_behavioral_checks.py +++ b/tests/test_behavioral_checks.py @@ -81,23 +81,29 @@ class TestStabilityCheck: """Tests for _check_stability() — score drift detection.""" def test_stable_scores_pass(self) -> None: - result = _check_stability({ - "stability": {"score_variance": 0.01, "run_count": 5}, - }) + result = _check_stability( + { + "stability": {"score_variance": 0.01, "run_count": 5}, + } + ) assert result.passed is True assert result.score > 0.5 def test_drifting_scores_fail(self) -> None: - result = _check_stability({ - "stability": {"score_variance": 0.2, "run_count": 5}, - }) + result = _check_stability( + { + "stability": {"score_variance": 0.2, "run_count": 5}, + } + ) assert result.passed is False assert result.score < 0.5 def test_insufficient_history_passes(self) -> None: - result = _check_stability({ - "stability": {"score_variance": 0.0, "run_count": 2}, - }) + result = _check_stability( + { + "stability": {"score_variance": 0.0, "run_count": 2}, + } + ) assert result.passed is True assert "Insufficient history" in result.message @@ -111,9 +117,11 @@ def test_none_stability_passes(self) -> None: assert result.passed is True def test_at_threshold_score_is_half(self) -> None: - result = _check_stability({ - "stability": {"score_variance": 0.1, "run_count": 5}, - }) + result = _check_stability( + { + "stability": {"score_variance": 0.1, "run_count": 5}, + } + ) assert result.passed is True assert result.score == pytest.approx(0.5) @@ -122,9 +130,11 @@ def test_no_data_has_status_sentinel(self) -> None: assert result.details["status"] == "no_data" def test_insufficient_history_has_status_sentinel(self) -> None: - result = _check_stability({ - "stability": {"score_variance": 0.0, "run_count": 1}, - }) + result = _check_stability( + { + "stability": {"score_variance": 0.0, "run_count": 1}, + } + ) assert result.details["status"] == "no_data" @@ -132,30 +142,38 @@ class TestEdgeCaseCheck: """Tests for _check_edge_case_results() — edge case pass rate.""" def test_all_passed(self) -> None: - result = _check_edge_case_results({ - "edge_cases": {"total": 3, "passed": 3}, - }) + result = _check_edge_case_results( + { + "edge_cases": {"total": 3, "passed": 3}, + } + ) assert result.passed is True assert result.score == 1.0 def test_some_failed(self) -> None: - result = _check_edge_case_results({ - "edge_cases": {"total": 4, "passed": 1}, - }) + result = _check_edge_case_results( + { + "edge_cases": {"total": 4, "passed": 1}, + } + ) assert result.passed is False assert result.score == 0.25 def test_all_failed(self) -> None: - result = _check_edge_case_results({ - "edge_cases": {"total": 3, "passed": 0}, - }) + result = _check_edge_case_results( + { + "edge_cases": {"total": 3, "passed": 0}, + } + ) assert result.passed is False assert result.score == 0.0 def test_no_edge_cases_defined(self) -> None: - result = _check_edge_case_results({ - "edge_cases": {"total": 0, "passed": 0}, - }) + result = _check_edge_case_results( + { + "edge_cases": {"total": 0, "passed": 0}, + } + ) assert result.passed is True assert result.score == 1.0 @@ -165,9 +183,11 @@ def test_no_edge_case_data(self) -> None: assert result.score == 1.0 def test_details_contain_counts(self) -> None: - result = _check_edge_case_results({ - "edge_cases": {"total": 5, "passed": 3}, - }) + result = _check_edge_case_results( + { + "edge_cases": {"total": 5, "passed": 3}, + } + ) assert result.details["total"] == 5 assert result.details["passed"] == 3 assert result.details["pass_rate"] == 0.6 @@ -177,16 +197,20 @@ class TestFailureModeCheck: """Tests for _check_failure_mode() — failure handling scores.""" def test_good_score_passes(self) -> None: - result = _check_failure_mode({ - "failure_mode": {"score": 0.8, "threshold": 0.5}, - }) + result = _check_failure_mode( + { + "failure_mode": {"score": 0.8, "threshold": 0.5}, + } + ) assert result.passed is True assert result.score == 0.8 def test_low_score_fails(self) -> None: - result = _check_failure_mode({ - "failure_mode": {"score": 0.3, "threshold": 0.5}, - }) + result = _check_failure_mode( + { + "failure_mode": {"score": 0.3, "threshold": 0.5}, + } + ) assert result.passed is False assert result.score == 0.3 @@ -323,6 +347,7 @@ def test_no_behavioral_data_check_fails(self) -> None: def test_behavioral_check_in_certified_checks(self) -> None: from abevalflow.certification import CERTIFIED_CHECKS + assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS def test_backward_compatible_no_behavioral_data(self) -> None: @@ -413,11 +438,13 @@ class TestEdgeCaseValidation: def test_no_edge_cases_dir_passes(self, tmp_path: Path) -> None: from scripts.validate import _check_edge_cases + errors = _check_edge_cases(tmp_path) assert errors == [] def test_valid_edge_cases(self, tmp_path: Path) -> None: from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" edge_dir.mkdir() (edge_dir / "empty_input.md").write_text("Test with empty input") @@ -427,6 +454,7 @@ def test_valid_edge_cases(self, tmp_path: Path) -> None: def test_empty_md_file(self, tmp_path: Path) -> None: from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" edge_dir.mkdir() (edge_dir / "empty_input.md").write_text("") @@ -436,6 +464,7 @@ def test_empty_md_file(self, tmp_path: Path) -> None: def test_non_md_files(self, tmp_path: Path) -> None: from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" edge_dir.mkdir() (edge_dir / "valid.md").write_text("Valid content") @@ -445,6 +474,7 @@ def test_non_md_files(self, tmp_path: Path) -> None: def test_empty_dir(self, tmp_path: Path) -> None: from scripts.validate import _check_edge_cases + edge_dir = tmp_path / "edge_cases" edge_dir.mkdir() errors = _check_edge_cases(tmp_path) @@ -456,6 +486,7 @@ class TestExtractBehavioralData: def test_extracts_std_reward(self, tmp_path: Path) -> None: from scripts.aggregate_scorecard import _extract_behavioral_data + report = {"summary": {"treatment": {"std_reward": 0.15}}} (tmp_path / "report.json").write_text(json.dumps(report)) data = _extract_behavioral_data(tmp_path, gates=[]) @@ -464,6 +495,7 @@ def test_extracts_std_reward(self, tmp_path: Path) -> None: def test_extracts_edge_cases_from_gate(self, tmp_path: Path) -> None: from scripts.aggregate_scorecard import _extract_behavioral_data + gate = GateResult( gate_type=GateType.BEHAVIORAL, gate_name="edge-case", @@ -477,6 +509,7 @@ def test_extracts_edge_cases_from_gate(self, tmp_path: Path) -> None: def test_returns_none_for_empty_report(self, tmp_path: Path) -> None: from scripts.aggregate_scorecard import _extract_behavioral_data + report = {"summary": {"treatment": {}}} (tmp_path / "report.json").write_text(json.dumps(report)) data = _extract_behavioral_data(tmp_path, gates=[]) @@ -484,11 +517,13 @@ def test_returns_none_for_empty_report(self, tmp_path: Path) -> None: def test_returns_none_when_no_report_and_no_gates(self, tmp_path: Path) -> None: from scripts.aggregate_scorecard import _extract_behavioral_data + data = _extract_behavioral_data(tmp_path, gates=[]) assert data is None def test_combines_std_reward_and_gate_edge_cases(self, tmp_path: Path) -> None: from scripts.aggregate_scorecard import _extract_behavioral_data + report = {"summary": {"treatment": {"std_reward": 0.2}}} (tmp_path / "report.json").write_text(json.dumps(report)) gate = GateResult( @@ -527,6 +562,7 @@ def _make_submission(self, tmp_path: Path) -> Path: def test_edge_case_dirs_created(self, tmp_path: Path) -> None: from scripts.scaffold import scaffold_submission + sub = self._make_submission(tmp_path) output = tmp_path / "output" scaffold_submission(sub, output) @@ -536,6 +572,7 @@ def test_edge_case_dirs_created(self, tmp_path: Path) -> None: def test_edge_case_uses_alternative_instruction(self, tmp_path: Path) -> None: from scripts.scaffold import scaffold_submission + sub = self._make_submission(tmp_path) output = tmp_path / "output" scaffold_submission(sub, output) @@ -546,6 +583,7 @@ def test_edge_case_uses_alternative_instruction(self, tmp_path: Path) -> None: def test_no_edge_cases_no_extra_dirs(self, tmp_path: Path) -> None: from scripts.scaffold import scaffold_submission + sub = tmp_path / "no-edge" sub.mkdir() (sub / "metadata.yaml").write_text("name: no-edge\n") @@ -583,21 +621,33 @@ def test_build_analysis_includes_edge_cases(self, tmp_path: Path) -> None: results = tmp_path / "results" treatment = results / "treatment" / "job" / "task__001" treatment.mkdir(parents=True) - (treatment / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": 0.8}, - })) + (treatment / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": 0.8}, + } + ) + ) control = results / "control" / "job" / "task__001" control.mkdir(parents=True) - (control / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": 0.3}, - })) + (control / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": 0.3}, + } + ) + ) edge_root = tmp_path / "edge_results" edge_dir = edge_root / "tasks-treatment-edge-empty_input" / "job" / "task__001" edge_dir.mkdir(parents=True) - (edge_dir / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": 0.6}, - })) + (edge_dir / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": 0.6}, + } + ) + ) result = build_analysis( results_dir=results, @@ -614,14 +664,22 @@ def test_build_analysis_no_edge_cases(self, tmp_path: Path) -> None: results = tmp_path / "results" treatment = results / "treatment" / "job" / "task__001" treatment.mkdir(parents=True) - (treatment / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": 0.8}, - })) + (treatment / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": 0.8}, + } + ) + ) control = results / "control" / "job" / "task__001" control.mkdir(parents=True) - (control / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": 0.3}, - })) + (control / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": 0.3}, + } + ) + ) result = build_analysis( results_dir=results, @@ -682,18 +740,26 @@ def test_scaffold_analyze_certify(self, tmp_path: Path) -> None: for i, reward in enumerate([0.8, 0.85] if variant == "treatment" else [0.3, 0.35]): trial_dir = results / variant / "job" / f"task__{i:03d}" trial_dir.mkdir(parents=True) - (trial_dir / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": reward}, - })) + (trial_dir / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": reward}, + } + ) + ) # 4. Create mock results for edge cases edge_results = tmp_path / "edge_results" for edge_name, reward in [("adversarial", 0.7), ("empty_input", 0.6)]: trial_dir = edge_results / f"tasks-treatment-edge-{edge_name}" / "job" / "task__001" trial_dir.mkdir(parents=True) - (trial_dir / "result.json").write_text(json.dumps({ - "verifier_result": {"reward": reward}, - })) + (trial_dir / "result.json").write_text( + json.dumps( + { + "verifier_result": {"reward": reward}, + } + ) + ) # 5. Analyze with edge case results analysis = build_analysis( @@ -716,9 +782,27 @@ def test_scaffold_analyze_certify(self, tmp_path: Path) -> None: # 7. Verify certification sees it result = compute_certification( gates=[ - GateResult(gate_name="evaluation", gate_type=GateType.ENGINE, passed=True, score=0.9, mode=GateMode.BLOCK), - GateResult(gate_name="security", gate_type=GateType.SECURITY, passed=True, score=1.0, mode=GateMode.BLOCK), - GateResult(gate_name="quality", gate_type=GateType.QUALITY, passed=True, score=0.8, mode=GateMode.WARN), + GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + ), + GateResult( + gate_name="security", + gate_type=GateType.SECURITY, + passed=True, + score=1.0, + mode=GateMode.BLOCK, + ), + GateResult( + gate_name="quality", + gate_type=GateType.QUALITY, + passed=True, + score=0.8, + mode=GateMode.WARN, + ), ], validation_passed=True, metadata_valid=True, @@ -737,6 +821,7 @@ class TestBehavioralGateRegistry: def test_edge_case_gate_registered(self) -> None: from abevalflow.gates.behavioral import get_all_behavioral_gates, get_behavioral_gate + gate = get_behavioral_gate("edge-case") assert isinstance(gate, EdgeCaseGate) @@ -746,6 +831,7 @@ def test_edge_case_gate_registered(self) -> None: def test_unknown_gate_raises(self) -> None: from abevalflow.gates.behavioral import get_behavioral_gate + with pytest.raises(KeyError, match="Unknown behavioral gate"): get_behavioral_gate("nonexistent") From 6bb801e801d4751acd9fadb832264e5430d4b9b9 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Mon, 13 Jul 2026 15:22:22 +0300 Subject: [PATCH 03/10] fix: address blocking review findings - Keep ENTERPRISE_BEHAVIORAL_TESTING out of CERTIFIED_CHECKS and profiles until pipeline wiring exists (prevents Certified regression) - Restore gate_name='behavioral', policy_key=self.name convention in EdgeCaseGate (fixes Compass push and weighted scorecard lookups) - Treat missing edge case trial output as failure instead of silently skipping (prevents inflated pass rates) - Remove dead DEFAULT_THRESHOLDS entry for behavioral testing Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/certification.py | 3 +- abevalflow/gates/behavioral/edge_case.py | 10 +-- config/certification_profiles.yaml | 6 +- scripts/analyze.py | 8 ++ tests/test_behavioral_checks.py | 94 +++++++++--------------- tests/test_certification.py | 5 +- 6 files changed, 52 insertions(+), 74 deletions(-) diff --git a/abevalflow/certification.py b/abevalflow/certification.py index 7511bf87..bb60cb94 100644 --- a/abevalflow/certification.py +++ b/abevalflow/certification.py @@ -94,7 +94,7 @@ class CheckId(StrEnum): CERTIFIED_CHECKS = [ CheckId.ENTERPRISE_STRUCTURE_VALIDATION, CheckId.ENTERPRISE_SECURITY_REVIEW, - CheckId.ENTERPRISE_BEHAVIORAL_TESTING, + # CheckId.ENTERPRISE_BEHAVIORAL_TESTING, # Implemented but not wired into pipeline yet CheckId.ADVANCED_AGENT_VALIDATION, # CheckId.CONTINUOUS_OPTIMIZATION, # Not yet implemented ] @@ -314,7 +314,6 @@ def get_level_result(self, level: CertificationLevel) -> LevelResult | None: CheckId.ADVANCED_AGENT_VALIDATION: 0.8, CheckId.ADVANCED_SECURITY_VALIDATION: 0.9, CheckId.INSTRUCTION_QUALITY: 0.7, - CheckId.ENTERPRISE_BEHAVIORAL_TESTING: 0.6, } BEHAVIORAL_SUB_CHECK_WEIGHTS = { diff --git a/abevalflow/gates/behavioral/edge_case.py b/abevalflow/gates/behavioral/edge_case.py index 89991a1f..bdd7d5aa 100644 --- a/abevalflow/gates/behavioral/edge_case.py +++ b/abevalflow/gates/behavioral/edge_case.py @@ -38,7 +38,7 @@ def evaluate( if gate_policy.mode == GateMode.DISABLED: return GateResult( gate_type=GateType.BEHAVIORAL, - gate_name=self.name, + gate_name="behavioral", policy_key=self.name, passed=True, score=1.0, @@ -50,7 +50,7 @@ def evaluate( if not report_path.exists(): return GateResult( gate_type=GateType.BEHAVIORAL, - gate_name=self.name, + gate_name="behavioral", policy_key=self.name, passed=True, score=1.0, @@ -65,7 +65,7 @@ def evaluate( logger.error("Failed to read report.json: %s", e) return GateResult( gate_type=GateType.BEHAVIORAL, - gate_name=self.name, + gate_name="behavioral", policy_key=self.name, passed=False, score=0.0, @@ -77,7 +77,7 @@ def evaluate( if not edge_case_results: return GateResult( gate_type=GateType.BEHAVIORAL, - gate_name=self.name, + gate_name="behavioral", policy_key=self.name, passed=True, score=1.0, @@ -98,7 +98,7 @@ def evaluate( return GateResult( gate_type=GateType.BEHAVIORAL, - gate_name=self.name, + gate_name="behavioral", policy_key=self.name, passed=passed, score=score, diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index 7803a9ff..da431797 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -29,7 +29,7 @@ profiles: certified: checks: - enterprise_structure_validation - - enterprise_behavioral_testing + # - enterprise_behavioral_testing # Implemented but not wired into pipeline yet - advanced_agent_validation # MCP Servers (Standardized Tool APIs) - API contracts and stability @@ -86,7 +86,7 @@ profiles: certified: checks: - enterprise_structure_validation - - enterprise_behavioral_testing + # - enterprise_behavioral_testing # Implemented but not wired into pipeline yet - advanced_agent_validation # - safety_toxicity_bias_guardrails # Not yet implemented @@ -113,7 +113,7 @@ profiles: checks: - enterprise_structure_validation - enterprise_security_review - - enterprise_behavioral_testing + # - enterprise_behavioral_testing # Implemented but not wired into pipeline yet - advanced_agent_validation # Not yet implemented: # - continuous_optimization diff --git a/scripts/analyze.py b/scripts/analyze.py index ddf9660b..549548fa 100644 --- a/scripts/analyze.py +++ b/scripts/analyze.py @@ -250,6 +250,14 @@ def analyze_edge_cases( edge_name = edge_dir.name.removeprefix("tasks-treatment-edge-") trials = parse_variant_trials(edge_dir) if not trials: + logger.warning("Edge case '%s' has no trial output — counted as failure", edge_name) + edge_results.append( + EdgeCaseResult( + name=edge_name, + summary=VariantSummary(), + passed=False, + ) + ) continue summary = compute_variant_summary(trials) mean_r = summary.mean_reward or 0.0 diff --git a/tests/test_behavioral_checks.py b/tests/test_behavioral_checks.py index 4826a66d..335dfc1c 100644 --- a/tests/test_behavioral_checks.py +++ b/tests/test_behavioral_checks.py @@ -312,43 +312,38 @@ def _make_all_gates(self) -> list[GateResult]: ), ] - def test_behavioral_data_enables_certified_check(self) -> None: + def test_behavioral_data_produces_passing_check(self) -> None: behavioral_data = { "std_reward": 0.1, "edge_cases": {"total": 3, "passed": 3}, } - result = compute_certification( - gates=self._make_all_gates(), - validation_passed=True, - metadata_valid=True, - has_eval_assets=True, - behavioral_data=behavioral_data, - ) - behavioral_check = next( - (c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING), - None, - ) - assert behavioral_check is not None - assert behavioral_check.passed is True + result = _compute_behavioral_testing_check(behavioral_data) + assert result.passed is True + assert result.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING - def test_no_behavioral_data_check_fails(self) -> None: + def test_no_behavioral_data_produces_failing_check(self) -> None: + result = _compute_behavioral_testing_check({}) + assert result.passed is False + + def test_behavioral_check_not_in_certified_checks_until_pipeline_wired(self) -> None: + from abevalflow.certification import CERTIFIED_CHECKS + + assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING not in CERTIFIED_CHECKS + + def test_behavioral_data_stored_but_not_required_for_certified(self) -> None: + behavioral_data = { + "std_reward": 0.1, + "edge_cases": {"total": 3, "passed": 3}, + } result = compute_certification( gates=self._make_all_gates(), validation_passed=True, metadata_valid=True, has_eval_assets=True, + behavioral_data=behavioral_data, ) - behavioral_check = next( - (c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING), - None, - ) - assert behavioral_check is not None - assert behavioral_check.passed is False - - def test_behavioral_check_in_certified_checks(self) -> None: - from abevalflow.certification import CERTIFIED_CHECKS - - assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS + # Check is computed and stored but doesn't block Certified level + assert result.certified.passed is True def test_backward_compatible_no_behavioral_data(self) -> None: result = compute_certification( @@ -605,6 +600,16 @@ def test_no_edge_cases_no_extra_dirs(self, tmp_path: Path) -> None: class TestEdgeCaseAnalysis: """Tests for edge case analysis in analyze.py.""" + def test_missing_trial_output_counted_as_failure(self, tmp_path: Path) -> None: + from scripts.analyze import analyze_edge_cases + + edge_dir = tmp_path / "tasks-treatment-edge-empty_input" + edge_dir.mkdir() + results = analyze_edge_cases(tmp_path) + assert len(results) == 1 + assert results[0].name == "empty_input" + assert results[0].passed is False + def test_edge_case_result_model(self) -> None: result = EdgeCaseResult( name="empty_input", @@ -770,7 +775,7 @@ def test_scaffold_analyze_certify(self, tmp_path: Path) -> None: assert len(analysis.edge_case_results) == 2 assert all(ec.passed for ec in analysis.edge_case_results) - # 6. Simulate the unified data path: std_reward from report + edge cases from gate + # 6. Verify behavioral check passes with the extracted data behavioral_data = { "std_reward": analysis.summary.treatment.std_reward, "edge_cases": { @@ -778,40 +783,7 @@ def test_scaffold_analyze_certify(self, tmp_path: Path) -> None: "passed": sum(1 for ec in analysis.edge_case_results if ec.passed), }, } - - # 7. Verify certification sees it - result = compute_certification( - gates=[ - GateResult( - gate_name="evaluation", - gate_type=GateType.ENGINE, - passed=True, - score=0.9, - mode=GateMode.BLOCK, - ), - GateResult( - gate_name="security", - gate_type=GateType.SECURITY, - passed=True, - score=1.0, - mode=GateMode.BLOCK, - ), - GateResult( - gate_name="quality", - gate_type=GateType.QUALITY, - passed=True, - score=0.8, - mode=GateMode.WARN, - ), - ], - validation_passed=True, - metadata_valid=True, - has_eval_assets=True, - behavioral_data=behavioral_data, - ) - behavioral_check = next( - c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING - ) + behavioral_check = _compute_behavioral_testing_check(behavioral_data) assert behavioral_check.passed is True assert behavioral_check.score > 0.0 diff --git a/tests/test_certification.py b/tests/test_certification.py index af517f98..9e0f2bd4 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -353,10 +353,9 @@ def test_trusted_has_expected_checks(self) -> None: def test_certified_has_expected_checks(self) -> None: assert CheckId.ENTERPRISE_STRUCTURE_VALIDATION in CERTIFIED_CHECKS assert CheckId.ENTERPRISE_SECURITY_REVIEW in CERTIFIED_CHECKS - assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS assert CheckId.ADVANCED_AGENT_VALIDATION in CERTIFIED_CHECKS - # continuous_optimization not yet implemented - assert len(CERTIFIED_CHECKS) == 4 + # behavioral_testing and continuous_optimization not yet wired into pipeline + assert len(CERTIFIED_CHECKS) == 3 class TestCertificationPolicy: From a111da729ab69400061732b78be3db4445cafac6 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Mon, 13 Jul 2026 15:27:06 +0300 Subject: [PATCH 04/10] fix: EdgeCaseGate fail-closed in BLOCK mode when report.json missing Consistent with security gates which fail when scan files are missing in BLOCK mode. WARN mode still passes gracefully. Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/gates/behavioral/edge_case.py | 11 +++++++++++ tests/test_behavioral_checks.py | 7 +++++++ 2 files changed, 18 insertions(+) diff --git a/abevalflow/gates/behavioral/edge_case.py b/abevalflow/gates/behavioral/edge_case.py index bdd7d5aa..5839b6d2 100644 --- a/abevalflow/gates/behavioral/edge_case.py +++ b/abevalflow/gates/behavioral/edge_case.py @@ -48,6 +48,17 @@ def evaluate( report_path = reports_dir / "report.json" if not report_path.exists(): + if gate_policy.mode == GateMode.BLOCK: + return GateResult( + gate_type=GateType.BEHAVIORAL, + gate_name="behavioral", + policy_key=self.name, + passed=False, + score=0.0, + mode=gate_policy.mode, + message="FAIL: report.json missing (required in block mode)", + details={"status": "not_found"}, + ) return GateResult( gate_type=GateType.BEHAVIORAL, gate_name="behavioral", diff --git a/tests/test_behavioral_checks.py b/tests/test_behavioral_checks.py index 335dfc1c..43ee94ac 100644 --- a/tests/test_behavioral_checks.py +++ b/tests/test_behavioral_checks.py @@ -366,6 +366,13 @@ def test_no_report_json(self, tmp_path: Path) -> None: assert result.passed is True assert result.gate_type == GateType.BEHAVIORAL + def test_no_report_block_mode_fails(self, tmp_path: Path) -> None: + policy = GatePolicy(gates={"behavioral": GatePolicyItem(mode=GateMode.BLOCK)}) + gate = EdgeCaseGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is False + assert result.score == 0.0 + def test_disabled_mode(self, tmp_path: Path) -> None: policy = GatePolicy(gates={"behavioral": GatePolicyItem(mode=GateMode.DISABLED)}) gate = EdgeCaseGate() From cc7e3db2e7470feb8addd610bf642a2dfb3780e0 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 14 Jul 2026 15:43:46 +0300 Subject: [PATCH 05/10] fix: rebase onto main, add consistency+mean interaction tests - Resolve conflicts with APPENG-5623 (operational policy) merge - Pass operational_policy_result in certification integration tests - Add tests verifying consistency and mean reward are independent: low mean + low variance passes consistency (quality caught elsewhere), high mean + high variance fails consistency (flaky behavior caught) Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/test_behavioral_checks.py | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/tests/test_behavioral_checks.py b/tests/test_behavioral_checks.py index 43ee94ac..30b03b3e 100644 --- a/tests/test_behavioral_checks.py +++ b/tests/test_behavioral_checks.py @@ -9,6 +9,7 @@ from abevalflow.certification import ( CheckId, + CheckResult, _check_consistency, _check_edge_case_results, _check_failure_mode, @@ -76,6 +77,16 @@ def test_no_data_has_status_sentinel(self) -> None: result = _check_consistency({}) assert result.details["status"] == "no_data" + def test_low_mean_low_variance_passes_consistency(self) -> None: + """Consistency only measures reliability, not quality.""" + result = _check_consistency({"std_reward": 0.02}) + assert result.passed is True + + def test_high_mean_high_variance_fails_consistency(self) -> None: + """Good average doesn't help if results are unpredictable.""" + result = _check_consistency({"std_reward": 0.4}) + assert result.passed is False + class TestStabilityCheck: """Tests for _check_stability() — score drift detection.""" @@ -330,6 +341,15 @@ def test_behavioral_check_not_in_certified_checks_until_pipeline_wired(self) -> assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING not in CERTIFIED_CHECKS + def _passing_operational_policy(self) -> CheckResult: + return CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=True, + score=1.0, + message="Passed", + ) + def test_behavioral_data_stored_but_not_required_for_certified(self) -> None: behavioral_data = { "std_reward": 0.1, @@ -340,9 +360,9 @@ def test_behavioral_data_stored_but_not_required_for_certified(self) -> None: validation_passed=True, metadata_valid=True, has_eval_assets=True, + operational_policy_result=self._passing_operational_policy(), behavioral_data=behavioral_data, ) - # Check is computed and stored but doesn't block Certified level assert result.certified.passed is True def test_backward_compatible_no_behavioral_data(self) -> None: @@ -351,6 +371,7 @@ def test_backward_compatible_no_behavioral_data(self) -> None: validation_passed=True, metadata_valid=True, has_eval_assets=True, + operational_policy_result=self._passing_operational_policy(), ) assert result.foundational.passed is True assert result.trusted.passed is True From 3900067660b17e379633fe14ef83759c20b6fd60 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 16 Jul 2026 11:41:00 +0300 Subject: [PATCH 06/10] feat: wire edge case evaluation via ASE + enable behavioral testing Pipeline wiring: - Add ase-eval-edge-cases step: runs separate ASE per edge case .md file - Add ase-aggregate-edge-cases step: produces per-edge-case pass/fail - Merge edge case results into report.json in analyze task - New scripts: generate_edge_case_evals.py, aggregate_edge_case_evals.py Certification: - Uncomment ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS - Enable in skill profile only (not agent/full per Guy's request) Cleanup: - Remove dead Harbor-based analyze_edge_cases() and --edge-case-results-dir - Simplify scaffold.py: removed Harbor edge case scaffolding since edge cases now run via ASE directly from the submission directory - Make EdgeCaseResult.summary optional for ASE-produced results - Add integration test: low-mean skill blocked by engine gate even when consistency passes Co-Authored-By: Claude Opus 4.6 (1M context) --- abevalflow/certification.py | 2 +- abevalflow/report.py | 6 +- config/certification_profiles.yaml | 2 +- pipeline/tasks/phases/evaluate.yaml | 137 ++++++++ .../post/analyze-and-check-degradation.yaml | 16 + scripts/aggregate_edge_case_evals.py | 116 +++++++ scripts/analyze.py | 77 +---- scripts/generate_edge_case_evals.py | 173 ++++++++++ scripts/scaffold.py | 50 +-- tests/test_behavioral_checks.py | 312 ++++++++---------- tests/test_certification.py | 5 +- 11 files changed, 605 insertions(+), 291 deletions(-) create mode 100644 scripts/aggregate_edge_case_evals.py create mode 100644 scripts/generate_edge_case_evals.py diff --git a/abevalflow/certification.py b/abevalflow/certification.py index bb60cb94..31f549d9 100644 --- a/abevalflow/certification.py +++ b/abevalflow/certification.py @@ -94,7 +94,7 @@ class CheckId(StrEnum): CERTIFIED_CHECKS = [ CheckId.ENTERPRISE_STRUCTURE_VALIDATION, CheckId.ENTERPRISE_SECURITY_REVIEW, - # CheckId.ENTERPRISE_BEHAVIORAL_TESTING, # Implemented but not wired into pipeline yet + CheckId.ENTERPRISE_BEHAVIORAL_TESTING, CheckId.ADVANCED_AGENT_VALIDATION, # CheckId.CONTINUOUS_OPTIMIZATION, # Not yet implemented ] diff --git a/abevalflow/report.py b/abevalflow/report.py index 239f0759..38cd3aa8 100644 --- a/abevalflow/report.py +++ b/abevalflow/report.py @@ -166,8 +166,12 @@ class EdgeCaseResult(BaseModel): """Result of a single edge case evaluation.""" name: str = Field(description="Edge case name (derived from filename)") - summary: VariantSummary = Field(description="Trial stats for this edge case") + summary: VariantSummary | None = Field( + default=None, + description="Trial stats (present for Harbor-based eval, absent for ASE-based eval)", + ) passed: bool = Field(description="Whether the edge case passed verification") + score: float | None = Field(default=None, description="Edge case score (0.0-1.0)") class AnalysisResult(BaseModel): diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index da431797..82553cb6 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -29,7 +29,7 @@ profiles: certified: checks: - enterprise_structure_validation - # - enterprise_behavioral_testing # Implemented but not wired into pipeline yet + - enterprise_behavioral_testing - advanced_agent_validation # MCP Servers (Standardized Tool APIs) - API contracts and stability diff --git a/pipeline/tasks/phases/evaluate.yaml b/pipeline/tasks/phases/evaluate.yaml index 564b90dd..8d1942df 100644 --- a/pipeline/tasks/phases/evaluate.yaml +++ b/pipeline/tasks/phases/evaluate.yaml @@ -410,6 +410,143 @@ spec: print(f"Recommendation: {rec}") COMPUTE_REWARDS + # Step 5b: ASE - Run Edge Case Evaluations (one ASE run per edge case) + - name: ase-eval-edge-cases + image: registry.access.redhat.com/ubi9/nodejs-20:latest + env: + - name: OPENAI_API_KEY + value: "$(params.llm-api-key)" + script: | + #!/usr/bin/env bash + set -euo pipefail + + EVAL_ENGINE="$(params.eval-engine)" + if [ "$EVAL_ENGINE" != "harbor" ] && [ "$EVAL_ENGINE" != "both" ]; then + echo "Skipping edge case eval (eval-engine=$EVAL_ENGINE)" + exit 0 + fi + + SUBMISSION_PATH="$(workspaces.source.path)/submissions/$(params.submission-dir)" + EDGE_CASES_DIR="$SUBMISSION_PATH/edge_cases" + + if [ ! -d "$EDGE_CASES_DIR" ]; then + echo "No edge_cases/ directory — skipping edge case evaluation" + exit 0 + fi + + MD_FILES=$(find "$EDGE_CASES_DIR" -name "*.md" -type f | sort) + MD_COUNT=$(echo "$MD_FILES" | grep -c . || true) + if [ "$MD_COUNT" -eq 0 ]; then + echo "edge_cases/ has no .md files — skipping" + exit 0 + fi + + echo "=== EVALUATE PHASE: ASE Edge Case Evaluation ($MD_COUNT edge cases) ===" + + PIPELINE_DIR="$(workspaces.source.path)/_pipeline" + RESULTS_BASE="$(workspaces.source.path)/eval-results/$(params.submission-name)/edge-cases" + mkdir -p "$RESULTS_BASE" + + # Find SKILL.md + SKILL_MD="" + if [ -f "$SUBMISSION_PATH/skills/SKILL.md" ]; then + SKILL_MD="$SUBMISSION_PATH/skills/SKILL.md" + else + for dir in "$SUBMISSION_PATH"/skills/*/; do + if [ -f "${dir}SKILL.md" ]; then + SKILL_MD="${dir}SKILL.md" + break + fi + done + fi + + if [ -z "$SKILL_MD" ]; then + echo "WARNING: No SKILL.md found — edge case evaluation skipped" + exit 0 + fi + + # Install ASE once + npm install --global agent-skills-eval 2>&1 | tail -3 + + LLM_MODEL="$(params.llm-model)" + BASE_URL="$(params.llm-api-base)" + JUDGE_MODEL="$(params.ase-judge-model)" + [ -z "$JUDGE_MODEL" ] && JUDGE_MODEL="$LLM_MODEL" + [[ "$BASE_URL" != */v1 ]] && BASE_URL="${BASE_URL}/v1" + + # Run a separate ASE evaluation for each edge case + EDGE_NUM=0 + for MD_FILE in $MD_FILES; do + EDGE_NAME=$(basename "$MD_FILE" .md) + EDGE_NUM=$((EDGE_NUM + 1)) + echo "--- Edge case $EDGE_NUM/$MD_COUNT: $EDGE_NAME ---" + + PROMPT=$(cat "$MD_FILE") + if [ -z "$PROMPT" ]; then + echo "Skipping empty edge case: $EDGE_NAME" + continue + fi + + # Create isolated skill dir for this edge case + EDGE_SKILL_DIR="$(workspaces.source.path)/_eval_tmp/edge-case-$EDGE_NAME" + rm -rf "$EDGE_SKILL_DIR" + mkdir -p "$EDGE_SKILL_DIR/evals" + cp "$SKILL_MD" "$EDGE_SKILL_DIR/SKILL.md" + + # Generate single-entry evals.json for this edge case + cd "$PIPELINE_DIR" + export PYTHONPATH="$PIPELINE_DIR" + python scripts/generate_edge_case_evals.py \ + --submission-dir "$SUBMISSION_PATH" \ + --output-dir "$EDGE_SKILL_DIR/evals" \ + --edge-case-file "$MD_FILE" + + mv "$EDGE_SKILL_DIR/evals/edge-case-evals.json" "$EDGE_SKILL_DIR/evals/evals.json" + + # Run ASE for this single edge case + EDGE_RESULTS="$RESULTS_BASE/$EDGE_NAME" + mkdir -p "$EDGE_RESULTS" + + agent-skills-eval "$EDGE_SKILL_DIR" --baseline --layout iteration --report \ + --base-url "$BASE_URL" --target "$LLM_MODEL" --judge "$JUDGE_MODEL" \ + --concurrency 1 --workspace "$EDGE_RESULTS/iteration-1" \ + --api-key-env OPENAI_API_KEY || true + + echo "Edge case '$EDGE_NAME' complete" + done + + echo "All edge case evaluations complete" + + # Step 5c: Aggregate edge case results + - name: ase-aggregate-edge-cases + image: registry.access.redhat.com/ubi9/python-311:9.6 + script: | + #!/usr/bin/env bash + set -euo pipefail + + EVAL_ENGINE="$(params.eval-engine)" + if [ "$EVAL_ENGINE" != "harbor" ] && [ "$EVAL_ENGINE" != "both" ]; then + exit 0 + fi + + RESULTS_DIR="$(workspaces.source.path)/eval-results/$(params.submission-name)/edge-cases" + REPORTS_DIR="$(workspaces.source.path)/reports/$(params.submission-name)" + + if [ ! -d "$RESULTS_DIR" ]; then + echo "No edge case results to aggregate" + exit 0 + fi + + echo "=== EVALUATE PHASE: Aggregate Edge Case Results ===" + PIPELINE_DIR="$(workspaces.source.path)/_pipeline" + cd "$PIPELINE_DIR" + export PYTHONPATH="$PIPELINE_DIR" + pip install --quiet --no-cache-dir pyyaml pydantic 2>/dev/null || true + + python scripts/aggregate_edge_case_evals.py \ + --results-dir "$RESULTS_DIR" \ + --output-dir "$REPORTS_DIR" + # Step 6: ASE - Run Evaluation - name: ase-eval image: registry.access.redhat.com/ubi9/nodejs-20:latest diff --git a/pipeline/tasks/post/analyze-and-check-degradation.yaml b/pipeline/tasks/post/analyze-and-check-degradation.yaml index 9f937e06..626fa340 100644 --- a/pipeline/tasks/post/analyze-and-check-degradation.yaml +++ b/pipeline/tasks/post/analyze-and-check-degradation.yaml @@ -242,6 +242,22 @@ spec: python3 -c "import sys;sys.path.insert(0,'$PIPELINE_DIR');from pathlib import Path;from abevalflow.report import AnalysisResult;from scripts.analyze import render_markdown;d=Path(sys.argv[1]);r=AnalysisResult.model_validate_json((d/'report.json').read_text());(d/'report.md').write_text(render_markdown(r))" "$REPORT_DIR" fi + # --- 5b. Merge edge case results into report.json (if present) --- + EDGE_RESULTS="$REPORT_DIR/edge-case-results.json" + if [ -f "$EDGE_RESULTS" ]; then + echo "Merging edge case results into report.json..." + python3 -c " +import json, sys +report_path = sys.argv[1] + '/report.json' +edge_path = sys.argv[2] +report = json.loads(open(report_path).read()) +edge = json.loads(open(edge_path).read()) +report['edge_case_results'] = edge +open(report_path, 'w').write(json.dumps(report, indent=2)) +print(f'Merged {len(edge)} edge case results') +" "$REPORT_DIR" "$EDGE_RESULTS" + fi + # --- 6. Extract results for Tekton --- echo -n "$REPORT_DIR" > "$(results.report-path.path)" diff --git a/scripts/aggregate_edge_case_evals.py b/scripts/aggregate_edge_case_evals.py new file mode 100644 index 00000000..25634c79 --- /dev/null +++ b/scripts/aggregate_edge_case_evals.py @@ -0,0 +1,116 @@ +"""Aggregate ASE edge case evaluation results into per-edge-case pass/fail. + +Reads grading.json files from the ASE edge case run and produces a JSON +array of per-edge-case results compatible with the EdgeCaseGate and +report.json's edge_case_results field. + +Usage:: + + python scripts/aggregate_edge_case_evals.py \\ + --results-dir /workspace/eval-results/my-submission/edge-cases \\ + --output-dir /workspace/reports/my-submission +""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +logger = logging.getLogger(__name__) + + +def aggregate_edge_case_results(results_dir: Path) -> list[dict]: + """Parse per-edge-case ASE results into pass/fail. + + Each edge case ran as a separate ASE invocation under its own + subdirectory: results_dir//iteration-1/... + + For each edge case subdirectory, finds the with_skill grading.json + and extracts the overall pass rate as the edge case score. + + Args: + results_dir: Directory containing per-edge-case result subdirs + (e.g. eval-results//edge-cases/) + + Returns: + List of dicts with {name, passed, score} per edge case. + """ + edge_results: list[dict] = [] + + for edge_dir in sorted(results_dir.iterdir()): + if not edge_dir.is_dir(): + continue + + edge_name = edge_dir.name + grading_files = sorted(edge_dir.rglob("grading.json")) + with_skill_gradings = [g for g in grading_files if "with_skill" in g.parts] + + if not with_skill_gradings: + logger.warning("Edge case '%s': no with_skill grading.json — counted as failure", edge_name) + edge_results.append({"name": edge_name, "passed": False, "score": 0.0}) + continue + + total_passed = 0 + total_count = 0 + for grading_path in with_skill_gradings: + try: + data = json.loads(grading_path.read_text()) + summary = data.get("summary", {}) + total_passed += summary.get("passed", 0) + total_count += summary.get("total", 0) + except (json.JSONDecodeError, OSError) as e: + logger.warning("Failed to parse %s: %s", grading_path, e) + + if total_count == 0: + edge_results.append({"name": edge_name, "passed": False, "score": 0.0}) + else: + score = total_passed / total_count + edge_results.append({"name": edge_name, "passed": score >= 0.5, "score": score}) + + return edge_results + + +def main(argv: list[str] | None = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + + parser = argparse.ArgumentParser( + description="Aggregate ASE edge case results into per-edge-case pass/fail", + ) + parser.add_argument( + "--results-dir", + type=Path, + required=True, + help="Path to ASE edge case results directory", + ) + parser.add_argument( + "--output-dir", + type=Path, + required=True, + help="Directory to write edge-case-results.json", + ) + + args = parser.parse_args(argv) + + if not args.results_dir.is_dir(): + logger.info("No edge case results directory found — nothing to aggregate") + return 0 + + results = aggregate_edge_case_results(args.results_dir) + + if not results: + logger.info("No edge case results to aggregate") + return 0 + + args.output_dir.mkdir(parents=True, exist_ok=True) + output_path = args.output_dir / "edge-case-results.json" + output_path.write_text(json.dumps(results, indent=2)) + logger.info("Wrote %d edge case results to %s", len(results), output_path) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/analyze.py b/scripts/analyze.py index 549548fa..a8a6c610 100644 --- a/scripts/analyze.py +++ b/scripts/analyze.py @@ -41,7 +41,6 @@ AnalysisResult, AnalysisSummary, DegradationResult, - EdgeCaseResult, Provenance, Recommendation, ScanMode, @@ -226,54 +225,6 @@ def build_a2a_analysis( ) -# --------------------------------------------------------------------------- -# Edge case analysis -# --------------------------------------------------------------------------- - - -def analyze_edge_cases( - workspace_dir: Path, - pass_threshold: float = 0.5, -) -> list[EdgeCaseResult]: - """Analyze edge case evaluation results. - - Args: - workspace_dir: Root directory containing tasks-treatment-edge-*/ - directories produced by scaffold.py. - pass_threshold: Minimum mean reward for an edge case to pass. - """ - edge_results: list[EdgeCaseResult] = [] - - for edge_dir in sorted(workspace_dir.glob("tasks-treatment-edge-*")): - if not edge_dir.is_dir(): - continue - edge_name = edge_dir.name.removeprefix("tasks-treatment-edge-") - trials = parse_variant_trials(edge_dir) - if not trials: - logger.warning("Edge case '%s' has no trial output — counted as failure", edge_name) - edge_results.append( - EdgeCaseResult( - name=edge_name, - summary=VariantSummary(), - passed=False, - ) - ) - continue - summary = compute_variant_summary(trials) - mean_r = summary.mean_reward or 0.0 - passed = mean_r >= pass_threshold - - edge_results.append( - EdgeCaseResult( - name=edge_name, - summary=summary, - passed=passed, - ) - ) - - return edge_results - - # --------------------------------------------------------------------------- # Statistics # --------------------------------------------------------------------------- @@ -380,15 +331,8 @@ def build_analysis( related_pr: str | None = None, llm_label: str | None = None, eval_engine: str | None = None, - edge_case_results_dir: Path | None = None, ) -> AnalysisResult: - """Parse results, compute stats, and assemble the full analysis model. - - Args: - edge_case_results_dir: Root directory containing tasks-treatment-edge-*/ - result directories. When provided, edge case results are analyzed - and included in the report. - """ + """Parse results, compute stats, and assemble the full analysis model.""" if eval_engine == "a2a" or is_a2a_results(results_dir): return build_a2a_analysis( results_dir=results_dir, @@ -434,16 +378,6 @@ def build_analysis( else: recommendation = Recommendation.PASS if primary_gap >= threshold else Recommendation.FAIL - edge_cases: list[EdgeCaseResult] = [] - if edge_case_results_dir is not None: - edge_cases = analyze_edge_cases(edge_case_results_dir) - if edge_cases: - logger.info( - "Edge cases: %d/%d passed", - sum(1 for ec in edge_cases if ec.passed), - len(edge_cases), - ) - return AnalysisResult( submission_name=submission_name, provenance=provenance or Provenance(), @@ -462,7 +396,6 @@ def build_analysis( "treatment": treatment_trials, "control": control_trials, }, - edge_case_results=edge_cases, ) @@ -704,13 +637,6 @@ def main(argv: list[str] | None = None) -> int: default=None, help="LLM model label for the report (e.g. 'Claude Sonnet 4.6 (vertex_ai)')", ) - parser.add_argument( - "--edge-case-results-dir", - type=Path, - default=None, - help="Root directory containing tasks-treatment-edge-*/ result directories", - ) - args = parser.parse_args(argv) if args.merge_degradation_from is not None: @@ -751,7 +677,6 @@ def main(argv: list[str] | None = None) -> int: related_pr=args.pr_url, llm_label=args.llm_label, eval_engine=args.eval_engine, - edge_case_results_dir=args.edge_case_results_dir, ) # Include security scan results if available diff --git a/scripts/generate_edge_case_evals.py b/scripts/generate_edge_case_evals.py new file mode 100644 index 00000000..48c19b26 --- /dev/null +++ b/scripts/generate_edge_case_evals.py @@ -0,0 +1,173 @@ +"""Generate ASE-compatible evals.json from edge_cases/*.md files. + +Converts each edge case markdown file into an eval entry with the file +content as the prompt and default behavioral assertions. + +Usage:: + + python scripts/generate_edge_case_evals.py \\ + --submission-dir /workspace/submissions/my-submission \\ + --output-dir /workspace/reports/my-submission +""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +logger = logging.getLogger(__name__) + +DEFAULT_ASSERTIONS = [ + "The response handles the edge case appropriately without crashing or producing nonsensical output", + "The response acknowledges limitations or uncertainty when the input is unusual", + "The response remains helpful and coherent despite the edge case conditions", +] + + +def generate_edge_case_evals( + submission_dir: Path, + skill_name: str | None = None, +) -> dict | None: + """Convert edge_cases/*.md files into an ASE evals.json structure. + + Args: + submission_dir: Path to the submission directory. + skill_name: Skill name for the evals. If None, derived from metadata.yaml. + + Returns: + Dict with evals.json structure, or None if no edge cases exist. + """ + edge_cases_dir = submission_dir / "edge_cases" + if not edge_cases_dir.is_dir(): + return None + + md_files = sorted(edge_cases_dir.glob("*.md")) + if not md_files: + return None + + if skill_name is None: + metadata_path = submission_dir / "metadata.yaml" + if metadata_path.exists(): + import yaml + + with metadata_path.open() as f: + meta = yaml.safe_load(f) or {} + skill_name = meta.get("name", "unknown-skill") + else: + skill_name = "unknown-skill" + + evals = [] + for md_file in md_files: + edge_name = md_file.stem + prompt = md_file.read_text().strip() + if not prompt: + logger.warning("Skipping empty edge case file: %s", md_file.name) + continue + + evals.append( + { + "id": f"edge-{edge_name}", + "name": f"Edge case: {edge_name.replace('_', ' ')}", + "prompt": prompt, + "assertions": list(DEFAULT_ASSERTIONS), + } + ) + + if not evals: + return None + + return { + "skill_name": skill_name, + "evals": evals, + } + + +def generate_single_edge_case_eval( + md_file: Path, + skill_name: str = "unknown-skill", +) -> dict | None: + """Generate an ASE evals.json for a single edge case .md file.""" + prompt = md_file.read_text().strip() + if not prompt: + return None + + edge_name = md_file.stem + return { + "skill_name": skill_name, + "evals": [ + { + "id": f"edge-{edge_name}", + "name": f"Edge case: {edge_name.replace('_', ' ')}", + "prompt": prompt, + "assertions": list(DEFAULT_ASSERTIONS), + } + ], + } + + +def main(argv: list[str] | None = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + + parser = argparse.ArgumentParser( + description="Generate ASE evals.json from edge_cases/*.md files", + ) + parser.add_argument( + "--submission-dir", + type=Path, + required=True, + help="Path to the submission directory containing edge_cases/", + ) + parser.add_argument( + "--output-dir", + type=Path, + required=True, + help="Directory to write edge-case-evals.json", + ) + parser.add_argument( + "--skill-name", + default=None, + help="Skill name (default: from metadata.yaml)", + ) + parser.add_argument( + "--edge-case-file", + type=Path, + default=None, + help="Generate evals for a single .md file instead of all edge_cases/", + ) + + args = parser.parse_args(argv) + + if not args.submission_dir.is_dir(): + logger.error("Submission directory does not exist: %s", args.submission_dir) + return 1 + + if args.edge_case_file: + skill_name = args.skill_name or "unknown-skill" + metadata_path = args.submission_dir / "metadata.yaml" + if metadata_path.exists() and not args.skill_name: + import yaml + + with metadata_path.open() as f: + meta = yaml.safe_load(f) or {} + skill_name = meta.get("name", skill_name) + result = generate_single_edge_case_eval(args.edge_case_file, skill_name) + else: + result = generate_edge_case_evals(args.submission_dir, args.skill_name) + + if result is None: + logger.info("No edge cases found — nothing to generate") + return 0 + + args.output_dir.mkdir(parents=True, exist_ok=True) + output_path = args.output_dir / "edge-case-evals.json" + output_path.write_text(json.dumps(result, indent=2)) + logger.info("Wrote %d edge case evals to %s", len(result["evals"]), output_path) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/scaffold.py b/scripts/scaffold.py index 3f7c73b3..459fffee 100644 --- a/scripts/scaffold.py +++ b/scripts/scaffold.py @@ -220,48 +220,16 @@ def _copy_root_dirs(submission_dir: Path, target_dir: Path) -> None: shutil.copytree(src, target_dir / rootdir, dirs_exist_ok=True) -def _scaffold_edge_cases( - submission_dir: Path, - output_dir: Path, - metadata: SubmissionMetadata, - strategy: ExperimentStrategy, - jinja_env: Environment, -) -> list[Path]: - """Scaffold per-edge-case treatment task directories. - - For each .md file in edge_cases/, creates a task directory identical to - the treatment variant but with the edge case instruction instead of the - main instruction.md. No control variant is needed for edge cases. +def _count_edge_cases(submission_dir: Path) -> int: + """Count edge case .md files in the submission's edge_cases/ directory. - Returns list of edge case task directory paths. + Edge cases are evaluated via ASE (not Harbor) so no scaffolding is needed. + The ASE pipeline step reads edge_cases/ directly from the submission. """ edge_cases_dir = submission_dir / "edge_cases" if not edge_cases_dir.is_dir(): - return [] - - context = _build_template_context(metadata, submission_dir, "treatment", strategy) - rendered = _render_templates(jinja_env, context) - strategy_srcs = [src for src, _ in context.get("copy_pairs", [])] - - # Same templates for all edge cases; only instruction.md differs per case - edge_case_dirs: list[Path] = [] - for md_file in sorted(edge_cases_dir.glob("*.md")): - edge_name = md_file.stem - edge_dir = output_dir / f"tasks-treatment-edge-{edge_name}" / metadata.name - - environment_dir = _write_rendered_templates(rendered, edge_dir) - _copy_submission_files(submission_dir, environment_dir, strategy_srcs, "treatment") - - # Two copies: environment/ for the Docker build context, - # task root for Harbor metadata/display (same as main scaffolding). - shutil.copy2(md_file, environment_dir / "instruction.md") - shutil.copy2(md_file, edge_dir / "instruction.md") - _copy_root_dirs(submission_dir, edge_dir) - - edge_case_dirs.append(edge_dir) - logger.info("Scaffolded edge case '%s' at %s", edge_name, edge_dir) - - return edge_case_dirs + return 0 + return len(list(edge_cases_dir.glob("*.md"))) def scaffold_submission( @@ -317,9 +285,9 @@ def scaffold_submission( logger.info("Scaffolded %s variant at %s", variant, target_dir) - edge_case_dirs = _scaffold_edge_cases(submission_dir, output_dir, metadata, strategy, jinja_env) - if edge_case_dirs: - logger.info("Scaffolded %d edge case(s)", len(edge_case_dirs)) + n_edge_cases = _count_edge_cases(submission_dir) + if n_edge_cases: + logger.info("Found %d edge case(s) (will be evaluated via ASE)", n_edge_cases) return treatment_dir, control_dir diff --git a/tests/test_behavioral_checks.py b/tests/test_behavioral_checks.py index 30b03b3e..fd0554d6 100644 --- a/tests/test_behavioral_checks.py +++ b/tests/test_behavioral_checks.py @@ -336,10 +336,10 @@ def test_no_behavioral_data_produces_failing_check(self) -> None: result = _compute_behavioral_testing_check({}) assert result.passed is False - def test_behavioral_check_not_in_certified_checks_until_pipeline_wired(self) -> None: + def test_behavioral_check_in_certified_checks(self) -> None: from abevalflow.certification import CERTIFIED_CHECKS - assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING not in CERTIFIED_CHECKS + assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS def _passing_operational_policy(self) -> CheckResult: return CheckResult( @@ -350,7 +350,7 @@ def _passing_operational_policy(self) -> CheckResult: message="Passed", ) - def test_behavioral_data_stored_but_not_required_for_certified(self) -> None: + def test_behavioral_data_required_for_certified(self) -> None: behavioral_data = { "std_reward": 0.1, "edge_cases": {"total": 3, "passed": 3}, @@ -365,7 +365,7 @@ def test_behavioral_data_stored_but_not_required_for_certified(self) -> None: ) assert result.certified.passed is True - def test_backward_compatible_no_behavioral_data(self) -> None: + def test_no_behavioral_data_blocks_certified(self) -> None: result = compute_certification( gates=self._make_all_gates(), validation_passed=True, @@ -375,6 +375,65 @@ def test_backward_compatible_no_behavioral_data(self) -> None: ) assert result.foundational.passed is True assert result.trusted.passed is True + assert result.certified.passed is False + + def test_low_mean_high_consistency_blocked_by_engine_gate(self) -> None: + """A skill with low variance but poor absolute scores still fails Certified. + + Consistency passes (low std_reward), but ADVANCED_AGENT_VALIDATION + requires engine score >= 0.8. This proves the two checks work together + as a safety net. + """ + low_score_engine = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.3, + mode=GateMode.BLOCK, + ) + behavioral_data = { + "std_reward": 0.02, + "edge_cases": {"total": 3, "passed": 3}, + } + result = compute_certification( + gates=[ + low_score_engine, + GateResult( + gate_name="security", + gate_type=GateType.SECURITY, + passed=True, + score=1.0, + mode=GateMode.BLOCK, + ), + GateResult( + gate_name="quality", + gate_type=GateType.QUALITY, + passed=True, + score=0.85, + mode=GateMode.WARN, + ), + ], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=self._passing_operational_policy(), + behavioral_data=behavioral_data, + ) + behavioral_check = next( + (c for c in result.certified.checks if c.check_id == CheckId.ENTERPRISE_BEHAVIORAL_TESTING), + None, + ) + assert behavioral_check is not None + assert behavioral_check.passed is True + + agent_check = next( + (c for c in result.certified.checks if c.check_id == CheckId.ADVANCED_AGENT_VALIDATION), + None, + ) + assert agent_check is not None + assert agent_check.passed is False + + assert result.certified.passed is False class TestEdgeCaseGate: @@ -562,54 +621,58 @@ def test_combines_std_reward_and_gate_edge_cases(self, tmp_path: Path) -> None: assert data["edge_cases"] == {"total": 3, "passed": 3} -class TestEdgeCaseScaffolding: - """Tests for edge case scaffolding.""" +class TestGenerateEdgeCaseEvals: + """Tests for generate_edge_case_evals.py.""" + + def test_generates_evals_from_edge_cases(self, tmp_path: Path) -> None: + from scripts.generate_edge_case_evals import generate_edge_case_evals - def _make_submission(self, tmp_path: Path) -> Path: sub = tmp_path / "my-skill" sub.mkdir() (sub / "metadata.yaml").write_text("name: my-skill\n") - (sub / "instruction.md").write_text("Main instruction") - skills = sub / "skills" - skills.mkdir() - (skills / "SKILL.md").write_text("# My Skill") - tests = sub / "tests" - tests.mkdir() - (tests / "test_outputs.py").write_text("pass") - - edge_cases = sub / "edge_cases" - edge_cases.mkdir() - (edge_cases / "empty_input.md").write_text("Test with empty input") - (edge_cases / "adversarial.md").write_text("Test with adversarial prompt") - return sub + edge_dir = sub / "edge_cases" + edge_dir.mkdir() + (edge_dir / "empty_input.md").write_text("Test with empty input") + (edge_dir / "adversarial.md").write_text("Try to break the skill") - def test_edge_case_dirs_created(self, tmp_path: Path) -> None: - from scripts.scaffold import scaffold_submission + result = generate_edge_case_evals(sub) + assert result is not None + assert result["skill_name"] == "my-skill" + assert len(result["evals"]) == 2 + assert result["evals"][0]["id"] == "edge-adversarial" + assert result["evals"][1]["id"] == "edge-empty_input" + assert "assertions" in result["evals"][0] - sub = self._make_submission(tmp_path) - output = tmp_path / "output" - scaffold_submission(sub, output) + def test_returns_none_without_edge_cases(self, tmp_path: Path) -> None: + from scripts.generate_edge_case_evals import generate_edge_case_evals - assert (output / "tasks-treatment-edge-adversarial" / "my-skill").is_dir() - assert (output / "tasks-treatment-edge-empty_input" / "my-skill").is_dir() + sub = tmp_path / "my-skill" + sub.mkdir() + result = generate_edge_case_evals(sub) + assert result is None - def test_edge_case_uses_alternative_instruction(self, tmp_path: Path) -> None: - from scripts.scaffold import scaffold_submission + def test_skips_empty_files(self, tmp_path: Path) -> None: + from scripts.generate_edge_case_evals import generate_edge_case_evals - sub = self._make_submission(tmp_path) - output = tmp_path / "output" - scaffold_submission(sub, output) + sub = tmp_path / "my-skill" + sub.mkdir() + (sub / "metadata.yaml").write_text("name: my-skill\n") + edge_dir = sub / "edge_cases" + edge_dir.mkdir() + (edge_dir / "empty.md").write_text("") + (edge_dir / "valid.md").write_text("A valid edge case") - edge_dir = output / "tasks-treatment-edge-empty_input" / "my-skill" - instruction = (edge_dir / "instruction.md").read_text() - assert instruction == "Test with empty input" + result = generate_edge_case_evals(sub) + assert result is not None + assert len(result["evals"]) == 1 + assert result["evals"][0]["id"] == "edge-valid" - def test_no_edge_cases_no_extra_dirs(self, tmp_path: Path) -> None: + def test_scaffold_does_not_create_edge_case_dirs(self, tmp_path: Path) -> None: from scripts.scaffold import scaffold_submission - sub = tmp_path / "no-edge" + sub = tmp_path / "my-skill" sub.mkdir() - (sub / "metadata.yaml").write_text("name: no-edge\n") + (sub / "metadata.yaml").write_text("name: my-skill\n") (sub / "instruction.md").write_text("Main instruction") skills = sub / "skills" skills.mkdir() @@ -617,6 +680,9 @@ def test_no_edge_cases_no_extra_dirs(self, tmp_path: Path) -> None: tests = sub / "tests" tests.mkdir() (tests / "test_outputs.py").write_text("pass") + edge_dir = sub / "edge_cases" + edge_dir.mkdir() + (edge_dir / "empty_input.md").write_text("Test with empty input") output = tmp_path / "output" scaffold_submission(sub, output) @@ -625,100 +691,51 @@ def test_no_edge_cases_no_extra_dirs(self, tmp_path: Path) -> None: assert edge_dirs == [] -class TestEdgeCaseAnalysis: - """Tests for edge case analysis in analyze.py.""" - - def test_missing_trial_output_counted_as_failure(self, tmp_path: Path) -> None: - from scripts.analyze import analyze_edge_cases +class TestEdgeCaseAggregation: + """Tests for ASE-based edge case aggregation.""" - edge_dir = tmp_path / "tasks-treatment-edge-empty_input" - edge_dir.mkdir() - results = analyze_edge_cases(tmp_path) - assert len(results) == 1 - assert results[0].name == "empty_input" - assert results[0].passed is False + def test_edge_case_result_model_with_score(self) -> None: + result = EdgeCaseResult(name="empty_input", passed=True, score=0.8) + assert result.name == "empty_input" + assert result.passed is True + assert result.summary is None + assert result.score == 0.8 - def test_edge_case_result_model(self) -> None: + def test_edge_case_result_model_with_summary(self) -> None: result = EdgeCaseResult( name="empty_input", summary=VariantSummary(n_trials=5, n_passed=4, pass_rate=0.8, mean_reward=0.7), passed=True, ) - assert result.name == "empty_input" - assert result.passed is True + assert result.summary is not None assert result.summary.pass_rate == 0.8 - def test_build_analysis_includes_edge_cases(self, tmp_path: Path) -> None: - from scripts.analyze import build_analysis - - results = tmp_path / "results" - treatment = results / "treatment" / "job" / "task__001" - treatment.mkdir(parents=True) - (treatment / "result.json").write_text( - json.dumps( - { - "verifier_result": {"reward": 0.8}, - } - ) - ) - control = results / "control" / "job" / "task__001" - control.mkdir(parents=True) - (control / "result.json").write_text( - json.dumps( - { - "verifier_result": {"reward": 0.3}, - } - ) - ) + def test_aggregate_per_edge_case_results(self, tmp_path: Path) -> None: + from scripts.aggregate_edge_case_evals import aggregate_edge_case_results - edge_root = tmp_path / "edge_results" - edge_dir = edge_root / "tasks-treatment-edge-empty_input" / "job" / "task__001" + edge_dir = tmp_path / "empty_input" / "iteration-1" / "eval-skill" / "with_skill" edge_dir.mkdir(parents=True) - (edge_dir / "result.json").write_text( + (edge_dir / "grading.json").write_text( json.dumps( { - "verifier_result": {"reward": 0.6}, + "summary": {"passed": 3, "failed": 0, "total": 3, "pass_rate": 1.0}, } ) ) - result = build_analysis( - results_dir=results, - submission_name="test-skill", - edge_case_results_dir=edge_root, - ) - assert len(result.edge_case_results) == 1 - assert result.edge_case_results[0].name == "empty_input" - assert result.edge_case_results[0].passed is True + results = aggregate_edge_case_results(tmp_path) + assert len(results) == 1 + assert results[0]["name"] == "empty_input" + assert results[0]["passed"] is True - def test_build_analysis_no_edge_cases(self, tmp_path: Path) -> None: - from scripts.analyze import build_analysis + def test_missing_grading_counted_as_failure(self, tmp_path: Path) -> None: + from scripts.aggregate_edge_case_evals import aggregate_edge_case_results - results = tmp_path / "results" - treatment = results / "treatment" / "job" / "task__001" - treatment.mkdir(parents=True) - (treatment / "result.json").write_text( - json.dumps( - { - "verifier_result": {"reward": 0.8}, - } - ) - ) - control = results / "control" / "job" / "task__001" - control.mkdir(parents=True) - (control / "result.json").write_text( - json.dumps( - { - "verifier_result": {"reward": 0.3}, - } - ) - ) - - result = build_analysis( - results_dir=results, - submission_name="test-skill", - ) - assert result.edge_case_results == [] + (tmp_path / "empty_input").mkdir() + results = aggregate_edge_case_results(tmp_path) + assert len(results) == 1 + assert results[0]["name"] == "empty_input" + assert results[0]["passed"] is False class TestLLMJudgeCriteria: @@ -739,78 +756,35 @@ def test_all_criteria_have_descriptions(self) -> None: class TestEdgeCaseEndToEnd: - """End-to-end: scaffold → mock results → analyze → certification.""" + """End-to-end: generate evals → mock results → certification.""" - def test_scaffold_analyze_certify(self, tmp_path: Path) -> None: - from scripts.analyze import build_analysis - from scripts.scaffold import scaffold_submission + def test_generate_evals_and_certify(self, tmp_path: Path) -> None: + from scripts.generate_edge_case_evals import generate_edge_case_evals # 1. Create submission with edge cases sub = tmp_path / "e2e-skill" sub.mkdir() (sub / "metadata.yaml").write_text("name: e2e-skill\n") - (sub / "instruction.md").write_text("Main instruction") - skills = sub / "skills" - skills.mkdir() - (skills / "SKILL.md").write_text("# E2E Skill") - tests = sub / "tests" - tests.mkdir() - (tests / "test_outputs.py").write_text("pass") edge_cases = sub / "edge_cases" edge_cases.mkdir() (edge_cases / "empty_input.md").write_text("Test with empty input") (edge_cases / "adversarial.md").write_text("Test with adversarial prompt") - # 2. Scaffold - output = tmp_path / "scaffold_output" - scaffold_submission(sub, output) - assert (output / "tasks-treatment-edge-adversarial" / "e2e-skill").is_dir() - assert (output / "tasks-treatment-edge-empty_input" / "e2e-skill").is_dir() - - # 3. Create mock Harbor results (2+ trials for std_reward computation) - results = tmp_path / "results" - for variant in ("treatment", "control"): - for i, reward in enumerate([0.8, 0.85] if variant == "treatment" else [0.3, 0.35]): - trial_dir = results / variant / "job" / f"task__{i:03d}" - trial_dir.mkdir(parents=True) - (trial_dir / "result.json").write_text( - json.dumps( - { - "verifier_result": {"reward": reward}, - } - ) - ) - - # 4. Create mock results for edge cases - edge_results = tmp_path / "edge_results" - for edge_name, reward in [("adversarial", 0.7), ("empty_input", 0.6)]: - trial_dir = edge_results / f"tasks-treatment-edge-{edge_name}" / "job" / "task__001" - trial_dir.mkdir(parents=True) - (trial_dir / "result.json").write_text( - json.dumps( - { - "verifier_result": {"reward": reward}, - } - ) - ) + # 2. Generate edge case evals + evals = generate_edge_case_evals(sub) + assert evals is not None + assert len(evals["evals"]) == 2 - # 5. Analyze with edge case results - analysis = build_analysis( - results_dir=results, - submission_name="e2e-skill", - edge_case_results_dir=edge_results, - ) - assert len(analysis.edge_case_results) == 2 - assert all(ec.passed for ec in analysis.edge_case_results) - - # 6. Verify behavioral check passes with the extracted data + # 3. Simulate edge case results (as pipeline would produce) behavioral_data = { - "std_reward": analysis.summary.treatment.std_reward, + "std_reward": 0.05, "edge_cases": { - "total": len(analysis.edge_case_results), - "passed": sum(1 for ec in analysis.edge_case_results if ec.passed), + "total": len(evals["evals"]), + "passed": len(evals["evals"]), }, } + + # 4. Verify behavioral check passes behavioral_check = _compute_behavioral_testing_check(behavioral_data) assert behavioral_check.passed is True assert behavioral_check.score > 0.0 diff --git a/tests/test_certification.py b/tests/test_certification.py index 9e0f2bd4..af517f98 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -353,9 +353,10 @@ def test_trusted_has_expected_checks(self) -> None: def test_certified_has_expected_checks(self) -> None: assert CheckId.ENTERPRISE_STRUCTURE_VALIDATION in CERTIFIED_CHECKS assert CheckId.ENTERPRISE_SECURITY_REVIEW in CERTIFIED_CHECKS + assert CheckId.ENTERPRISE_BEHAVIORAL_TESTING in CERTIFIED_CHECKS assert CheckId.ADVANCED_AGENT_VALIDATION in CERTIFIED_CHECKS - # behavioral_testing and continuous_optimization not yet wired into pipeline - assert len(CERTIFIED_CHECKS) == 3 + # continuous_optimization not yet implemented + assert len(CERTIFIED_CHECKS) == 4 class TestCertificationPolicy: From a64dabff79d98f79447ec833fae665c868008502 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 16 Jul 2026 11:56:38 +0300 Subject: [PATCH 07/10] fix: corrupt grading counted as failure, shared threshold, top-level yaml import - Corrupt grading.json now counts edge case as failed instead of skipping (prevents inflated pass rates) - Use shared DEFAULT_EDGE_CASE_PASS_THRESHOLD from certification.py instead of hardcoded 0.5 in aggregation script - Move yaml import to top level in generate_edge_case_evals.py Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/aggregate_edge_case_evals.py | 12 +++++++++--- scripts/generate_edge_case_evals.py | 6 ++---- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/scripts/aggregate_edge_case_evals.py b/scripts/aggregate_edge_case_evals.py index 25634c79..17bcfc7c 100644 --- a/scripts/aggregate_edge_case_evals.py +++ b/scripts/aggregate_edge_case_evals.py @@ -19,6 +19,8 @@ import sys from pathlib import Path +from abevalflow.certification import DEFAULT_EDGE_CASE_PASS_THRESHOLD + logger = logging.getLogger(__name__) @@ -55,6 +57,7 @@ def aggregate_edge_case_results(results_dir: Path) -> list[dict]: total_passed = 0 total_count = 0 + parse_failed = False for grading_path in with_skill_gradings: try: data = json.loads(grading_path.read_text()) @@ -62,13 +65,16 @@ def aggregate_edge_case_results(results_dir: Path) -> list[dict]: total_passed += summary.get("passed", 0) total_count += summary.get("total", 0) except (json.JSONDecodeError, OSError) as e: - logger.warning("Failed to parse %s: %s", grading_path, e) + logger.warning("Failed to parse %s — counted as failure: %s", grading_path, e) + parse_failed = True + break - if total_count == 0: + if parse_failed or total_count == 0: edge_results.append({"name": edge_name, "passed": False, "score": 0.0}) else: score = total_passed / total_count - edge_results.append({"name": edge_name, "passed": score >= 0.5, "score": score}) + passed = score >= DEFAULT_EDGE_CASE_PASS_THRESHOLD + edge_results.append({"name": edge_name, "passed": passed, "score": score}) return edge_results diff --git a/scripts/generate_edge_case_evals.py b/scripts/generate_edge_case_evals.py index 48c19b26..9c13c6d6 100644 --- a/scripts/generate_edge_case_evals.py +++ b/scripts/generate_edge_case_evals.py @@ -18,6 +18,8 @@ import sys from pathlib import Path +import yaml + logger = logging.getLogger(__name__) DEFAULT_ASSERTIONS = [ @@ -51,8 +53,6 @@ def generate_edge_case_evals( if skill_name is None: metadata_path = submission_dir / "metadata.yaml" if metadata_path.exists(): - import yaml - with metadata_path.open() as f: meta = yaml.safe_load(f) or {} skill_name = meta.get("name", "unknown-skill") @@ -148,8 +148,6 @@ def main(argv: list[str] | None = None) -> int: skill_name = args.skill_name or "unknown-skill" metadata_path = args.submission_dir / "metadata.yaml" if metadata_path.exists() and not args.skill_name: - import yaml - with metadata_path.open() as f: meta = yaml.safe_load(f) or {} skill_name = meta.get("name", skill_name) From 066421547909282a16406a1b578024c870bff16e Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 16 Jul 2026 12:15:23 +0300 Subject: [PATCH 08/10] fix: inline Python one-liner for YAML compatibility Multi-line Python block caused YAML parse error due to colons being interpreted as YAML mappings. Converted to one-liner matching existing enrichment step patterns. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../tasks/post/analyze-and-check-degradation.yaml | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/pipeline/tasks/post/analyze-and-check-degradation.yaml b/pipeline/tasks/post/analyze-and-check-degradation.yaml index 626fa340..aa4ea50d 100644 --- a/pipeline/tasks/post/analyze-and-check-degradation.yaml +++ b/pipeline/tasks/post/analyze-and-check-degradation.yaml @@ -246,16 +246,7 @@ spec: EDGE_RESULTS="$REPORT_DIR/edge-case-results.json" if [ -f "$EDGE_RESULTS" ]; then echo "Merging edge case results into report.json..." - python3 -c " -import json, sys -report_path = sys.argv[1] + '/report.json' -edge_path = sys.argv[2] -report = json.loads(open(report_path).read()) -edge = json.loads(open(edge_path).read()) -report['edge_case_results'] = edge -open(report_path, 'w').write(json.dumps(report, indent=2)) -print(f'Merged {len(edge)} edge case results') -" "$REPORT_DIR" "$EDGE_RESULTS" + python3 -c "import json,sys;d=sys.argv[1];e=sys.argv[2];r=json.loads(open(d+'/report.json').read());ec=json.loads(open(e).read());r['edge_case_results']=ec;open(d+'/report.json','w').write(json.dumps(r,indent=2));print(f'Merged {len(ec)} edge case results')" "$REPORT_DIR" "$EDGE_RESULTS" fi # --- 6. Extract results for Tekton --- From 0bc42983d9c831ecb4105d47f66c6029fe8ad782 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 16 Jul 2026 12:29:29 +0300 Subject: [PATCH 09/10] =?UTF-8?q?fix:=20address=20review=20=E2=80=94=20eva?= =?UTF-8?q?l-engine=20condition,=20profile=20comments,=20YAML=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Run edge case ASE for eval-engine=ase and both, not just harbor - Update agent/full profile comments: 'Enabled for skill profile only' - Fix inline Python YAML parse error (one-liner for compatibility) - PR description updated to document Certified breaking change Co-Authored-By: Claude Opus 4.6 (1M context) --- config/certification_profiles.yaml | 4 ++-- pipeline/tasks/phases/evaluate.yaml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index 82553cb6..f9ba3da1 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -86,7 +86,7 @@ profiles: certified: checks: - enterprise_structure_validation - # - enterprise_behavioral_testing # Implemented but not wired into pipeline yet + # - enterprise_behavioral_testing # Enabled for skill profile only - advanced_agent_validation # - safety_toxicity_bias_guardrails # Not yet implemented @@ -113,7 +113,7 @@ profiles: checks: - enterprise_structure_validation - enterprise_security_review - # - enterprise_behavioral_testing # Implemented but not wired into pipeline yet + # - enterprise_behavioral_testing # Enabled for skill profile only - advanced_agent_validation # Not yet implemented: # - continuous_optimization diff --git a/pipeline/tasks/phases/evaluate.yaml b/pipeline/tasks/phases/evaluate.yaml index 8d1942df..1f73e5a8 100644 --- a/pipeline/tasks/phases/evaluate.yaml +++ b/pipeline/tasks/phases/evaluate.yaml @@ -421,7 +421,7 @@ spec: set -euo pipefail EVAL_ENGINE="$(params.eval-engine)" - if [ "$EVAL_ENGINE" != "harbor" ] && [ "$EVAL_ENGINE" != "both" ]; then + if [ "$EVAL_ENGINE" != "harbor" ] && [ "$EVAL_ENGINE" != "ase" ] && [ "$EVAL_ENGINE" != "both" ]; then echo "Skipping edge case eval (eval-engine=$EVAL_ENGINE)" exit 0 fi @@ -525,7 +525,7 @@ spec: set -euo pipefail EVAL_ENGINE="$(params.eval-engine)" - if [ "$EVAL_ENGINE" != "harbor" ] && [ "$EVAL_ENGINE" != "both" ]; then + if [ "$EVAL_ENGINE" != "harbor" ] && [ "$EVAL_ENGINE" != "ase" ] && [ "$EVAL_ENGINE" != "both" ]; then exit 0 fi From 9403a749302d783b158bd15a25f4cf9dd85d91c5 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Thu, 16 Jul 2026 13:16:36 +0300 Subject: [PATCH 10/10] fix: generate edge case evals inline with node, no Python needed The ase-eval-edge-cases step uses a Node.js image which doesn't have python. Generate evals.json inline using node for JSON escaping. Co-Authored-By: Claude Opus 4.6 (1M context) --- pipeline/tasks/phases/evaluate.yaml | 30 ++++++++++++++++++++--------- 1 file changed, 21 insertions(+), 9 deletions(-) diff --git a/pipeline/tasks/phases/evaluate.yaml b/pipeline/tasks/phases/evaluate.yaml index 1f73e5a8..aa27f277 100644 --- a/pipeline/tasks/phases/evaluate.yaml +++ b/pipeline/tasks/phases/evaluate.yaml @@ -493,15 +493,27 @@ spec: mkdir -p "$EDGE_SKILL_DIR/evals" cp "$SKILL_MD" "$EDGE_SKILL_DIR/SKILL.md" - # Generate single-entry evals.json for this edge case - cd "$PIPELINE_DIR" - export PYTHONPATH="$PIPELINE_DIR" - python scripts/generate_edge_case_evals.py \ - --submission-dir "$SUBMISSION_PATH" \ - --output-dir "$EDGE_SKILL_DIR/evals" \ - --edge-case-file "$MD_FILE" - - mv "$EDGE_SKILL_DIR/evals/edge-case-evals.json" "$EDGE_SKILL_DIR/evals/evals.json" + # Generate single-entry evals.json using node (Python not available in this image) + node -e " + const fs = require('fs'); + const prompt = fs.readFileSync(process.argv[1], 'utf8').trim(); + const name = process.argv[2]; + const skill = process.argv[3]; + const evals = { + skill_name: skill, + evals: [{ + id: 'edge-' + name, + name: 'Edge case: ' + name, + prompt: prompt, + assertions: [ + 'The response handles the edge case appropriately without crashing or producing nonsensical output', + 'The response acknowledges limitations or uncertainty when the input is unusual', + 'The response remains helpful and coherent despite the edge case conditions' + ] + }] + }; + fs.writeFileSync(process.argv[4], JSON.stringify(evals, null, 2)); + " "$MD_FILE" "$EDGE_NAME" "$(params.submission-name)" "$EDGE_SKILL_DIR/evals/evals.json" # Run ASE for this single edge case EDGE_RESULTS="$RESULTS_BASE/$EDGE_NAME"