forked from cleveragents/cleveragents-core
aditya
3f6b255747
Add --no-estimate opt-out flag for plan use command and persist EstimationSkipped reason when estimation is skipped or fails. Add EstimationReport Pydantic domain model (identical to #649) with spec-aligned multi-dimensional output: cost range (min/max USD), expected steps, expected child plans, rollback risk (0.0-1.0), estimated duration in minutes, confidence (0.0-1.0), rationale, and optional historical basis. Add EstimationSkipped domain model with reason, timestamp, and optional actor_name fields for tracking when estimation is skipped via --no-estimate or when the estimation actor fails. Add estimation_produced to DecisionType enum and STRATEGIZE_TYPES. Add Alembic migration m6_006 adding estimation_report_json and estimation_skipped_json columns to v3_plans and updating the ck_decisions_type CHECK constraint. Wire --no-estimate through plan use CLI: when set, creates EstimationSkipped with reason and clears the estimation actor. Update _run_estimation() to persist EstimationSkipped on actor failure instead of only logging. Add docs/reference/estimation.md documenting the estimation feature, EstimationReport schema, EstimationSkipped schema, and CLI examples. Add benchmarks/estimation_actor_bench.py with ASV benchmarks for EstimationReport and EstimationSkipped operations. Add 13 Behave scenarios (estimation_skip.feature) and 6 Robot integration tests (estimation_skip.robot) covering EstimationSkipped model, EstimationReport model, plan field mutual exclusion, and --no-estimate CLI behavior. ISSUES CLOSED: #209
324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""Step definitions for estimation skip behavior tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.estimation import (
|
|
EstimationReport,
|
|
EstimationSkipped,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# EstimationSkipped model
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I create an EstimationSkipped with only reason "{reason}"')
|
|
def step_create_skipped(context: Context, reason: str) -> None:
|
|
context.estimation_skipped = EstimationSkipped(reason=reason)
|
|
|
|
|
|
@when('I create an EstimationSkipped with reason "{reason}" and actor "{actor}"')
|
|
def step_create_skipped_with_actor(context: Context, reason: str, actor: str) -> None:
|
|
context.estimation_skipped = EstimationSkipped(reason=reason, actor_name=actor)
|
|
|
|
|
|
@then("the EstimationSkipped should be created successfully")
|
|
def step_skipped_created(context: Context) -> None:
|
|
assert context.estimation_skipped is not None
|
|
|
|
|
|
@then('the EstimationSkipped reason should be "{reason}"')
|
|
def step_skipped_reason(context: Context, reason: str) -> None:
|
|
assert context.estimation_skipped.reason == reason
|
|
|
|
|
|
@then("the EstimationSkipped actor_name should be None")
|
|
def step_skipped_actor_none(context: Context) -> None:
|
|
assert context.estimation_skipped.actor_name is None
|
|
|
|
|
|
@then('the EstimationSkipped actor_name should be "{actor}"')
|
|
def step_skipped_actor_value(context: Context, actor: str) -> None:
|
|
assert context.estimation_skipped.actor_name == actor
|
|
|
|
|
|
@then("modifying the EstimationSkipped reason should raise an error")
|
|
def step_skipped_frozen(context: Context) -> None:
|
|
try:
|
|
context.estimation_skipped.reason = "modified" # type: ignore[misc]
|
|
raise AssertionError("Should have raised an error for frozen model")
|
|
except (ValidationError, TypeError, AttributeError):
|
|
pass # Expected: frozen model
|
|
|
|
|
|
@then("the EstimationSkipped should round-trip through model_dump and model_validate")
|
|
def step_skipped_round_trip(context: Context) -> None:
|
|
data = context.estimation_skipped.model_dump()
|
|
restored = EstimationSkipped.model_validate(data)
|
|
assert restored.reason == context.estimation_skipped.reason
|
|
assert restored.actor_name == context.estimation_skipped.actor_name
|
|
|
|
|
|
@when("I try to create an EstimationSkipped with empty reason")
|
|
def step_create_skipped_empty(context: Context) -> None:
|
|
context.skip_error = None
|
|
try:
|
|
EstimationSkipped(reason="")
|
|
except ValidationError as exc:
|
|
context.skip_error = exc
|
|
|
|
|
|
@then("an estimation skip validation error should be raised")
|
|
def step_skip_validation_error(context: Context) -> None:
|
|
assert context.skip_error is not None
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# EstimationReport model
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _make_valid_report() -> EstimationReport:
|
|
return EstimationReport(
|
|
cost_range_usd_min=0.01,
|
|
cost_range_usd_max=0.50,
|
|
expected_steps=10,
|
|
expected_child_plans=3,
|
|
rollback_risk=0.15,
|
|
estimated_duration_minutes=5.5,
|
|
confidence=0.85,
|
|
rationale="Based on similar tasks.",
|
|
historical_basis=("plan-001", "plan-002"),
|
|
)
|
|
|
|
|
|
@when("I create a valid EstimationReport")
|
|
def step_create_report(context: Context) -> None:
|
|
context.estimation_report = _make_valid_report()
|
|
|
|
|
|
@then("the EstimationReport should be created successfully")
|
|
def step_report_created(context: Context) -> None:
|
|
assert context.estimation_report is not None
|
|
|
|
|
|
@then("the EstimationReport confidence should be {value:g}")
|
|
def step_report_confidence(context: Context, value: float) -> None:
|
|
assert context.estimation_report.confidence == value
|
|
|
|
|
|
@when("I try to create an EstimationReport with max cost less than min cost")
|
|
def step_report_bad_cost(context: Context) -> None:
|
|
context.report_error = None
|
|
try:
|
|
EstimationReport(
|
|
cost_range_usd_min=1.0,
|
|
cost_range_usd_max=0.5,
|
|
expected_steps=1,
|
|
expected_child_plans=0,
|
|
rollback_risk=0.1,
|
|
estimated_duration_minutes=1.0,
|
|
confidence=0.5,
|
|
rationale="Bad cost range.",
|
|
)
|
|
except ValidationError as exc:
|
|
context.report_error = exc
|
|
|
|
|
|
@when("I try to create an EstimationReport with confidence {value:g}")
|
|
def step_report_bad_confidence(context: Context, value: float) -> None:
|
|
context.report_error = None
|
|
try:
|
|
EstimationReport(
|
|
cost_range_usd_min=0.0,
|
|
cost_range_usd_max=0.0,
|
|
expected_steps=0,
|
|
expected_child_plans=0,
|
|
rollback_risk=0.0,
|
|
estimated_duration_minutes=0.0,
|
|
confidence=value,
|
|
rationale="Bad confidence.",
|
|
)
|
|
except ValidationError as exc:
|
|
context.report_error = exc
|
|
|
|
|
|
@when("I try to create an EstimationReport with rollback risk {value}")
|
|
def step_report_bad_risk(context: Context, value: str) -> None:
|
|
context.report_error = None
|
|
try:
|
|
EstimationReport(
|
|
cost_range_usd_min=0.0,
|
|
cost_range_usd_max=0.0,
|
|
expected_steps=0,
|
|
expected_child_plans=0,
|
|
rollback_risk=float(value),
|
|
estimated_duration_minutes=0.0,
|
|
confidence=0.5,
|
|
rationale="Bad risk.",
|
|
)
|
|
except ValidationError as exc:
|
|
context.report_error = exc
|
|
|
|
|
|
@when("I try to create an EstimationReport with {count:d} historical basis entries")
|
|
def step_report_too_many_basis(context: Context, count: int) -> None:
|
|
context.report_error = None
|
|
try:
|
|
EstimationReport(
|
|
cost_range_usd_min=0.0,
|
|
cost_range_usd_max=0.0,
|
|
expected_steps=0,
|
|
expected_child_plans=0,
|
|
rollback_risk=0.0,
|
|
estimated_duration_minutes=0.0,
|
|
confidence=0.5,
|
|
rationale="Too many basis entries.",
|
|
historical_basis=tuple(f"plan-{i}" for i in range(count)),
|
|
)
|
|
except ValidationError as exc:
|
|
context.report_error = exc
|
|
|
|
|
|
@then("an estimation report validation error should be raised")
|
|
def step_report_validation_error(context: Context) -> None:
|
|
assert context.report_error is not None
|
|
|
|
|
|
@then("modifying the EstimationReport rationale should raise an error")
|
|
def step_report_frozen(context: Context) -> None:
|
|
try:
|
|
context.estimation_report.rationale = "modified" # type: ignore[misc]
|
|
raise AssertionError("Should have raised an error for frozen model")
|
|
except (ValidationError, TypeError, AttributeError):
|
|
pass # Expected: frozen model
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Plan fields
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a plan with estimation_skipped set")
|
|
def step_plan_with_skipped(context: Context) -> None:
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
)
|
|
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id=str(ULID())),
|
|
namespaced_name=NamespacedName.parse("local/test-skip"),
|
|
action_name="local/test-action",
|
|
description="Test plan with estimation skipped",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
estimation_skipped=EstimationSkipped(reason="Skipped for test"),
|
|
)
|
|
|
|
|
|
@when("I create a plan with estimation_report set")
|
|
def step_plan_with_report(context: Context) -> None:
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
)
|
|
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id=str(ULID())),
|
|
namespaced_name=NamespacedName.parse("local/test-report"),
|
|
action_name="local/test-action",
|
|
description="Test plan with estimation report",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
estimation_report=_make_valid_report(),
|
|
)
|
|
|
|
|
|
@then("the plan estimation_skipped should not be None")
|
|
def step_plan_skipped_not_none(context: Context) -> None:
|
|
assert context.plan.estimation_skipped is not None
|
|
|
|
|
|
@then("the plan estimation_report should be None")
|
|
def step_plan_report_none(context: Context) -> None:
|
|
assert context.plan.estimation_report is None
|
|
|
|
|
|
@then("the plan estimation_report should not be None")
|
|
def step_plan_report_not_none(context: Context) -> None:
|
|
assert context.plan.estimation_report is not None
|
|
|
|
|
|
@then("the plan estimation_skipped should be None")
|
|
def step_plan_skipped_none(context: Context) -> None:
|
|
assert context.plan.estimation_skipped is None
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# --no-estimate lifecycle integration
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@given("a lifecycle service with a test action")
|
|
def step_lifecycle_service(context: Context) -> None:
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
|
|
settings = Settings()
|
|
context.lifecycle_service = PlanLifecycleService(settings=settings)
|
|
context.lifecycle_service.create_action(
|
|
name="local/test-est-skip",
|
|
description="Test action for estimation skip",
|
|
definition_of_done="All tests pass",
|
|
strategy_actor="local/mock-strategy",
|
|
execution_actor="local/mock-execute",
|
|
estimation_actor="local/mock-estimation",
|
|
)
|
|
|
|
|
|
@when("I use the action with --no-estimate")
|
|
def step_use_action_no_estimate(context: Context) -> None:
|
|
plan = context.lifecycle_service.use_action(
|
|
action_name="local/test-est-skip",
|
|
)
|
|
# Simulate --no-estimate behavior
|
|
plan.estimation_skipped = EstimationSkipped(
|
|
reason="User opted out via --no-estimate",
|
|
actor_name=plan.estimation_actor,
|
|
)
|
|
plan.estimation_actor = None
|
|
context.plan = plan
|
|
# Also set est_plan for compatibility with shared step from estimation_actor_steps.py
|
|
context.est_plan = plan
|
|
|
|
|
|
@then("the plan should have estimation_skipped set")
|
|
def step_plan_has_skipped(context: Context) -> None:
|
|
assert context.plan.estimation_skipped is not None
|
|
|
|
|
|
@then('the estimation_skipped reason should mention "{text}"')
|
|
def step_skipped_reason_mention(context: Context, text: str) -> None:
|
|
assert text in context.plan.estimation_skipped.reason
|
|
|
|
|
|
# Note: "the plan estimation_actor should be None" is defined in
|
|
# estimation_actor_steps.py and reused here.
|