cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(validation): implement Fix-then-Revalidate orchestration loop for required validations #711

Merged
CoreRasurae merged 1 commits from feature/m6-fix-then-revalidate-loop into master 2026-03-24 23:12:37 +00:00
Conversation 17 Commits 1 Files Changed 13
+3120
13 changed files with 3120 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+14
View File
@@ -2,6 +2,19 @@
## Unreleased
- Added Fix-then-Revalidate orchestration loop for required validations:
bounded retry with configurable limits (0--100 per Safety Profile),
strategy revision escalation via ``auto_strategy_revision`` float
threshold, user escalation via ``needs_user_escalation`` result flag,
and domain events (``VALIDATION_FIX_ATTEMPTED``, ``VALIDATION_FIX_SUCCEEDED``,
``VALIDATION_FIX_EXHAUSTED``). Validation errors are treated as required
failures regardless of mode. Includes ``auto_validation_fix`` threshold,
per-resource retry tracking, early-exit signalling via ``None`` return from
``FixCallback``, event bus circuit breaker with lock-protected failure
counter, spec-required ``validation_summary`` and
``final_validation_results`` fields on the result model, DI container
registration per ADR-003, and structured logging via ``structlog``.
(#583)
- Added LSP resource types: `executable`, `lsp-server`, `lsp-workspace`,
`lsp-document` with parent/child hierarchy, auto-discovery rules, and
handler references. Registered in bootstrap, with YAML configurations,
@@ -361,6 +374,7 @@
`integration_tests` and `slow_integration_tests` sessions. Fixture files are
excluded from the main pabot runner via `tdd_fixture` tag. Includes 9 Robot
Framework integration test cases. (#628)
- Added TDD-style failing Behave BDD tests for the session list DI container
missing `db` provider bug. Three scenarios exercise `session list`,
`_get_session_service()`, and `session list --format json` through the real
benchmarks/bench_fix_revalidate.py
+245
View File
@@ -0,0 +1,245 @@
"""ASV benchmarks for the Fix-then-Revalidate orchestration loop.
Measures the performance of:
- Orchestrator creation
- Fix loop with immediate pass
- Fix loop with multiple retries
- Fix loop with terminal failure
- Model creation and serialization
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
try:
from cleveragents.application.services.fix_then_revalidate import (
FixAttemptRecord,
FixThenRevalidateOrchestrator,
FixThenRevalidateResult,
)
from cleveragents.application.services.validation_pipeline import (
ValidationPipeline,
ValidationResult,
)
from cleveragents.domain.models.core.tool import ValidationMode
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.fix_then_revalidate import (
FixAttemptRecord,
FixThenRevalidateOrchestrator,
FixThenRevalidateResult,
)
from cleveragents.application.services.validation_pipeline import (
ValidationPipeline,
ValidationResult,
)
from cleveragents.domain.models.core.tool import ValidationMode
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
return {"passed": True, "message": f"{validation_name} ok"}
def _make_failed_result(name: str) -> ValidationResult:
return ValidationResult(
validation_name=name,
resource_id="RES0001",
resource_name="bench-resource",
mode=ValidationMode.REQUIRED,
passed=False,
message=f"{name} failed",
data={"hint": "bench"},
duration_ms=1.0,
)
def _always_fix(result: ValidationResult) -> str:
return f"Fixed {result.validation_name}"
class _CountingRevalidate:
def __init__(self, pass_after: int) -> None:
self._pass_after = pass_after
self._count = 0
def __call__(self, result: ValidationResult) -> ValidationResult:
self._count += 1
passed = self._count >= self._pass_after
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=passed,
message="passed" if passed else "failing",
data=result.data,
duration_ms=1.0,
)
class _NeverPassRevalidate:
def __call__(self, result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=False,
message="failing",
data=result.data,
duration_ms=1.0,
)
def _build_pipeline() -> ValidationPipeline:
return ValidationPipeline(commands=[], executor=_mock_executor)
# ---------------------------------------------------------------------------
# Benchmark suites
# ---------------------------------------------------------------------------
class FixRevalidateCreationSuite:
"""Benchmarks for orchestrator instantiation."""
timeout = 60
def setup(self) -> None:
self.pipeline = _build_pipeline()
def time_create_orchestrator_default(self) -> None:
"""Benchmark creating an orchestrator with default settings."""
FixThenRevalidateOrchestrator(
validation_pipeline=self.pipeline,
)
def time_create_orchestrator_custom(self) -> None:
"""Benchmark creating an orchestrator with custom settings."""
FixThenRevalidateOrchestrator(
validation_pipeline=self.pipeline,
max_retries=5,
auto_strategy_revision=0.0,
)
class FixRevalidateLoopSuite:
"""Benchmarks for the fix loop execution."""
timeout = 60
def setup(self) -> None:
self.pipeline = _build_pipeline()
def time_fix_first_attempt(self) -> None:
"""Benchmark fix loop that succeeds on first attempt."""
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=self.pipeline,
max_retries=3,
)
failed = [_make_failed_result("bench-check")]
orchestrator.run_fix_loop(
plan_id="BENCH001",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=1),
)
def time_fix_on_second_attempt(self) -> None:
"""Benchmark fix loop that succeeds on the second attempt."""
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=self.pipeline,
max_retries=3,
)
failed = [_make_failed_result("bench-check")]
orchestrator.run_fix_loop(
plan_id="BENCH002",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=2),
)
def time_terminal_failure(self) -> None:
"""Benchmark fix loop that results in terminal failure."""
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=self.pipeline,
max_retries=3,
)
failed = [_make_failed_result("bench-check")]
orchestrator.run_fix_loop(
plan_id="BENCH003",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_NeverPassRevalidate(),
)
def time_empty_failures(self) -> None:
"""Benchmark fix loop with no failures (immediate pass)."""
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=self.pipeline,
max_retries=3,
)
orchestrator.run_fix_loop(
plan_id="BENCH004",
failed_results=[],
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=1),
)
class FixRevalidateModelSuite:
"""Benchmarks for model creation and serialization."""
timeout = 60
def time_create_attempt_record(self) -> None:
"""Benchmark creating a FixAttemptRecord."""
FixAttemptRecord(
attempt_number=1,
validation_name="bench-check",
fix_description="Applied fix",
success=True,
)
def time_create_result_model(self) -> None:
"""Benchmark creating a FixThenRevalidateResult."""
records = [
FixAttemptRecord(
attempt_number=i + 1,
validation_name=f"check-{i}",
fix_description=f"Fix {i}",
success=i == 2,
)
for i in range(3)
]
FixThenRevalidateResult(
validation_attempts=3,
final_passed=True,
validation_fix_history=records,
escalated=False,
terminal_failure=False,
)
def time_result_model_dump(self) -> None:
"""Benchmark serializing FixThenRevalidateResult to dict."""
records = [
FixAttemptRecord(
attempt_number=1,
validation_name="check-a",
fix_description="Fix a",
success=True,
),
]
result = FixThenRevalidateResult(
validation_attempts=1,
final_passed=True,
validation_fix_history=records,
)
result.model_dump()
features/fix_then_revalidate.feature
+61
View File
@@ -0,0 +1,61 @@
Feature: Fix-then-Revalidate orchestration loop
As an execution actor
I want a bounded retry loop that diagnoses, fixes, and re-validates
So that recoverable validation failures are resolved automatically
Background:
Given a fix-revalidate orchestrator with max_retries 3
# Happy path: first fix attempt resolves the validation failure
Scenario: Fix succeeds on first attempt
Given a required validation "check-syntax" that fails initially
And a fix callback that always succeeds
And a revalidate callback that passes after 1 fix
When the fix-revalidate loop runs for plan "PLAN001"
Then the fix-revalidate result should have validation_attempts 1
And the fix-revalidate result should have final_passed True
And the fix-revalidate result should not be escalated
And the fix-revalidate result should not be terminal failure
# Multiple attempts needed before the fix works
Scenario: Fix succeeds after multiple attempts
Given a required validation "check-lint" that fails initially
And a fix callback that always succeeds
And a revalidate callback that passes after 2 fixes
When the fix-revalidate loop runs for plan "PLAN002"
Then the fix-revalidate result should have validation_attempts 2
And the fix-revalidate result should have final_passed True
And the validation fix history should have 2 records
# Retry limit reached, auto_strategy_revision triggers escalation
Scenario: Retry limit exhausted with strategy revision
Given a fix-revalidate orchestrator with auto_strategy_revision enabled
And a required validation "check-security" that fails initially
And a fix callback that always succeeds
And a revalidate callback that never passes
When the fix-revalidate loop runs for plan "PLAN003"
Then the fix-revalidate result should have final_passed False
And the fix-revalidate result should be escalated
And the fix-revalidate result should not be terminal failure
And the fix-revalidate result should have validation_attempts 3
# Terminal failure when all attempts fail and no strategy revision
Scenario: Terminal failure when all attempts fail
Given a required validation "check-deploy" that fails initially
And a fix callback that always succeeds
And a revalidate callback that never passes
When the fix-revalidate loop runs for plan "PLAN004"
Then the fix-revalidate result should have final_passed False
And the fix-revalidate result should need user escalation
And the fix-revalidate result should not be terminal failure
And the validation fix history should have 3 records
# Informational validations do not trigger the fix loop
Scenario: Informational validations do not trigger fix loop
Given an informational validation "check-style" that fails
And a fix callback that always succeeds
And a revalidate callback that passes after 1 fix
When the fix-revalidate loop runs for plan "PLAN005"
Then the fix-revalidate result should have validation_attempts 0
And the fix-revalidate result should have final_passed True
And the validation fix history should have 0 records
features/fix_then_revalidate_coverage_boost.feature
+348
View File
@@ -0,0 +1,348 @@
Feature: Fix-then-Revalidate coverage boost
As a developer
I want to cover edge cases and validation paths in FixThenRevalidateOrchestrator
So that code coverage meets the 97% threshold
# Covers: __init__ validation
Scenario: Orchestrator rejects None validation_pipeline
When the fix-revalidate orchestrator is created with None pipeline
Then a fix-revalidate validation error should be raised with "validation_pipeline must not be None"
# Covers: __init__ validation
Scenario: Orchestrator rejects non-integer max_retries
When the fix-revalidate orchestrator is created with non-integer max_retries
Then a fix-revalidate validation error should be raised with "max_retries must be an integer"
# Covers: __init__ validation
Scenario: Orchestrator rejects max_retries below minimum
When the fix-revalidate orchestrator is created with max_retries -1
Then a fix-revalidate validation error should be raised with "max_retries must be between"
# Covers: __init__ validation
Scenario: Orchestrator rejects max_retries above maximum
When the fix-revalidate orchestrator is created with max_retries 101
Then a fix-revalidate validation error should be raised with "max_retries must be between"
# Covers: max_retries property
Scenario: Orchestrator exposes max_retries property
Given a fix-revalidate coverage orchestrator with max_retries 5
Then the fix-revalidate max_retries property should return 5
# Covers: auto_strategy_revision property
Scenario: Orchestrator exposes auto_strategy_revision property
Given a fix-revalidate coverage orchestrator with auto_strategy_revision enabled
Then the fix-revalidate auto_strategy_revision property should be 0.0
# Covers: get_retry_count empty plan_id
Scenario: get_retry_count rejects empty plan_id
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate get_retry_count is called with empty plan_id
Then a fix-revalidate validation error should be raised with "plan_id must not be empty"
# Covers: get_retry_count empty validation_name
Scenario: get_retry_count rejects empty validation_name
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate get_retry_count is called with empty validation_name
Then a fix-revalidate validation error should be raised with "validation_name must not be empty"
# Covers: get_retry_count returns count
Scenario: get_retry_count returns zero for unseen validation
Given a fix-revalidate coverage orchestrator with max_retries 3
Then the fix-revalidate get_retry_count for plan "P1" validation "check-a" should be 0
# Covers: reset_retry_counts empty plan_id
Scenario: reset_retry_counts rejects empty plan_id
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate reset_retry_counts is called with empty plan_id
Then a fix-revalidate validation error should be raised with "plan_id must not be empty"
# Covers: run_fix_loop empty plan_id
Scenario: run_fix_loop rejects empty plan_id
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate run_fix_loop is called with empty plan_id
Then a fix-revalidate validation error should be raised with "plan_id must not be empty"
# Covers: run_fix_loop None failed_results
Scenario: run_fix_loop rejects None failed_results
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate run_fix_loop is called with None failed_results
Then a fix-revalidate validation error should be raised with "failed_results must not be None"
# Covers: run_fix_loop None fix_callback
Scenario: run_fix_loop rejects None fix_callback
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate run_fix_loop is called with None fix_callback
Then a fix-revalidate validation error should be raised with "fix_callback must not be None"
# Covers: run_fix_loop None revalidate_callback
Scenario: run_fix_loop rejects None revalidate_callback
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate run_fix_loop is called with None revalidate_callback
Then a fix-revalidate validation error should be raised with "revalidate_callback must not be None"
# Covers: fix_callback exception handling
Scenario: fix_callback exception is treated as failed attempt
Given a fix-revalidate coverage orchestrator with max_retries 3
And a required coverage validation "check-syntax" that fails
And a fix-revalidate coverage fix callback that raises an exception
And a fix-revalidate coverage revalidate callback that never passes
When the fix-revalidate coverage loop runs for plan "P-EXC1"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should need user escalation
And the fix-revalidate coverage validation fix history should have 3 records
And each fix-revalidate coverage validation fix history record should have success False
# Covers: revalidate_callback exception handling
Scenario: revalidate_callback exception is treated as failed validation
Given a fix-revalidate coverage orchestrator with max_retries 3
And a required coverage validation "check-lint" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that raises an exception
When the fix-revalidate coverage loop runs for plan "P-EXC2"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should need user escalation
And the fix-revalidate coverage validation fix history should have 3 records
# Covers: multiple simultaneous validation failures
Scenario: Multiple validations fail simultaneously with mixed outcomes
Given a fix-revalidate coverage orchestrator with max_retries 3
And a required coverage validation "check-a" that fails
And a required coverage validation "check-b" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback where check-a passes after 1 and check-b never passes
When the fix-revalidate coverage loop runs for plan "P-MULTI"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should need user escalation
And the fix-revalidate coverage validation fix history should have 4 records
# Covers: reset_retry_counts then re-run
Scenario: Reset retry counts allows fresh retry budget
Given a fix-revalidate coverage orchestrator with max_retries 2
And a required coverage validation "check-retry" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that never passes
When the fix-revalidate coverage loop runs for plan "P-RESET"
And the fix-revalidate coverage retry counts are reset for plan "P-RESET"
And a fix-revalidate coverage revalidate callback that passes after 1
And the fix-revalidate coverage loop runs again for plan "P-RESET" with validation "check-retry"
Then the fix-revalidate coverage result should have final_passed True
And the fix-revalidate coverage result should have validation_attempts 1
# Covers: validation fix history record field validation
Scenario: Fix history records contain correct field values
Given a fix-revalidate coverage orchestrator with max_retries 3
And a required coverage validation "check-fields" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that passes after 2
When the fix-revalidate coverage loop runs for plan "P-FIELDS"
Then the fix-revalidate coverage validation fix history should have 2 records
And fix-revalidate coverage validation fix history record 1 should have validation_name "check-fields"
And fix-revalidate coverage validation fix history record 1 should have attempt_number 1
And fix-revalidate coverage validation fix history record 1 should have success False
And fix-revalidate coverage validation fix history record 2 should have success True
# Covers: max_retries=1 boundary
Scenario: Fix loop works with max_retries 1
Given a fix-revalidate coverage orchestrator with max_retries 1
And a required coverage validation "check-bound1" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that passes after 1
When the fix-revalidate coverage loop runs for plan "P-BOUND1"
Then the fix-revalidate coverage result should have final_passed True
And the fix-revalidate coverage result should have validation_attempts 1
# Covers: max_retries=10 boundary
Scenario: Fix loop works with max_retries 10
Given a fix-revalidate coverage orchestrator with max_retries 10
And a required coverage validation "check-bound10" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that passes after 9
When the fix-revalidate coverage loop runs for plan "P-BOUND10"
Then the fix-revalidate coverage result should have final_passed True
And the fix-revalidate coverage result should have validation_attempts 9
# Covers: event_bus property
Scenario: Orchestrator exposes event_bus property
Given a fix-revalidate coverage orchestrator with max_retries 3
Then the fix-revalidate event_bus property should be None
# Covers: exhausted re-invocation with auto-reset
Scenario: Second invocation for exhausted validation logs warning
Given a fix-revalidate coverage orchestrator with max_retries 1
And a required coverage validation "check-exhausted" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that never passes
When the fix-revalidate coverage loop runs for plan "P-EXHAUST"
And the fix-revalidate coverage loop runs again for plan "P-EXHAUST" with validation "check-exhausted"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should have validation_attempts 1
And the fix-revalidate coverage result should need user escalation
# Covers: bool type rejected for max_retries
Scenario: Orchestrator rejects boolean max_retries True
When the fix-revalidate orchestrator is created with boolean max_retries True
Then a fix-revalidate validation error should be raised with "max_retries must be an integer"
Scenario: Orchestrator rejects boolean max_retries False
When the fix-revalidate orchestrator is created with boolean max_retries False
Then a fix-revalidate validation error should be raised with "max_retries must be an integer"
# Covers: EventBus domain event emission
Scenario: EventBus receives VALIDATION_FIX_ATTEMPTED and SUCCEEDED events on fix success
Given a fix-revalidate coverage orchestrator with max_retries 3 and a mock event bus
And a required coverage validation "check-events" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that passes after 1
When the fix-revalidate coverage loop runs for plan "P-EVENTS"
Then the mock event bus should have received 2 events
And mock event bus event 1 should have event_type "validation.fix_attempted"
And mock event bus event 2 should have event_type "validation.fix_succeeded"
Scenario: EventBus receives VALIDATION_FIX_EXHAUSTED event on exhaustion
Given a fix-revalidate coverage orchestrator with max_retries 1 and a mock event bus
And a required coverage validation "check-exhaust-evt" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that never passes
When the fix-revalidate coverage loop runs for plan "P-EXHEVT"
Then the fix-revalidate coverage result should need user escalation
And the mock event bus should have received 2 events
And mock event bus event 1 should have event_type "validation.fix_attempted"
And mock event bus event 2 should have event_type "validation.fix_exhausted"
Scenario: EventBus emit exception does not crash the fix loop
Given a fix-revalidate coverage orchestrator with max_retries 1 and a failing event bus
And a required coverage validation "check-bus-err" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that passes after 1
When the fix-revalidate coverage loop runs for plan "P-BUSERR"
Then the fix-revalidate coverage result should have final_passed True
# Covers: FixAttemptRecord Pydantic field constraints
Scenario: FixAttemptRecord rejects attempt_number below 1
When a fix-revalidate FixAttemptRecord is created with attempt_number 0
Then a fix-revalidate pydantic validation error should be raised
Scenario: FixAttemptRecord rejects whitespace-only validation_name
When a fix-revalidate FixAttemptRecord is created with whitespace validation_name
Then a fix-revalidate pydantic validation error should be raised
# Covers: fix_description truncation to 2000 chars
Scenario: Fix callback returning over-length description is truncated
Given a fix-revalidate coverage orchestrator with max_retries 1
And a required coverage validation "check-trunc" that fails
And a fix-revalidate coverage fix callback that returns a 3000 char description
And a fix-revalidate coverage revalidate callback that passes after 1
When the fix-revalidate coverage loop runs for plan "P-TRUNC"
Then the fix-revalidate coverage result should have final_passed True
And the fix-revalidate coverage validation fix history should have 1 records
And fix-revalidate coverage validation fix history record 1 fix_description should be at most 2000 chars
# Covers: FixThenRevalidateResult model_validator
Scenario: FixThenRevalidateResult rejects escalated and terminal_failure both True
When a fix-revalidate FixThenRevalidateResult is created with escalated True and terminal_failure True
Then a fix-revalidate pydantic validation error should be raised
# Covers: validation_name max_length on FixAttemptRecord
Scenario: FixAttemptRecord rejects validation_name exceeding 255 chars
When a fix-revalidate FixAttemptRecord is created with validation_name exceeding 255 chars
Then a fix-revalidate pydantic validation error should be raised
# Covers: revalidate callback returning mismatched validation_name
Scenario: Revalidate callback returning wrong validation_name is treated as failure
Given a fix-revalidate coverage orchestrator with max_retries 2
And a required coverage validation "check-mismatch" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that returns a different validation_name
When the fix-revalidate coverage loop runs for plan "P-MISMATCH"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should need user escalation
# Covers: auto_strategy_revision type validation
Scenario: Orchestrator rejects boolean auto_strategy_revision
When the fix-revalidate orchestrator is created with boolean auto_strategy_revision
Then a fix-revalidate validation error should be raised with "auto_strategy_revision must be a float"
# Covers: auto_strategy_revision range validation
Scenario: Orchestrator rejects auto_strategy_revision out of range
When the fix-revalidate orchestrator is created with auto_strategy_revision 1.5
Then a fix-revalidate validation error should be raised with "auto_strategy_revision must be between"
# Covers: auto_validation_fix type validation
Scenario: Orchestrator rejects boolean auto_validation_fix
When the fix-revalidate orchestrator is created with boolean auto_validation_fix
Then a fix-revalidate validation error should be raised with "auto_validation_fix must be a float"
# Covers: auto_validation_fix range validation
Scenario: Orchestrator rejects auto_validation_fix out of range
When the fix-revalidate orchestrator is created with auto_validation_fix -0.1
Then a fix-revalidate validation error should be raised with "auto_validation_fix must be between"
# Covers: event_bus type validation
Scenario: Orchestrator rejects non-EventBus event_bus
When the fix-revalidate orchestrator is created with invalid event_bus type
Then a fix-revalidate validation error should be raised with "event_bus must be an EventBus"
# Covers: callable() validation on fix_callback
Scenario: run_fix_loop rejects non-callable fix_callback
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate run_fix_loop is called with non-callable fix_callback
Then a fix-revalidate validation error should be raised with "fix_callback must be callable"
# Covers: callable() validation on revalidate_callback
Scenario: run_fix_loop rejects non-callable revalidate_callback
Given a fix-revalidate coverage orchestrator with max_retries 3
When the fix-revalidate run_fix_loop is called with non-callable revalidate_callback
Then a fix-revalidate validation error should be raised with "revalidate_callback must be callable"
# Covers: fix_callback returning None signals unfixable
Scenario: Fix callback returning None escalates immediately
Given a fix-revalidate coverage orchestrator with max_retries 3
And a required coverage validation "check-unfixable" that fails
And a fix-revalidate coverage fix callback that returns None
And a fix-revalidate coverage revalidate callback that never passes
When the fix-revalidate coverage loop runs for plan "P-UNFIXABLE"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should need user escalation
And the fix-revalidate coverage validation fix history should have 1 records
# Covers: passed=True with error treated as failure
Scenario: Revalidate returning passed True with error is treated as failure
Given a fix-revalidate coverage orchestrator with max_retries 1
And a required coverage validation "check-contradictory" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that returns passed with error
When the fix-revalidate coverage loop runs for plan "P-CONTRADICT"
Then the fix-revalidate coverage result should have final_passed False
And the fix-revalidate coverage result should need user escalation
# Covers: final_passed=True with escalated=True rejected
Scenario: FixThenRevalidateResult rejects final_passed True with escalated True
When a fix-revalidate FixThenRevalidateResult is created with final_passed True and escalated True
Then a fix-revalidate pydantic validation error should be raised
# Covers: auto_validation_fix property accessor
Scenario: Orchestrator exposes auto_validation_fix property
Given a fix-revalidate coverage orchestrator with max_retries 3
Then the fix-revalidate auto_validation_fix property should be 0.0
# Covers: EventBus receives VALIDATION_FIX_EXHAUSTED on escalation path
Scenario: EventBus receives VALIDATION_FIX_EXHAUSTED event on escalation
Given a fix-revalidate coverage orchestrator with max_retries 1 and mock event bus and escalation
And a required coverage validation "check-esc-evt" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that never passes
When the fix-revalidate coverage loop runs for plan "P-ESCEVT"
Then the mock event bus should have received 2 events
And mock event bus event 1 should have event_type "validation.fix_attempted"
And mock event bus event 2 should have event_type "validation.fix_exhausted"
# Covers: mixed required and informational failures
Scenario: Mixed required and informational failures only fix required
Given a fix-revalidate coverage orchestrator with max_retries 3
And a required coverage validation "check-req" that fails
And an informational coverage validation "check-info" that fails
And a fix-revalidate coverage fix callback that always succeeds
And a fix-revalidate coverage revalidate callback that passes after 1
When the fix-revalidate coverage loop runs for plan "P-MIXED"
Then the fix-revalidate coverage result should have final_passed True
And the fix-revalidate coverage result should have validation_attempts 1
features/steps/fix_then_revalidate_coverage_boost_steps.py
+962
View File
@@ -0,0 +1,962 @@
"""Step definitions for Fix-then-Revalidate coverage boost scenarios.
Covers validation edge cases, property accessors, error paths,
callback exception handling, multi-failure scenarios, and boundary
tests to ensure >=97% coverage on fix_then_revalidate.py.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError as PydanticValidationError
from cleveragents.application.services.fix_then_revalidate import (
FixAttemptRecord,
FixThenRevalidateOrchestrator,
FixThenRevalidateResult,
)
from cleveragents.application.services.validation_pipeline import (
ValidationPipeline,
ValidationResult,
)
from cleveragents.core.exceptions import ValidationError
from cleveragents.domain.models.core.tool import ValidationMode
from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.types import EventType
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
return {"passed": True, "message": f"{validation_name} ok"}
def _make_pipeline() -> ValidationPipeline:
return ValidationPipeline(commands=[], executor=_mock_executor)
def _always_fix(result: ValidationResult) -> str:
return f"Fixed {result.validation_name}"
def _always_pass(result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=True,
message="passed",
data=result.data,
duration_ms=1.0,
)
def _make_failed_result(
name: str,
mode: ValidationMode = ValidationMode.REQUIRED,
) -> ValidationResult:
return ValidationResult(
validation_name=name,
resource_id="RES0001",
resource_name="test-resource",
mode=mode,
passed=False,
message=f"{name} failed: expected condition not met",
data={"hint": "check configuration"},
duration_ms=10.0,
)
def _raising_fix(result: ValidationResult) -> str:
raise RuntimeError(f"Fix callback crash for {result.validation_name}")
class _RaisingRevalidateCallback:
"""Revalidate callback that always raises."""
def __call__(self, result: ValidationResult) -> ValidationResult:
raise RuntimeError(f"Revalidate callback crash for {result.validation_name}")
class _NeverPassRevalidateCallback:
"""Revalidate callback that never passes."""
def __call__(self, result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=False,
message="still failing",
data=result.data,
duration_ms=5.0,
)
class _CountingRevalidateCallback:
"""Revalidate callback that passes after N calls."""
def __init__(self, pass_after: int) -> None:
self._pass_after = pass_after
self._count = 0
def __call__(self, result: ValidationResult) -> ValidationResult:
self._count += 1
passed = self._count >= self._pass_after
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=passed,
message=f"{'passed' if passed else 'still failing'} "
f"(attempt {self._count})",
data=result.data,
duration_ms=5.0,
)
class _PerValidationRevalidateCallback:
"""Revalidate callback with per-validation pass-after config."""
def __init__(self, config: dict[str, int | None]) -> None:
self._config = config
self._counts: dict[str, int] = {}
def __call__(self, result: ValidationResult) -> ValidationResult:
vname = result.validation_name
self._counts[vname] = self._counts.get(vname, 0) + 1
pass_after = self._config.get(vname)
passed = pass_after is not None and self._counts[vname] >= pass_after
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=passed,
message=f"{'passed' if passed else 'still failing'} "
f"(attempt {self._counts[vname]})",
data=result.data,
duration_ms=5.0,
)
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("a fix-revalidate coverage orchestrator with max_retries {n:d}")
def step_given_coverage_orchestrator(context: Context, n: int) -> None:
context.frc_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=n,
auto_strategy_revision=1.0,
)
context.frc_error = None
context.frc_failed_results = []
context.frc_fix_callback = _always_fix
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
@given("a fix-revalidate coverage orchestrator with auto_strategy_revision enabled")
def step_given_coverage_orchestrator_asr(context: Context) -> None:
context.frc_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=3,
auto_strategy_revision=0.0,
)
context.frc_error = None
context.frc_failed_results = []
context.frc_fix_callback = _always_fix
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
@given('a required coverage validation "{name}" that fails')
def step_given_required_coverage_validation(context: Context, name: str) -> None:
context.frc_failed_results.append(_make_failed_result(name))
@given("a fix-revalidate coverage fix callback that raises an exception")
def step_given_raising_fix_callback(context: Context) -> None:
context.frc_fix_callback = _raising_fix
@given("a fix-revalidate coverage fix callback that always succeeds")
def step_given_always_fix_callback(context: Context) -> None:
context.frc_fix_callback = _always_fix
@given("a fix-revalidate coverage revalidate callback that never passes")
def step_given_never_pass_revalidate(context: Context) -> None:
context.frc_revalidate_callback = _NeverPassRevalidateCallback()
@given("a fix-revalidate coverage revalidate callback that raises an exception")
def step_given_raising_revalidate(context: Context) -> None:
context.frc_revalidate_callback = _RaisingRevalidateCallback()
@given(
"a fix-revalidate coverage revalidate callback where "
"check-a passes after 1 and check-b never passes"
)
def step_given_per_validation_revalidate(context: Context) -> None:
context.frc_revalidate_callback = _PerValidationRevalidateCallback(
config={"check-a": 1, "check-b": None}
)
@given("a fix-revalidate coverage revalidate callback that passes after {n:d}")
def step_given_counting_revalidate(context: Context, n: int) -> None:
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=n)
# ---------------------------------------------------------------------------
# When steps (constructor validation)
# ---------------------------------------------------------------------------
@when("the fix-revalidate orchestrator is created with None pipeline")
def step_when_create_none_pipeline(context: Context) -> None:
context.frc_error = None
try:
bad_pipeline: Any = None
FixThenRevalidateOrchestrator(
validation_pipeline=bad_pipeline,
max_retries=3,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with non-integer max_retries")
def step_when_create_non_int_max_retries(context: Context) -> None:
context.frc_error = None
try:
bad_max_retries: Any = "three"
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=bad_max_retries,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with max_retries {n:d}")
def step_when_create_bad_max_retries(context: Context, n: int) -> None:
context.frc_error = None
try:
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=n,
)
except ValidationError as exc:
context.frc_error = str(exc)
# ---------------------------------------------------------------------------
# When steps (method validation)
# ---------------------------------------------------------------------------
@when("the fix-revalidate get_retry_count is called with empty plan_id")
def step_when_get_retry_count_empty_plan(context: Context) -> None:
context.frc_error = None
try:
context.frc_orchestrator.get_retry_count("", "check-a")
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate get_retry_count is called with empty validation_name")
def step_when_get_retry_count_empty_vname(context: Context) -> None:
context.frc_error = None
try:
context.frc_orchestrator.get_retry_count("P1", "")
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate reset_retry_counts is called with empty plan_id")
def step_when_reset_empty_plan(context: Context) -> None:
context.frc_error = None
try:
context.frc_orchestrator.reset_retry_counts("")
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate run_fix_loop is called with empty plan_id")
def step_when_run_loop_empty_plan(context: Context) -> None:
context.frc_error = None
try:
context.frc_orchestrator.run_fix_loop(
plan_id="",
failed_results=[],
fix_callback=_always_fix,
revalidate_callback=_always_pass,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate run_fix_loop is called with None failed_results")
def step_when_run_loop_none_results(context: Context) -> None:
context.frc_error = None
try:
bad_results: Any = None
context.frc_orchestrator.run_fix_loop(
plan_id="P1",
failed_results=bad_results,
fix_callback=_always_fix,
revalidate_callback=_always_pass,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate run_fix_loop is called with None fix_callback")
def step_when_run_loop_none_fix(context: Context) -> None:
context.frc_error = None
try:
bad_callback: Any = None
context.frc_orchestrator.run_fix_loop(
plan_id="P1",
failed_results=[],
fix_callback=bad_callback,
revalidate_callback=_always_pass,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate run_fix_loop is called with None revalidate_callback")
def step_when_run_loop_none_revalidate(context: Context) -> None:
context.frc_error = None
try:
bad_callback: Any = None
context.frc_orchestrator.run_fix_loop(
plan_id="P1",
failed_results=[],
fix_callback=_always_fix,
revalidate_callback=bad_callback,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when('the fix-revalidate coverage loop runs for plan "{plan_id}"')
def step_when_run_coverage_loop(context: Context, plan_id: str) -> None:
result: FixThenRevalidateResult = context.frc_orchestrator.run_fix_loop(
plan_id=plan_id,
failed_results=context.frc_failed_results,
fix_callback=context.frc_fix_callback,
revalidate_callback=context.frc_revalidate_callback,
)
context.frc_result = result
@when('the fix-revalidate coverage retry counts are reset for plan "{plan_id}"')
def step_when_reset_coverage_counts(context: Context, plan_id: str) -> None:
context.frc_orchestrator.reset_retry_counts(plan_id)
@when("a fix-revalidate coverage revalidate callback that passes after {n:d}")
def step_when_set_counting_revalidate(context: Context, n: int) -> None:
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=n)
@when(
'the fix-revalidate coverage loop runs again for plan "{plan_id}" '
'with validation "{vname}"'
)
def step_when_run_coverage_loop_again(
context: Context, plan_id: str, vname: str
) -> None:
failed_results = [_make_failed_result(vname)]
result: FixThenRevalidateResult = context.frc_orchestrator.run_fix_loop(
plan_id=plan_id,
failed_results=failed_results,
fix_callback=context.frc_fix_callback,
revalidate_callback=context.frc_revalidate_callback,
)
context.frc_result = result
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then('a fix-revalidate validation error should be raised with "{text}"')
def step_then_validation_error(context: Context, text: str) -> None:
assert context.frc_error is not None, (
"Expected a ValidationError but none was raised"
)
assert text in context.frc_error, (
f"Expected error containing '{text}', got '{context.frc_error}'"
)
@then("the fix-revalidate max_retries property should return {n:d}")
def step_then_max_retries_property(context: Context, n: int) -> None:
assert context.frc_orchestrator.max_retries == n, (
f"Expected max_retries={n}, got {context.frc_orchestrator.max_retries}"
)
@then("the fix-revalidate auto_strategy_revision property should be 0.0")
def step_then_asr_property(context: Context) -> None:
assert context.frc_orchestrator.auto_strategy_revision == 0.0, (
"Expected auto_strategy_revision=0.0"
)
@then(
'the fix-revalidate get_retry_count for plan "{plan_id}" '
'validation "{vname}" should be {n:d}'
)
def step_then_retry_count(context: Context, plan_id: str, vname: str, n: int) -> None:
count = context.frc_orchestrator.get_retry_count(plan_id, vname)
assert count == n, f"Expected retry count {n}, got {count}"
@then("the fix-revalidate coverage result should have final_passed {expected}")
def step_then_coverage_final_passed(context: Context, expected: str) -> None:
result: FixThenRevalidateResult = context.frc_result
expected_bool = expected.strip().lower() == "true"
assert result.final_passed is expected_bool, (
f"Expected final_passed={expected_bool}, got {result.final_passed}"
)
@then("the fix-revalidate coverage result should be terminal failure")
def step_then_coverage_terminal(context: Context) -> None:
result: FixThenRevalidateResult = context.frc_result
assert result.terminal_failure is True, "Expected terminal_failure=True"
@then("the fix-revalidate coverage result should not be terminal failure")
def step_then_coverage_not_terminal(context: Context) -> None:
result: FixThenRevalidateResult = context.frc_result
assert result.terminal_failure is False, "Expected terminal_failure=False"
@then("the fix-revalidate coverage result should need user escalation")
def step_then_coverage_needs_user_escalation(context: Context) -> None:
result: FixThenRevalidateResult = context.frc_result
assert result.needs_user_escalation is True, "Expected needs_user_escalation=True"
@then("the fix-revalidate coverage validation fix history should have {n:d} records")
def step_then_coverage_fix_history_count(context: Context, n: int) -> None:
result: FixThenRevalidateResult = context.frc_result
assert len(result.validation_fix_history) == n, (
f"Expected {n} fix history records, got {len(result.validation_fix_history)}"
)
@then(
"each fix-revalidate coverage validation fix history record "
"should have success False"
)
def step_then_all_records_failed(context: Context) -> None:
result: FixThenRevalidateResult = context.frc_result
for i, rec in enumerate(result.validation_fix_history):
assert rec.success is False, (
f"Record {i + 1} has success={rec.success}, expected False"
)
@then("the fix-revalidate coverage result should have validation_attempts {n:d}")
def step_then_coverage_total_attempts(context: Context, n: int) -> None:
result: FixThenRevalidateResult = context.frc_result
assert result.validation_attempts == n, (
f"Expected validation_attempts={n}, got {result.validation_attempts}"
)
@then(
"fix-revalidate coverage validation fix history record {idx:d} "
'should have validation_name "{name}"'
)
def step_then_record_vname(context: Context, idx: int, name: str) -> None:
result: FixThenRevalidateResult = context.frc_result
record = result.validation_fix_history[idx - 1]
assert record.validation_name == name, (
f"Record {idx} validation_name={record.validation_name}, expected {name}"
)
@then(
"fix-revalidate coverage validation fix history record {idx:d} "
"should have attempt_number {n:d}"
)
def step_then_record_attempt_number(context: Context, idx: int, n: int) -> None:
result: FixThenRevalidateResult = context.frc_result
record = result.validation_fix_history[idx - 1]
assert record.attempt_number == n, (
f"Record {idx} attempt_number={record.attempt_number}, expected {n}"
)
@then(
"fix-revalidate coverage validation fix history record {idx:d} "
"should have success {expected}"
)
def step_then_record_success(context: Context, idx: int, expected: str) -> None:
result: FixThenRevalidateResult = context.frc_result
record = result.validation_fix_history[idx - 1]
expected_bool = expected.strip().lower() == "true"
assert record.success is expected_bool, (
f"Record {idx} success={record.success}, expected {expected_bool}"
)
@then("the fix-revalidate event_bus property should be None")
def step_then_event_bus_none(context: Context) -> None:
assert context.frc_orchestrator.event_bus is None, "Expected event_bus=None"
# ---------------------------------------------------------------------------
# Mock EventBus for M4 review fix
# ---------------------------------------------------------------------------
class _MockEventBus:
"""Mock EventBus that records emitted events."""
def __init__(self) -> None:
self.events: list[DomainEvent] = []
def emit(self, event: DomainEvent) -> None:
self.events.append(event)
def subscribe(
self,
event_type: EventType,
handler: Any,
) -> None:
pass # Not needed for this test
class _FailingEventBus:
"""Mock EventBus whose emit always raises."""
def emit(self, event: DomainEvent) -> None:
raise RuntimeError("EventBus emit failure")
def subscribe(
self,
event_type: EventType,
handler: Any,
) -> None:
pass
# ---------------------------------------------------------------------------
# Given steps for new review-fix scenarios
# ---------------------------------------------------------------------------
@given(
"a fix-revalidate coverage orchestrator with max_retries {n:d} and a mock event bus"
)
def step_given_coverage_orchestrator_with_mock_bus(context: Context, n: int) -> None:
context.frc_mock_event_bus = _MockEventBus()
context.frc_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=n,
auto_strategy_revision=1.0,
event_bus=context.frc_mock_event_bus,
)
context.frc_error = None
context.frc_failed_results = []
context.frc_fix_callback = _always_fix
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
@given(
"a fix-revalidate coverage orchestrator with max_retries {n:d} "
"and a failing event bus"
)
def step_given_coverage_orchestrator_with_failing_bus(context: Context, n: int) -> None:
context.frc_mock_event_bus = _FailingEventBus()
context.frc_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
auto_strategy_revision=1.0,
max_retries=n,
event_bus=context.frc_mock_event_bus,
)
context.frc_error = None
context.frc_failed_results = []
context.frc_fix_callback = _always_fix
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
# ---------------------------------------------------------------------------
# When steps for new review-fix scenarios
# ---------------------------------------------------------------------------
@when("the fix-revalidate orchestrator is created with boolean max_retries True")
def step_when_create_bool_true_max_retries(context: Context) -> None:
context.frc_error = None
try:
bad_retries: Any = True
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=bad_retries,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with boolean max_retries False")
def step_when_create_bool_false_max_retries(context: Context) -> None:
context.frc_error = None
try:
bad_retries: Any = False
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=bad_retries,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("a fix-revalidate FixAttemptRecord is created with attempt_number 0")
def step_when_create_record_bad_attempt(context: Context) -> None:
context.frc_pydantic_error = None
try:
FixAttemptRecord(
attempt_number=0,
validation_name="check-a",
fix_description="test",
success=False,
)
except PydanticValidationError as exc:
context.frc_pydantic_error = str(exc)
@when("a fix-revalidate FixAttemptRecord is created with whitespace validation_name")
def step_when_create_record_whitespace_vname(context: Context) -> None:
context.frc_pydantic_error = None
try:
FixAttemptRecord(
attempt_number=1,
validation_name=" ",
fix_description="test",
success=False,
)
except PydanticValidationError as exc:
context.frc_pydantic_error = str(exc)
# ---------------------------------------------------------------------------
# Then steps for new review-fix scenarios
# ---------------------------------------------------------------------------
@then("the mock event bus should have received {n:d} events")
def step_then_mock_bus_event_count(context: Context, n: int) -> None:
bus: _MockEventBus = context.frc_mock_event_bus
assert len(bus.events) == n, (
f"Expected {n} events, got {len(bus.events)}: "
f"{[e.event_type for e in bus.events]}"
)
@then('mock event bus event {idx:d} should have event_type "{etype}"')
def step_then_mock_bus_event_type(context: Context, idx: int, etype: str) -> None:
bus: _MockEventBus = context.frc_mock_event_bus
event = bus.events[idx - 1]
assert event.event_type.value == etype, (
f"Event {idx} type={event.event_type.value}, expected {etype}"
)
@then("a fix-revalidate pydantic validation error should be raised")
def step_then_pydantic_error(context: Context) -> None:
assert context.frc_pydantic_error is not None, (
"Expected a Pydantic ValidationError but none was raised"
)
# ---------------------------------------------------------------------------
# Steps for B3, A1, R1, B2 review fixes
# ---------------------------------------------------------------------------
def _long_description_fix(result: ValidationResult) -> str:
"""Fix callback that returns a description exceeding 2000 chars."""
return "x" * 3000
class _MismatchedRevalidateCallback:
"""Revalidate callback that returns a result with the wrong name."""
def __call__(self, result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name="wrong-validation-name",
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=True,
message="passed but with wrong name",
data=result.data,
duration_ms=1.0,
)
@given("a fix-revalidate coverage fix callback that returns a 3000 char description")
def step_given_long_fix_callback(context: Context) -> None:
context.frc_fix_callback = _long_description_fix
@given(
"a fix-revalidate coverage revalidate callback "
"that returns a different validation_name"
)
def step_given_mismatched_revalidate(context: Context) -> None:
context.frc_revalidate_callback = _MismatchedRevalidateCallback()
@when(
"a fix-revalidate FixThenRevalidateResult is created with "
"escalated True and terminal_failure True"
)
def step_when_create_result_both_flags(context: Context) -> None:
context.frc_pydantic_error = None
try:
FixThenRevalidateResult(
validation_attempts=3,
final_passed=False,
escalated=True,
terminal_failure=True,
)
except PydanticValidationError as exc:
context.frc_pydantic_error = str(exc)
@when(
"a fix-revalidate FixAttemptRecord is created "
"with validation_name exceeding 255 chars"
)
def step_when_create_record_long_vname(context: Context) -> None:
context.frc_pydantic_error = None
try:
FixAttemptRecord(
attempt_number=1,
validation_name="v" * 256,
fix_description="test",
success=False,
)
except PydanticValidationError as exc:
context.frc_pydantic_error = str(exc)
@then(
"fix-revalidate coverage validation fix history record {idx:d} "
"fix_description should be at most {n:d} chars"
)
def step_then_record_fix_description_length(context: Context, idx: int, n: int) -> None:
result: FixThenRevalidateResult = context.frc_result
record = result.validation_fix_history[idx - 1]
assert len(record.fix_description) <= n, (
f"Record {idx} fix_description has {len(record.fix_description)} chars, "
f"expected at most {n}"
)
# ---------------------------------------------------------------------------
# New steps for review-fix round 2/3 coverage
# ---------------------------------------------------------------------------
@when("the fix-revalidate orchestrator is created with boolean auto_strategy_revision")
def step_when_create_bool_asr(context: Context) -> None:
context.frc_error = None
try:
bad_asr: Any = True
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=3,
auto_strategy_revision=bad_asr,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with auto_strategy_revision {val}")
def step_when_create_bad_asr_range(context: Context, val: str) -> None:
context.frc_error = None
try:
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=3,
auto_strategy_revision=float(val),
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with boolean auto_validation_fix")
def step_when_create_bool_avf(context: Context) -> None:
context.frc_error = None
try:
bad_avf: Any = True
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=3,
auto_validation_fix=bad_avf,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with auto_validation_fix {val}")
def step_when_create_bad_avf_range(context: Context, val: str) -> None:
context.frc_error = None
try:
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=3,
auto_validation_fix=float(val),
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate orchestrator is created with invalid event_bus type")
def step_when_create_bad_event_bus(context: Context) -> None:
context.frc_error = None
try:
bad_bus: Any = 42
FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=3,
event_bus=bad_bus,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate run_fix_loop is called with non-callable fix_callback")
def step_when_run_loop_non_callable_fix(context: Context) -> None:
context.frc_error = None
try:
bad_fix: Any = "not_callable"
context.frc_orchestrator.run_fix_loop(
plan_id="P1",
failed_results=[],
fix_callback=bad_fix,
revalidate_callback=_always_pass,
)
except ValidationError as exc:
context.frc_error = str(exc)
@when("the fix-revalidate run_fix_loop is called with non-callable revalidate_callback")
def step_when_run_loop_non_callable_revalidate(context: Context) -> None:
context.frc_error = None
try:
bad_revalidate: Any = "not_callable"
context.frc_orchestrator.run_fix_loop(
plan_id="P1",
failed_results=[],
fix_callback=_always_fix,
revalidate_callback=bad_revalidate,
)
except ValidationError as exc:
context.frc_error = str(exc)
def _none_fix(result: ValidationResult) -> None:
"""Fix callback that returns None to signal unfixable."""
return None
@given("a fix-revalidate coverage fix callback that returns None")
def step_given_none_fix_callback(context: Context) -> None:
context.frc_fix_callback = _none_fix
class _PassedWithErrorRevalidateCallback:
"""Revalidate callback that returns passed=True with an error set."""
def __call__(self, result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=True,
message="passed but has error",
data=result.data,
duration_ms=1.0,
error="runtime error in validation",
)
@given("a fix-revalidate coverage revalidate callback that returns passed with error")
def step_given_passed_with_error_revalidate(context: Context) -> None:
context.frc_revalidate_callback = _PassedWithErrorRevalidateCallback()
@when(
"a fix-revalidate FixThenRevalidateResult is created "
"with final_passed True and escalated True"
)
def step_when_create_result_final_passed_escalated(context: Context) -> None:
context.frc_pydantic_error = None
try:
FixThenRevalidateResult(
validation_attempts=1,
final_passed=True,
escalated=True,
)
except PydanticValidationError as exc:
context.frc_pydantic_error = str(exc)
@then("the fix-revalidate auto_validation_fix property should be {val}")
def step_then_avf_property(context: Context, val: str) -> None:
expected = float(val)
actual = context.frc_orchestrator.auto_validation_fix
assert actual == expected, f"Expected auto_validation_fix={expected}, got {actual}"
@given(
"a fix-revalidate coverage orchestrator with max_retries {n:d} "
"and mock event bus and escalation"
)
def step_given_coverage_orchestrator_mock_bus_escalation(
context: Context, n: int
) -> None:
context.frc_mock_event_bus = _MockEventBus()
context.frc_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=_make_pipeline(),
max_retries=n,
auto_strategy_revision=0.0,
event_bus=context.frc_mock_event_bus,
)
context.frc_error = None
context.frc_failed_results = []
context.frc_fix_callback = _always_fix
context.frc_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
@given('an informational coverage validation "{name}" that fails')
def step_given_informational_coverage_validation(context: Context, name: str) -> None:
context.frc_failed_results.append(
_make_failed_result(name, ValidationMode.INFORMATIONAL)
)
features/steps/fix_then_revalidate_steps.py
+246
View File
@@ -0,0 +1,246 @@
"""Step definitions for Fix-then-Revalidate orchestration scenarios.
All step names are prefixed with 'fix-revalidate' context to avoid
AmbiguousStep collisions with existing step definitions.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.fix_then_revalidate import (
FixThenRevalidateOrchestrator,
FixThenRevalidateResult,
)
from cleveragents.application.services.validation_pipeline import (
ValidationPipeline,
ValidationResult,
)
from cleveragents.domain.models.core.tool import ValidationMode
# ---------------------------------------------------------------------------
# Mock helpers
# ---------------------------------------------------------------------------
def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
"""Always-passing mock executor for pipeline construction."""
return {"passed": True, "message": f"{validation_name} ok"}
def _make_failed_result(
name: str,
mode: ValidationMode = ValidationMode.REQUIRED,
) -> ValidationResult:
"""Create a failed ValidationResult for testing."""
return ValidationResult(
validation_name=name,
resource_id="RES0001",
resource_name="test-resource",
mode=mode,
passed=False,
message=f"{name} failed: expected condition not met",
data={"hint": "check configuration"},
duration_ms=10.0,
)
class _CountingRevalidateCallback:
"""Revalidate callback that passes after a configurable number of fixes."""
def __init__(self, pass_after: int) -> None:
self._pass_after = pass_after
self._call_count = 0
def __call__(self, result: ValidationResult) -> ValidationResult:
self._call_count += 1
passed = self._call_count >= self._pass_after
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=passed,
message=f"{'passed' if passed else 'still failing'} "
f"(attempt {self._call_count})",
data=result.data,
duration_ms=5.0,
)
class _NeverPassRevalidateCallback:
"""Revalidate callback that never passes."""
def __call__(self, result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=False,
message="still failing",
data=result.data,
duration_ms=5.0,
)
def _always_fix(result: ValidationResult) -> str:
"""Fix callback that always returns a fix description."""
return f"Applied fix for {result.validation_name}: adjusted configuration"
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("a fix-revalidate orchestrator with max_retries {n:d}")
def step_given_orchestrator(context: Context, n: int) -> None:
pipeline = ValidationPipeline(
commands=[],
executor=_mock_executor,
)
context.fr_pipeline = pipeline
context.fr_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=n,
auto_strategy_revision=1.0,
)
context.fr_failed_results = [] # list[ValidationResult]
context.fr_fix_callback = _always_fix
context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
@given("a fix-revalidate orchestrator with auto_strategy_revision enabled")
def step_given_orchestrator_auto_revision(context: Context) -> None:
pipeline = ValidationPipeline(
commands=[],
executor=_mock_executor,
)
context.fr_pipeline = pipeline
context.fr_orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
auto_strategy_revision=0.0,
)
context.fr_failed_results = []
context.fr_fix_callback = _always_fix
context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=1)
@given('a required validation "{name}" that fails initially')
def step_given_required_validation_fails(context: Context, name: str) -> None:
result = _make_failed_result(name, ValidationMode.REQUIRED)
context.fr_failed_results.append(result)
@given('an informational validation "{name}" that fails')
def step_given_informational_validation_fails(context: Context, name: str) -> None:
result = _make_failed_result(name, ValidationMode.INFORMATIONAL)
context.fr_failed_results.append(result)
@given("a fix callback that always succeeds")
def step_given_fix_callback(context: Context) -> None:
context.fr_fix_callback = _always_fix
@given("a revalidate callback that passes after {n:d} fix")
def step_given_revalidate_passes_after_n(context: Context, n: int) -> None:
context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=n)
@given("a revalidate callback that passes after {n:d} fixes")
def step_given_revalidate_passes_after_n_plural(context: Context, n: int) -> None:
context.fr_revalidate_callback = _CountingRevalidateCallback(pass_after=n)
@given("a revalidate callback that never passes")
def step_given_revalidate_never_passes(context: Context) -> None:
context.fr_revalidate_callback = _NeverPassRevalidateCallback()
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when('the fix-revalidate loop runs for plan "{plan_id}"')
def step_when_run_fix_loop(context: Context, plan_id: str) -> None:
orchestrator: FixThenRevalidateOrchestrator = context.fr_orchestrator
result: FixThenRevalidateResult = orchestrator.run_fix_loop(
plan_id=plan_id,
failed_results=context.fr_failed_results,
fix_callback=context.fr_fix_callback,
revalidate_callback=context.fr_revalidate_callback,
)
context.fr_result = result
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the fix-revalidate result should have validation_attempts {n:d}")
def step_then_total_attempts(context: Context, n: int) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.validation_attempts == n, (
f"Expected validation_attempts={n}, got {result.validation_attempts}"
)
@then("the fix-revalidate result should have final_passed {expected}")
def step_then_final_passed(context: Context, expected: str) -> None:
result: FixThenRevalidateResult = context.fr_result
expected_bool = expected.strip().lower() == "true"
assert result.final_passed is expected_bool, (
f"Expected final_passed={expected_bool}, got {result.final_passed}"
)
@then("the fix-revalidate result should not be escalated")
def step_then_not_escalated(context: Context) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.escalated is False, "Expected escalated=False"
@then("the fix-revalidate result should be escalated")
def step_then_escalated(context: Context) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.escalated is True, "Expected escalated=True"
@then("the fix-revalidate result should not be terminal failure")
def step_then_not_terminal(context: Context) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.terminal_failure is False, "Expected terminal_failure=False"
@then("the fix-revalidate result should be terminal failure")
def step_then_terminal(context: Context) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.terminal_failure is True, "Expected terminal_failure=True"
@then("the validation fix history should have {n:d} records")
def step_then_fix_history_count(context: Context, n: int) -> None:
result: FixThenRevalidateResult = context.fr_result
assert len(result.validation_fix_history) == n, (
f"Expected {n} fix history records, got {len(result.validation_fix_history)}"
)
@then("the fix-revalidate result should need user escalation")
def step_then_needs_user_escalation(context: Context) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.needs_user_escalation is True, "Expected needs_user_escalation=True"
@then("the fix-revalidate result should not need user escalation")
def step_then_not_needs_user_escalation(context: Context) -> None:
result: FixThenRevalidateResult = context.fr_result
assert result.needs_user_escalation is False, "Expected needs_user_escalation=False"
robot/fix_then_revalidate_integration.robot
+59
View File
@@ -0,0 +1,59 @@
*** Settings ***
Documentation Integration tests for Fix-then-Revalidate orchestration loop
Library Process
Library OperatingSystem
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_fix_then_revalidate.py
*** Test Cases ***
Fix Then Revalidate Models Are Valid
[Documentation] Verify FixAttemptRecord and FixThenRevalidateResult Pydantic models
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} models cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} models-ok
Fix Succeeds On First Attempt
[Documentation] Fix loop resolves validation on first fix attempt
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} fix_first cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} fix-first-ok
Fix Succeeds After Multiple Attempts
[Documentation] Fix loop resolves validation after multiple fix attempts
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} fix_multiple cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} fix-multiple-ok
Terminal Failure When All Attempts Exhausted
[Documentation] Plan reaches terminal failure when retries exhausted
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} terminal_failure cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} terminal-failure-ok
Escalation To Strategy Revision
[Documentation] Loop escalates to strategy revision when auto_strategy_revision enabled
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} escalation cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} escalation-ok
Informational Validations Skipped
[Documentation] Informational validations do not trigger fix loop
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} informational_skipped cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} informational-skipped-ok
Retry Counting Per Validation Per Plan
[Documentation] Verify retry counting and reset per validation per plan
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} retry_counting cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} retry-counting-ok
Unknown Command Returns Error
[Documentation] Running with unknown command exits with code 1
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} nonexistent_cmd cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 1
Should Contain ${result.stderr} Unknown command
robot/helper_fix_then_revalidate.py
+304
View File
@@ -0,0 +1,304 @@
"""Helper script for Robot Framework fix-then-revalidate integration tests.
Self-contained Python helper that exercises the FixThenRevalidateOrchestrator
and its models, printing sentinel strings on success and exiting
with code 1 on failure.
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.fix_then_revalidate import (
FixAttemptRecord,
FixThenRevalidateOrchestrator,
FixThenRevalidateResult,
)
from cleveragents.application.services.validation_pipeline import (
ValidationPipeline,
ValidationResult,
)
from cleveragents.domain.models.core.tool import ValidationMode
# ---------------------------------------------------------------------------
# Mock helpers
# ---------------------------------------------------------------------------
def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
return {"passed": True, "message": f"{validation_name} ok"}
def _make_failed_result(
name: str,
mode: ValidationMode = ValidationMode.REQUIRED,
) -> ValidationResult:
return ValidationResult(
validation_name=name,
resource_id="RES0001",
resource_name="test-resource",
mode=mode,
passed=False,
message=f"{name} failed",
data={"hint": "check config"},
duration_ms=10.0,
)
def _always_fix(result: ValidationResult) -> str:
return f"Fixed {result.validation_name}"
class _CountingRevalidate:
def __init__(self, pass_after: int) -> None:
self._pass_after = pass_after
self._count = 0
def __call__(self, result: ValidationResult) -> ValidationResult:
self._count += 1
passed = self._count >= self._pass_after
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=passed,
message="passed" if passed else "still failing",
data=result.data,
duration_ms=5.0,
)
class _NeverPassRevalidate:
def __call__(self, result: ValidationResult) -> ValidationResult:
return ValidationResult(
validation_name=result.validation_name,
resource_id=result.resource_id,
resource_name=result.resource_name,
mode=result.mode,
passed=False,
message="still failing",
data=result.data,
duration_ms=5.0,
)
def _check(condition: bool, message: str) -> None:
"""Explicit check that works even with Python -O optimization."""
if not condition:
raise AssertionError(message)
# ---------------------------------------------------------------------------
# Sub-commands
# ---------------------------------------------------------------------------
def test_models() -> None:
"""Verify FixAttemptRecord and FixThenRevalidateResult models."""
record = FixAttemptRecord(
attempt_number=1,
validation_name="check-syntax",
fix_description="Adjusted indentation",
success=True,
)
_check(record.attempt_number == 1, "attempt_number should be 1")
_check(record.success is True, "success should be True")
result = FixThenRevalidateResult(
validation_attempts=2,
final_passed=True,
validation_fix_history=[record],
escalated=False,
terminal_failure=False,
)
_check(result.validation_attempts == 2, "validation_attempts should be 2")
_check(result.final_passed is True, "final_passed should be True")
_check(
len(result.validation_fix_history) == 1,
"validation_fix_history should have 1 record",
)
print("models-ok")
def test_fix_succeeds_first_attempt() -> None:
"""Fix succeeds on first attempt."""
pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
)
failed = [_make_failed_result("check-syntax")]
result = orchestrator.run_fix_loop(
plan_id="PLAN001",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=1),
)
_check(result.final_passed is True, "final_passed should be True")
_check(result.validation_attempts == 1, "validation_attempts should be 1")
_check(
len(result.validation_fix_history) == 1,
"validation_fix_history should have 1 record",
)
_check(
result.validation_fix_history[0].success is True,
"first fix history record should be success",
)
print("fix-first-ok")
def test_fix_succeeds_after_multiple() -> None:
"""Fix succeeds after multiple attempts."""
pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
)
failed = [_make_failed_result("check-lint")]
result = orchestrator.run_fix_loop(
plan_id="PLAN002",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=2),
)
_check(result.final_passed is True, "final_passed should be True")
_check(result.validation_attempts == 2, "validation_attempts should be 2")
print("fix-multiple-ok")
def test_terminal_failure() -> None:
"""Terminal failure when all attempts exhausted and no strategy revision."""
pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
auto_strategy_revision=1.0,
)
failed = [_make_failed_result("check-deploy")]
result = orchestrator.run_fix_loop(
plan_id="PLAN003",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_NeverPassRevalidate(),
)
_check(result.final_passed is False, "final_passed should be False")
_check(result.needs_user_escalation is True, "needs_user_escalation should be True")
_check(result.escalated is False, "escalated should be False")
print("terminal-failure-ok")
def test_escalation() -> None:
"""Escalation to strategy revision when auto_strategy_revision is 0.0."""
pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
auto_strategy_revision=0.0,
)
failed = [_make_failed_result("check-security")]
result = orchestrator.run_fix_loop(
plan_id="PLAN004",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_NeverPassRevalidate(),
)
_check(result.final_passed is False, "final_passed should be False")
_check(result.escalated is True, "escalated should be True")
_check(result.terminal_failure is False, "terminal_failure should be False")
print("escalation-ok")
def test_informational_skipped() -> None:
"""Informational validations are skipped by the fix loop."""
pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
)
failed = [_make_failed_result("check-style", ValidationMode.INFORMATIONAL)]
result = orchestrator.run_fix_loop(
plan_id="PLAN005",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=1),
)
_check(result.final_passed is True, "final_passed should be True")
_check(result.validation_attempts == 0, "validation_attempts should be 0")
_check(
len(result.validation_fix_history) == 0,
"validation_fix_history should be empty",
)
print("informational-skipped-ok")
def test_retry_counting() -> None:
"""Verify retry counting per validation per plan."""
pipeline = ValidationPipeline(commands=[], executor=_mock_executor)
orchestrator = FixThenRevalidateOrchestrator(
validation_pipeline=pipeline,
max_retries=3,
)
_check(
orchestrator.get_retry_count("P1", "check-a") == 0,
"initial retry count should be 0",
)
failed = [_make_failed_result("check-a")]
orchestrator.run_fix_loop(
plan_id="P1",
failed_results=failed,
fix_callback=_always_fix,
revalidate_callback=_CountingRevalidate(pass_after=2),
)
_check(
orchestrator.get_retry_count("P1", "check-a") == 0,
"retry count should be 0 after run (auto-reset)",
)
orchestrator.reset_retry_counts("P1")
_check(
orchestrator.get_retry_count("P1", "check-a") == 0,
"retry count should be 0 after explicit reset (no-op)",
)
print("retry-counting-ok")
# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------
COMMANDS = {
"models": test_models,
"fix_first": test_fix_succeeds_first_attempt,
"fix_multiple": test_fix_succeeds_after_multiple,
"terminal_failure": test_terminal_failure,
"escalation": test_escalation,
"informational_skipped": test_informational_skipped,
"retry_counting": test_retry_counting,
}
def main() -> None:
if len(sys.argv) < 2:
print("Usage: helper_fix_then_revalidate.py <command>", file=sys.stderr)
sys.exit(1)
cmd = sys.argv[1]
func = COMMANDS.get(cmd)
if func is None:
print(f"Unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
try:
func()
except Exception as exc:
print(f"FAIL: {exc}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
src/cleveragents/application/container.py
+10
View File
@@ -42,6 +42,9 @@ from cleveragents.application.services.decomposition_service import (
from cleveragents.application.services.execution_environment_resolver import (
ExecutionEnvironmentResolver,
)
from cleveragents.application.services.fix_then_revalidate import (
FixThenRevalidateOrchestrator,
)
from cleveragents.application.services.multi_project_service import (
MultiProjectService,
)
@@ -712,6 +715,13 @@ class Container(containers.DeclarativeContainer):
ExecutionEnvironmentResolver,
)
# Fix-then-Revalidate Orchestrator - Factory (per-plan lifecycle, #583)
fix_then_revalidate_orchestrator = providers.Factory(
FixThenRevalidateOrchestrator,
validation_pipeline=None,
event_bus=event_bus,
)
# Plugin Manager - Singleton (shared plugin lifecycle management)
plugin_manager = providers.Singleton(PluginManager)
src/cleveragents/application/services/__init__.py
+15
View File
@@ -135,6 +135,15 @@ if TYPE_CHECKING:
from cleveragents.application.services.execution_environment_resolver import (
ExecutionEnvironmentResolver as ExecutionEnvironmentResolver,
)
from cleveragents.application.services.fix_then_revalidate import (
FixAttemptRecord as FixAttemptRecord,
)
from cleveragents.application.services.fix_then_revalidate import (
FixThenRevalidateOrchestrator as FixThenRevalidateOrchestrator,
)
from cleveragents.application.services.fix_then_revalidate import (
FixThenRevalidateResult as FixThenRevalidateResult,
)
from cleveragents.application.services.fusion_engine import (
FusionConfig as FusionConfig,
)
@@ -440,6 +449,12 @@ _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
"execution_environment_resolver",
"ExecutionEnvironmentResolver",
),
"FixAttemptRecord": ("fix_then_revalidate", "FixAttemptRecord"),
"FixThenRevalidateOrchestrator": (
"fix_then_revalidate",
"FixThenRevalidateOrchestrator",
),
"FixThenRevalidateResult": ("fix_then_revalidate", "FixThenRevalidateResult"),
"FusionConfig": ("fusion_engine", "FusionConfig"),
"FusionEngine": ("fusion_engine", "FusionEngine"),
"FusionResult": ("fusion_engine", "FusionResult"),
src/cleveragents/application/services/fix_then_revalidate.py
+838
View File
@@ -0,0 +1,838 @@
"""Fix-then-Revalidate orchestration loop for required validations.
Implements the diagnosis -> self-fix -> re-validation -> retry limit ->
strategy revision escalation -> user escalation -> terminal failure flow
described in the specification section "Validation Failure Handling."
When a required validation fails during the Execute phase, this
orchestrator drives a bounded retry loop:
1. **Diagnosis** -- the execution actor examines the validation's
``message`` and ``data`` fields to understand the failure.
2. **Self-fix attempt** -- the actor attempts to correct the issue
within strategy constraints (via ``fix_callback``).
3. **Re-validation** -- the failing validation is re-invoked (via
``revalidate_callback``).
4. **Retry limit** -- configurable via the ``max_retries`` constructor
parameter (default 3, range 0--100 per spec Safety Profile
``max_retries_per_step``).
5. **Strategy revision** -- if the failure cannot be resolved and the
``auto_strategy_revision`` confidence threshold is met, the result
signals escalation so the caller can trigger a strategy revision.
6. **User escalation** -- if strategy revision is not available, the
result signals that user guidance is needed via
``agents plan prompt``.
7. **Terminal failure** -- reserved for the calling ``PlanExecutor``
when user escalation also fails; the plan transitions to
``state: failed`` and the sandbox is preserved for inspection.
Callback exceptions are caught and treated as failed attempts to ensure
retry loop resilience. This prevents a single misbehaving callback from
crashing the entire orchestration loop.
Validation errors (runtime exceptions during validation execution) are
treated as required failures regardless of the validation's mode, per
the specification section "Validation Error vs. Validation Failure."
"""
from __future__ import annotations
import re
import threading
from collections import defaultdict
from collections.abc import Callable
from datetime import UTC, datetime
import structlog
from pydantic import BaseModel, ConfigDict, Field, model_validator
from cleveragents.application.services.validation_pipeline import (
ValidationPipeline,
ValidationResult,
)
from cleveragents.core.exceptions import ValidationError
from cleveragents.domain.models.core.tool import ValidationMode
from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.protocol import EventBus
from cleveragents.infrastructure.events.types import EventType
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# ULID pattern for plan_id validation (matches domain model convention)
# ---------------------------------------------------------------------------
_ULID_PATTERN = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
# ---------------------------------------------------------------------------
# Callback type aliases
# ---------------------------------------------------------------------------
FixCallback = Callable[[ValidationResult], str | None]
"""Callable that attempts to fix a failed validation.
Receives the failed :class:`ValidationResult` and returns a
human-readable description of the fix that was applied, or ``None``
to signal that the failure is unfixable within current strategy
constraints and the orchestrator should escalate immediately.
May raise exceptions, which are caught by the orchestrator and
treated as failed fix attempts.
"""
RevalidateCallback = Callable[[ValidationResult], ValidationResult]
"""Callable that re-runs a single validation after a fix attempt.
Receives the original failed :class:`ValidationResult` (for context)
and returns a fresh :class:`ValidationResult` from re-execution.
"""
# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------
_MAX_VALIDATION_NAME_LEN = 255
class FixAttemptRecord(BaseModel):
"""Record of a single fix-then-revalidate attempt.
Attributes:
timestamp: When the attempt was made.
attempt_number: 1-based attempt counter for this validation.
validation_name: Name of the validation that was retried.
fix_description: Human-readable description of the fix applied.
success: Whether re-validation passed after the fix.
"""
timestamp: datetime = Field(
default_factory=lambda: datetime.now(tz=UTC),
description="When the attempt was made (UTC)",
)
attempt_number: int = Field(..., ge=1, description="1-based attempt counter")
validation_name: str = Field(
...,
min_length=1,
max_length=_MAX_VALIDATION_NAME_LEN,
description="Validation that was retried",
)
fix_description: str = Field(
..., max_length=2000, description="Description of the fix applied"
)
success: bool = Field(..., description="Whether re-validation passed after the fix")
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
)
class FixThenRevalidateResult(BaseModel):
"""Aggregated result from the fix-then-revalidate orchestration loop.
Attributes:
validation_attempts: Total fix attempts across all validations.
final_passed: Whether all required validations eventually passed.
validation_fix_history: Ordered list of all fix attempt records.
escalated: Whether the loop escalated to strategy revision.
needs_user_escalation: Whether user guidance is needed.
terminal_failure: Whether the plan reached terminal failure.
validation_summary: Array of all intermediate validation results
collected during the loop, including attempt numbers.
final_validation_results: Snapshot of the final validation state
for each processed validation.
"""
validation_attempts: int = Field(
..., ge=0, description="Total fix attempts across all validations"
)
final_passed: bool = Field(
..., description="Whether all required validations eventually passed"
)
validation_fix_history: list[FixAttemptRecord] = Field(
default_factory=list,
description="Ordered list of all fix attempt records",
)
escalated: bool = Field(
default=False,
description="Whether the loop escalated to strategy revision",
)
needs_user_escalation: bool = Field(
default=False,
description="Whether user guidance is needed via agents plan prompt",
)
terminal_failure: bool = Field(
default=False,
description="Whether the plan reached terminal failure",
)
validation_summary: list[ValidationResult] = Field(
default_factory=list,
description=(
"Array of all intermediate validation results collected "
"during the loop, including per-attempt outcomes"
),
)
final_validation_results: list[ValidationResult] = Field(
default_factory=list,
description=(
"Snapshot of the final validation state for each processed validation"
),
)
@model_validator(mode="after")
def _check_mutually_exclusive_flags(self) -> FixThenRevalidateResult:
"""Enforce mutual exclusivity of outcome flags.
Invalid combinations:
- At most one of ``escalated``, ``needs_user_escalation``, and
``terminal_failure`` can be True.
- ``final_passed`` True with any failure/escalation flag True.
"""
flags = [self.escalated, self.needs_user_escalation, self.terminal_failure]
if sum(flags) > 1:
msg = (
"At most one of escalated, needs_user_escalation, "
"and terminal_failure can be True"
)
raise ValueError(msg)
if self.final_passed and any(flags):
msg = (
"final_passed cannot be True when escalated, "
"needs_user_escalation, or terminal_failure is True"
)
raise ValueError(msg)
return self
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
)
# ---------------------------------------------------------------------------
# Orchestrator
# ---------------------------------------------------------------------------
_DEFAULT_MAX_RETRIES = 3
_MIN_MAX_RETRIES = 0
_MAX_MAX_RETRIES = 100
_MAX_FIX_DESCRIPTION_LEN = 2000
_MAX_EVENT_BUS_CONSECUTIVE_FAILURES = 5
def _sanitize_for_log(value: str) -> str:
"""Remove control characters from a string for safe log output."""
return value.replace("\n", "\\n").replace("\r", "\\r").replace("\t", "\\t")
class FixThenRevalidateOrchestrator:
"""Orchestrates the fix-then-revalidate loop for failed validations.
Only processes **required** validations that have failed;
informational failures are silently skipped. Validation errors
(runtime exceptions) are treated as required failures regardless
of the validation's mode.
Args:
validation_pipeline: The ``ValidationPipeline`` instance used for
context (not re-invoked directly -- use callbacks instead).
max_retries: Maximum fix attempts per validation (default 3,
range 0--100 per spec Safety Profile ``max_retries_per_step``).
auto_strategy_revision: Confidence threshold (0.0--1.0) for
automatic strategy revision when retries are exhausted.
0.0 = always auto-escalate, 1.0 = always require manual
approval.
auto_validation_fix: Confidence threshold (0.0--1.0) for
automatic validation fix attempts. 0.0 = always auto,
1.0 = always manual. Stored for caller use.
event_bus: Optional event bus for emitting domain events.
Raises:
ValidationError: If *validation_pipeline* is ``None``,
*max_retries* is out of range, or type constraints
are violated.
"""
def __init__(
self,
validation_pipeline: ValidationPipeline,
max_retries: int = _DEFAULT_MAX_RETRIES,
auto_strategy_revision: float = 0.0,
auto_validation_fix: float = 0.0,
event_bus: EventBus | None = None,
) -> None:
if validation_pipeline is None:
raise ValidationError("validation_pipeline must not be None")
if isinstance(max_retries, bool) or not isinstance(max_retries, int):
raise ValidationError("max_retries must be an integer")
if max_retries < _MIN_MAX_RETRIES or max_retries > _MAX_MAX_RETRIES:
raise ValidationError(
f"max_retries must be between {_MIN_MAX_RETRIES} and "
f"{_MAX_MAX_RETRIES}, got {max_retries}"
)
if not isinstance(auto_strategy_revision, (int, float)) or isinstance(
auto_strategy_revision, bool
):
raise ValidationError("auto_strategy_revision must be a float")
if not (0.0 <= float(auto_strategy_revision) <= 1.0):
raise ValidationError(
"auto_strategy_revision must be between 0.0 and 1.0, "
f"got {auto_strategy_revision}"
)
if not isinstance(auto_validation_fix, (int, float)) or isinstance(
auto_validation_fix, bool
):
raise ValidationError("auto_validation_fix must be a float")
if not (0.0 <= float(auto_validation_fix) <= 1.0):
raise ValidationError(
"auto_validation_fix must be between 0.0 and 1.0, "
f"got {auto_validation_fix}"
)
if event_bus is not None and not isinstance(event_bus, EventBus):
raise ValidationError("event_bus must be an EventBus instance or None")
self._pipeline = validation_pipeline
self._max_retries = max_retries
self._auto_strategy_revision = float(auto_strategy_revision)
self._auto_validation_fix = float(auto_validation_fix)
self._event_bus = event_bus
# Per-(plan_id, validation_key) retry counters.
# Inner key is (validation_name, resource_id) to prevent
# same-named validations on different resources from sharing
# a retry budget.
self._retry_counts: dict[str, dict[tuple[str, str], int]] = defaultdict(
lambda: defaultdict(int)
)
self._lock = threading.RLock()
self._event_bus_consecutive_failures = 0
@property
def max_retries(self) -> int:
"""Maximum fix attempts per validation."""
return self._max_retries
@property
def auto_strategy_revision(self) -> float:
"""Confidence threshold for automatic strategy revision."""
return self._auto_strategy_revision
@property
def auto_validation_fix(self) -> float:
"""Confidence threshold for automatic validation fix attempts."""
return self._auto_validation_fix
@property
def event_bus(self) -> EventBus | None:
"""Return the event bus, if configured."""
return self._event_bus
def get_retry_count(
self,
plan_id: str,
validation_name: str,
resource_id: str = "",
) -> int:
"""Return the current fix attempt count for a validation in a plan.
Args:
plan_id: The plan identifier.
validation_name: The validation name.
resource_id: The resource identifier (defaults to empty string
for backward compatibility).
Returns:
Number of fix attempts already made (0 if none).
"""
if not plan_id:
raise ValidationError("plan_id must not be empty")
if not validation_name:
raise ValidationError("validation_name must not be empty")
vkey = (validation_name, resource_id)
with self._lock:
plan_counts = self._retry_counts.get(plan_id)
if plan_counts is None:
return 0
return plan_counts.get(vkey, 0)
def reset_retry_counts(self, plan_id: str) -> None:
"""Reset all retry counters for a plan.
Args:
plan_id: The plan identifier.
Raises:
ValidationError: If *plan_id* is empty.
"""
if not plan_id:
raise ValidationError("plan_id must not be empty")
with self._lock:
self._retry_counts.pop(plan_id, None)
def run_fix_loop(
self,
plan_id: str,
failed_results: list[ValidationResult],
fix_callback: FixCallback,
revalidate_callback: RevalidateCallback,
*,
actor_name: str | None = None,
) -> FixThenRevalidateResult:
"""Run the fix-then-revalidate loop for failed validations.
Only **required** validations are processed. Informational
failures are silently skipped unless they have a validation error,
in which case they are treated as required failures per the
specification section "Validation Error vs. Validation Failure."
Args:
plan_id: Identifier of the plan being executed (ULID format).
failed_results: List of failed validation results to process.
fix_callback: Callable that attempts to fix a failure.
revalidate_callback: Callable that re-runs a validation.
actor_name: Optional actor name for event correlation.
Returns:
:class:`FixThenRevalidateResult` describing the loop outcome.
Raises:
ValidationError: If required arguments are invalid.
"""
if not plan_id:
raise ValidationError("plan_id must not be empty")
if failed_results is None:
raise ValidationError("failed_results must not be None")
if fix_callback is None:
raise ValidationError("fix_callback must not be None")
if not callable(fix_callback):
raise ValidationError("fix_callback must be callable")
if revalidate_callback is None:
raise ValidationError("revalidate_callback must not be None")
if not callable(revalidate_callback):
raise ValidationError("revalidate_callback must be callable")
# Filter to required failures AND validation errors (regardless
# of mode). Per spec: validation errors are "always treated as
# required failure regardless of mode."
required_failures = [
r
for r in failed_results
if (r.mode == ValidationMode.REQUIRED and not r.passed)
or (r.error is not None)
]
if not required_failures:
log = logger.bind(plan_id=plan_id)
log.info("no_required_validation_failures_to_fix")
return FixThenRevalidateResult(
validation_attempts=0,
final_passed=True,
validation_fix_history=[],
escalated=False,
needs_user_escalation=False,
terminal_failure=False,
validation_summary=[],
final_validation_results=[],
)
validation_fix_history: list[FixAttemptRecord] = []
validation_summary: list[ValidationResult] = []
final_validation_results: list[ValidationResult] = []
all_resolved = True
for failed_result in required_failures:
resolved, last_result, intermediate_results = self._fix_single_validation(
plan_id=plan_id,
failed_result=failed_result,
fix_callback=fix_callback,
revalidate_callback=revalidate_callback,
validation_fix_history=validation_fix_history,
actor_name=actor_name,
)
validation_summary.extend(intermediate_results)
final_validation_results.append(last_result)
if not resolved:
all_resolved = False
validation_attempts = len(validation_fix_history)
# Auto-reset retry counts for this plan after the loop completes
# to prevent cross-invocation budget exhaustion when the caller
# re-invokes after strategy revision.
self.reset_retry_counts(plan_id)
if all_resolved:
log = logger.bind(plan_id=plan_id, validation_attempts=validation_attempts)
log.info("all_required_validations_resolved")
return FixThenRevalidateResult(
validation_attempts=validation_attempts,
final_passed=True,
validation_fix_history=validation_fix_history,
escalated=False,
needs_user_escalation=False,
terminal_failure=False,
validation_summary=validation_summary,
final_validation_results=final_validation_results,
)
# --- Escalation path: strategy revision ---
if self._auto_strategy_revision < 1.0:
log = logger.bind(plan_id=plan_id)
log.warning("fix_loop_exhausted_escalating_to_strategy_revision")
self._emit_event(
EventType.VALIDATION_FIX_EXHAUSTED,
plan_id=plan_id,
actor_name=actor_name,
details={
"validation_attempts": validation_attempts,
"escalated": True,
"needs_user_escalation": False,
"terminal_failure": False,
},
)
return FixThenRevalidateResult(
validation_attempts=validation_attempts,
final_passed=False,
validation_fix_history=validation_fix_history,
escalated=True,
needs_user_escalation=False,
terminal_failure=False,
validation_summary=validation_summary,
final_validation_results=final_validation_results,
)
# --- User escalation (strategy revision not available) ---
log = logger.bind(plan_id=plan_id)
log.warning("fix_loop_exhausted_needs_user_escalation")
self._emit_event(
EventType.VALIDATION_FIX_EXHAUSTED,
plan_id=plan_id,
actor_name=actor_name,
details={
"validation_attempts": validation_attempts,
"escalated": False,
"needs_user_escalation": True,
"terminal_failure": False,
},
)
return FixThenRevalidateResult(
validation_attempts=validation_attempts,
final_passed=False,
validation_fix_history=validation_fix_history,
escalated=False,
needs_user_escalation=True,
terminal_failure=False,
validation_summary=validation_summary,
final_validation_results=final_validation_results,
)
def _fix_single_validation(
self,
plan_id: str,
failed_result: ValidationResult,
fix_callback: FixCallback,
revalidate_callback: RevalidateCallback,
validation_fix_history: list[FixAttemptRecord],
actor_name: str | None = None,
) -> tuple[bool, ValidationResult, list[ValidationResult]]:
"""Attempt to fix and revalidate a single failed validation.
Callback exceptions are caught and treated as failed attempts
to ensure retry loop resilience.
Args:
plan_id: The plan identifier.
failed_result: The validation result that failed.
fix_callback: Callback to attempt a fix.
revalidate_callback: Callback to re-run the validation.
validation_fix_history: Mutable list to append records to.
actor_name: Optional actor name for event correlation.
Returns:
Tuple of (resolved, last_result, intermediate_results) where
resolved is ``True`` if the validation was fixed, last_result
is the final ``ValidationResult`` for this validation, and
intermediate_results is a list of all ``ValidationResult``
objects produced during the attempts (for validation_summary).
"""
vname = failed_result.validation_name
safe_vname = _sanitize_for_log(vname)
resource_id = failed_result.resource_id
# Defensive truncation to prevent FixAttemptRecord Pydantic error
# when ValidationResult.validation_name exceeds max_length.
truncated_vname = vname[:_MAX_VALIDATION_NAME_LEN]
vkey = (vname, resource_id)
attempted = False
last_result = failed_result
intermediate_results: list[ValidationResult] = []
for _ in range(self._max_retries):
# Atomically claim a retry slot. This prevents the TOCTOU
# race where concurrent threads could both read the same
# counter value, pass the budget guard, and exceed the
# configured retry limit.
with self._lock:
plan_counts = self._retry_counts.get(plan_id)
current = plan_counts.get(vkey, 0) if plan_counts is not None else 0
if current >= self._max_retries:
break
attempt_num = current + 1
self._retry_counts[plan_id][vkey] = attempt_num
attempted = True
log = logger.bind(
plan_id=plan_id,
validation_name=safe_vname,
attempt=attempt_num,
max_retries=self._max_retries,
)
log.info("fix_attempt_started")
# Step 1: Diagnosis + fix attempt.
# Exceptions from the callback are treated as failed fix
# attempts to ensure loop resilience.
try:
fix_description = fix_callback(failed_result)
except Exception as exc:
fix_description = (
f"Fix callback raised {type(exc).__name__} for "
f"'{safe_vname}' (attempt {attempt_num})"
)
log.warning(
"fix_callback_exception",
error_type=type(exc).__name__,
error_msg=str(exc)[:200],
)
record = FixAttemptRecord(
attempt_number=attempt_num,
validation_name=truncated_vname,
fix_description=fix_description[:_MAX_FIX_DESCRIPTION_LEN],
success=False,
)
validation_fix_history.append(record)
self._emit_event(
EventType.VALIDATION_FIX_ATTEMPTED,
plan_id=plan_id,
actor_name=actor_name,
details={
"validation_name": vname,
"attempt_number": attempt_num,
"success": False,
"error": "fix_callback_exception",
},
)
continue
# Handle early-exit signal: None means "unfixable, escalate"
if fix_description is None:
log.info("fix_callback_signalled_unfixable")
record = FixAttemptRecord(
attempt_number=attempt_num,
validation_name=truncated_vname,
fix_description="Callback signalled unfixable",
success=False,
)
validation_fix_history.append(record)
self._emit_event(
EventType.VALIDATION_FIX_ATTEMPTED,
plan_id=plan_id,
actor_name=actor_name,
details={
"validation_name": vname,
"attempt_number": attempt_num,
"success": False,
"error": "unfixable_within_strategy",
},
)
return False, last_result, intermediate_results
# Defensive type coercion for non-string returns
if not isinstance(fix_description, str):
fix_description = str(fix_description)
# Truncate fix_description to prevent PydanticValidationError
# from the max_length constraint on FixAttemptRecord.
fix_description = fix_description[:_MAX_FIX_DESCRIPTION_LEN]
# Step 2: Re-validation.
# Exceptions are treated as validation errors (passed=False).
try:
new_result = revalidate_callback(failed_result)
except Exception as exc:
log.warning(
"revalidate_callback_exception",
error_type=type(exc).__name__,
error_msg=str(exc)[:200],
)
new_result = ValidationResult(
validation_name=failed_result.validation_name,
resource_id=failed_result.resource_id,
resource_name=failed_result.resource_name,
mode=failed_result.mode,
passed=False,
message=(
f"Revalidation raised an exception (attempt {attempt_num})"
),
data=failed_result.data,
duration_ms=0.0,
error=(f"revalidate_callback exception (attempt {attempt_num})"),
)
# Verify the returned result matches the expected validation.
if new_result.validation_name != vname:
log.warning(
"revalidate_validation_name_mismatch",
expected=safe_vname,
got=_sanitize_for_log(new_result.validation_name),
)
new_result = ValidationResult(
validation_name=vname,
resource_id=failed_result.resource_id,
resource_name=failed_result.resource_name,
mode=failed_result.mode,
passed=False,
message=(
f"Revalidation returned mismatched "
f"validation_name (expected '{safe_vname}', "
f"got '{_sanitize_for_log(new_result.validation_name)}')"
),
data=failed_result.data,
duration_ms=new_result.duration_ms,
error=(f"validation_name mismatch (attempt {attempt_num})"),
)
# Handle contradictory passed=True with error set.
# Per spec: validation errors are always treated as required
# failure regardless of mode.
if new_result.passed and new_result.error is not None:
log.warning(
"contradictory_passed_with_error",
error=new_result.error[:200],
)
new_result = ValidationResult(
validation_name=new_result.validation_name,
resource_id=new_result.resource_id,
resource_name=new_result.resource_name,
mode=new_result.mode,
passed=False,
message=(
f"Validation reported passed=True with error "
f"set; treating as failed: {new_result.error}"
),
data=new_result.data,
duration_ms=new_result.duration_ms,
error=new_result.error,
)
last_result = new_result
intermediate_results.append(new_result)
record = FixAttemptRecord(
attempt_number=attempt_num,
validation_name=truncated_vname,
fix_description=fix_description,
success=new_result.passed,
)
validation_fix_history.append(record)
self._emit_event(
EventType.VALIDATION_FIX_ATTEMPTED,
plan_id=plan_id,
actor_name=actor_name,
details={
"validation_name": vname,
"attempt_number": attempt_num,
"success": new_result.passed,
},
)
if new_result.passed:
log.info("validation_passed_after_fix")
self._emit_event(
EventType.VALIDATION_FIX_SUCCEEDED,
plan_id=plan_id,
actor_name=actor_name,
details={
"validation_name": vname,
"attempt_number": attempt_num,
},
)
return True, last_result, intermediate_results
# Update failed_result for next iteration's diagnosis
failed_result = new_result
if not attempted:
log = logger.bind(
plan_id=plan_id,
validation_name=safe_vname,
current_count=self.get_retry_count(plan_id, vname, resource_id),
max_retries=self._max_retries,
)
log.warning("retry_budget_already_exhausted")
else:
log = logger.bind(
plan_id=plan_id,
validation_name=safe_vname,
max_retries=self._max_retries,
)
log.warning("retry_limit_exhausted")
return False, last_result, intermediate_results
def _emit_event(
self,
event_type: EventType,
plan_id: str,
details: dict[str, object],
actor_name: str | None = None,
) -> None:
"""Emit a domain event if an event bus is configured.
Includes a simple circuit breaker: after
``_MAX_EVENT_BUS_CONSECUTIVE_FAILURES`` consecutive failures,
event emission is silently disabled until a successful emit
resets the counter.
Args:
event_type: The event type to emit.
plan_id: The plan identifier for correlation.
details: Event-specific payload.
actor_name: Optional actor name for event correlation.
"""
if self._event_bus is None:
return
with self._lock:
if (
self._event_bus_consecutive_failures
>= _MAX_EVENT_BUS_CONSECUTIVE_FAILURES
):
return
try:
self._event_bus.emit(
DomainEvent(
event_type=event_type,
plan_id=plan_id,
actor_name=actor_name,
details=details,
)
)
with self._lock:
self._event_bus_consecutive_failures = 0
except Exception:
with self._lock:
self._event_bus_consecutive_failures += 1
consecutive_failures = self._event_bus_consecutive_failures
log = logger.bind(
event_type=str(event_type),
plan_id=plan_id,
consecutive_failures=consecutive_failures,
)
log.warning("event_bus_emit_failed")
__all__ = [
"FixAttemptRecord",
"FixCallback",
"FixThenRevalidateOrchestrator",
"FixThenRevalidateResult",
"RevalidateCallback",
]
src/cleveragents/application/services/plan_executor.py
+15
View File
@@ -23,6 +23,9 @@ from ulid import ULID
from cleveragents.application.services.autonomy_guardrail_service import (
AutonomyGuardrailService,
)
from cleveragents.application.services.fix_then_revalidate import (
FixThenRevalidateOrchestrator,
)
from cleveragents.application.services.plan_execution_context import (
PlanExecutionContext,
RuntimeExecuteActor,
@@ -299,6 +302,7 @@ class PlanExecutor:
metrics_emitter: MetricsEmitter | None = None,
strategize_actor: Any | None = None,
execute_actor: Any | None = None,
fix_revalidate_orchestrator: FixThenRevalidateOrchestrator | None = None,
) -> None:
"""Initialize the plan executor.
@@ -322,6 +326,9 @@ class PlanExecutor:
execute_actor: Optional custom execute actor. When ``None``,
the default ``ExecuteStubActor`` is used. Pass an
``LLMExecuteActor`` for real LLM execution.
fix_revalidate_orchestrator: Optional orchestrator for
fix-then-revalidate loops when required validations
fail during execution (Forgejo #583).
"""
if lifecycle_service is None:
raise ValidationError("lifecycle_service must not be None")
@@ -333,6 +340,7 @@ class PlanExecutor:
self._checkpoint_manager = checkpoint_manager
self._guardrail_service = guardrail_service
self._metrics_emitter = metrics_emitter
self._fix_revalidate_orchestrator = fix_revalidate_orchestrator
self._strategize_actor = strategize_actor or StrategizeStubActor()
self._execute_actor = execute_actor or ExecuteStubActor()
self._logger = logger.bind(service="plan_executor")
@@ -382,6 +390,13 @@ class PlanExecutor:
"""Return the execution context, if configured."""
return self._execution_context
@property
def fix_revalidate_orchestrator(
self,
) -> FixThenRevalidateOrchestrator | None:
"""Return the fix-then-revalidate orchestrator, if configured."""
return self._fix_revalidate_orchestrator
def _try_create_checkpoint(
self,
plan_id: str,
src/cleveragents/infrastructure/events/types.py
+3
View File
@@ -86,6 +86,9 @@ class EventType(StrEnum):
VALIDATION_STARTED = "validation.started"
VALIDATION_PASSED = "validation.passed"
VALIDATION_FAILED = "validation.failed"
VALIDATION_FIX_ATTEMPTED = "validation.fix_attempted"
VALIDATION_FIX_SUCCEEDED = "validation.fix_succeeded"
VALIDATION_FIX_EXHAUSTED = "validation.fix_exhausted"
# --- Session ---
SESSION_CREATED = "session.created"