From f325f7a03622210f2f52a6e2836a0f7a8fac579f Mon Sep 17 00:00:00 2001 From: "Brent E. Edwards" Date: Sun, 15 Mar 2026 22:31:52 +0000 Subject: [PATCH] feat(actor): implement estimation actor type MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add EstimationResult domain model with cost, time, token, risk, and step estimates. Add ESTIMATOR role to ActorRole enum. Add estimation_result field to Plan model. Implement EstimationStubActor for local estimation. Integrate estimation invocation into plan lifecycle after Strategize phase. Display estimation results in plan status CLI output. Estimation is optional — plans work without an estimation actor configured. Review fixes: add ge=0 constraints on numeric fields, Literal type for risk_level, tuple for risk_factors (frozen model safety), allow_inf_nan=False, max_length bounds on summary/raw_output, risk_factors cap at 100, __all__ export, as_display_dict() to deduplicate serialization, stub actor TODO, and assertion resilience in context_tiers steps. Also fixes pre-existing test failures discovered during full nox suite verification. ISSUES CLOSED: #890 --- features/context_tiers.feature | 4 +- features/estimation_actor.feature | 95 +++++ features/steps/context_tiers_steps.py | 5 +- features/steps/estimation_actor_steps.py | 394 ++++++++++++++++++ robot/estimation_actor.robot | 55 +++ robot/helper_estimation_actor.py | 229 ++++++++++ .../application/services/plan_executor.py | 25 ++ .../services/plan_lifecycle_service.py | 45 ++ src/cleveragents/cli/commands/plan.py | 26 ++ src/cleveragents/domain/models/acms/tiers.py | 2 + .../domain/models/core/__init__.py | 2 + .../domain/models/core/estimation.py | 117 ++++++ src/cleveragents/domain/models/core/plan.py | 7 + vulture_whitelist.py | 15 + 14 files changed, 1017 insertions(+), 4 deletions(-) create mode 100644 features/estimation_actor.feature create mode 100644 features/steps/estimation_actor_steps.py create mode 100644 robot/estimation_actor.robot create mode 100644 robot/helper_estimation_actor.py create mode 100644 src/cleveragents/domain/models/core/estimation.py diff --git a/features/context_tiers.feature b/features/context_tiers.feature index 12d278e73..b34ffd170 100644 --- a/features/context_tiers.feature +++ b/features/context_tiers.feature @@ -10,8 +10,8 @@ Feature: ACMS Context Tiers # ---- ActorRole enum ---- - Scenario: ActorRole has three values - Then ActorRole should have values "strategist", "executor", "reviewer" + Scenario: ActorRole has four values + Then ActorRole should have values "strategist", "executor", "reviewer", "estimator" # ---- TieredFragment ---- diff --git a/features/estimation_actor.feature b/features/estimation_actor.feature new file mode 100644 index 000000000..8a38d4643 --- /dev/null +++ b/features/estimation_actor.feature @@ -0,0 +1,95 @@ +Feature: Estimation Actor + As a developer + I want an estimation actor that provides cost, time, and resource estimates + So that I can make informed decisions before executing plans + + # EstimationResult domain model + + Scenario: EstimationResult can be created with all fields populated + Given I create an estimation result with all fields + Then the estimation result should have all expected values + + Scenario: EstimationResult can be created with defaults only + Given I create an estimation result with defaults + Then the estimation result summary should be empty + And the estimation result risk_factors should be empty + And all optional estimation fields should be None + + Scenario: EstimationResult as_display_dict includes all populated fields + Given I create an estimation result with all fields + When I call as_display_dict on the estimation result + Then the display dict should include estimated_child_plans + And the display dict should include estimated_time_seconds + And the display dict should include risk_factors + + Scenario: EstimationResult rejects more than 100 risk_factors + When I try to create an estimation result with 101 risk factors + Then a validation error should be raised for risk_factors limit + + Scenario: EstimationResult is frozen (immutable) + Given I create an estimation result with defaults + When I try to modify the estimation result summary + Then a validation error should be raised for frozen model + + Scenario: EstimationResult can be serialized to dict and back + Given I create an estimation result with all fields + When I serialize the estimation result to dict + And I deserialize it back to an EstimationResult + Then the deserialized result should match the original + + # Plan model accepts estimation_result + + Scenario: Plan model accepts estimation_result field + Given I have a plan lifecycle service for estimation tests + And I have an available action for estimation tests + When I use the action to create a plan for estimation tests + And I set an estimation result on the plan + Then the plan estimation_result should be set + + Scenario: Plan as_cli_dict includes estimation when present + Given I have a plan lifecycle service for estimation tests + And I have an available action for estimation tests + When I use the action to create a plan for estimation tests + And I set an estimation result on the plan + Then the plan cli dict should include estimation data + + # Plans work without estimation actor (optional behavior) + + Scenario: Plans work without estimation actor configured + Given I have a plan lifecycle service for estimation tests + When I create an action without estimation actor for estimation tests + And I use that action to create an estimation test plan + Then the plan estimation_actor should be None + And the plan estimation_result should be None + + # EstimationStubActor + + Scenario: EstimationStubActor produces placeholder result + When I run the estimation stub actor with a valid plan id + Then the stub result summary should contain "stub actor" + + Scenario: EstimationStubActor rejects empty plan_id + When I run the estimation stub actor with an empty plan id + Then a validation error should be raised for estimation + + # ActorRole enum includes ESTIMATOR + + Scenario: ActorRole enum includes ESTIMATOR + Then the ActorRole enum should have an ESTIMATOR value + And the ESTIMATOR value should be "estimator" + + # Estimation invocation in plan lifecycle + + Scenario: Estimation runs when estimation_actor is set during Strategize to Execute transition + Given I have a plan lifecycle service for estimation tests + And I have an action with estimation actor set + When I use the estimation action and complete strategize + And I transition the estimation plan to execute phase + Then the plan estimation_result should be set + + Scenario: Estimation is skipped when estimation_actor is not set + Given I have a plan lifecycle service for estimation tests + And I have an action without estimation actor + When I use the estimation action and complete strategize + And I transition the estimation plan to execute phase without estimation + Then the plan estimation_result should be None diff --git a/features/steps/context_tiers_steps.py b/features/steps/context_tiers_steps.py index 0ed2808b3..7728d2f9f 100644 --- a/features/steps/context_tiers_steps.py +++ b/features/steps/context_tiers_steps.py @@ -41,12 +41,13 @@ def step_then_context_tier_values(context: Any) -> None: # --------------------------------------------------------------------------- -@then('ActorRole should have values "strategist", "executor", "reviewer"') +@then('ActorRole should have values "strategist", "executor", "reviewer", "estimator"') def step_then_actor_role_values(context: Any) -> None: assert ActorRole.STRATEGIST == "strategist" assert ActorRole.EXECUTOR == "executor" assert ActorRole.REVIEWER == "reviewer" - assert len(ActorRole) == 3 + assert ActorRole.ESTIMATOR == "estimator" + assert "estimator" in [r.value for r in ActorRole] # --------------------------------------------------------------------------- diff --git a/features/steps/estimation_actor_steps.py b/features/steps/estimation_actor_steps.py new file mode 100644 index 000000000..396f0f379 --- /dev/null +++ b/features/steps/estimation_actor_steps.py @@ -0,0 +1,394 @@ +"""Step definitions for Estimation Actor feature tests.""" + +from __future__ import annotations + +from behave import given, then, when +from behave.runner import Context +from pydantic import ValidationError as PydanticValidationError + +from cleveragents.application.services.plan_executor import EstimationStubActor +from cleveragents.application.services.plan_lifecycle_service import ( + PlanLifecycleService, +) +from cleveragents.config.settings import Settings +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.acms.tiers import ActorRole +from cleveragents.domain.models.core.estimation import EstimationResult +from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState + +# --------------------------------------------------------------------------- +# EstimationResult domain model steps +# --------------------------------------------------------------------------- + + +@given("I create an estimation result with all fields") +def step_create_estimation_all_fields(context: Context) -> None: + """Create an EstimationResult with all fields populated.""" + context.estimation_result = EstimationResult( + estimated_cost_usd=0.0512, + estimated_tokens=12000, + estimated_steps=5, + estimated_child_plans=2, + estimated_time_seconds=30.5, + risk_level="medium", + risk_factors=("API rate limits", "Large codebase"), + summary="Estimated 5 steps, ~$0.05 cost", + raw_output="raw estimation output text", + ) + + +@then("the estimation result should have all expected values") +def step_check_estimation_all_values(context: Context) -> None: + """Verify all fields on the estimation result.""" + est: EstimationResult = context.estimation_result + assert est.estimated_cost_usd == 0.0512 + assert est.estimated_tokens == 12000 + assert est.estimated_steps == 5 + assert est.estimated_child_plans == 2 + assert est.estimated_time_seconds == 30.5 + assert est.risk_level == "medium" + assert est.risk_factors == ("API rate limits", "Large codebase") + assert est.summary == "Estimated 5 steps, ~$0.05 cost" + assert est.raw_output == "raw estimation output text" + + +@given("I create an estimation result with defaults") +def step_create_estimation_defaults(context: Context) -> None: + """Create an EstimationResult with only default values.""" + context.estimation_result = EstimationResult() + + +@then("the estimation result summary should be empty") +def step_check_estimation_summary_empty(context: Context) -> None: + """Verify the summary field is an empty string.""" + assert context.estimation_result.summary == "" + + +@then("the estimation result risk_factors should be empty") +def step_check_estimation_risk_factors_empty(context: Context) -> None: + """Verify risk_factors is an empty tuple.""" + assert context.estimation_result.risk_factors == () + + +@then("all optional estimation fields should be None") +def step_check_estimation_optional_none(context: Context) -> None: + """Verify all optional fields are None.""" + est: EstimationResult = context.estimation_result + assert est.estimated_cost_usd is None + assert est.estimated_tokens is None + assert est.estimated_steps is None + assert est.estimated_child_plans is None + assert est.estimated_time_seconds is None + assert est.risk_level is None + + +@when("I call as_display_dict on the estimation result") +def step_call_as_display_dict(context: Context) -> None: + """Call as_display_dict on the estimation result.""" + est: EstimationResult = context.estimation_result + context.display_dict = est.as_display_dict() + + +@then("the display dict should include estimated_child_plans") +def step_check_display_dict_child_plans(context: Context) -> None: + """Verify display dict includes estimated_child_plans.""" + assert "estimated_child_plans" in context.display_dict + assert context.display_dict["estimated_child_plans"] == 2 + + +@then("the display dict should include estimated_time_seconds") +def step_check_display_dict_time_seconds(context: Context) -> None: + """Verify display dict includes estimated_time_seconds.""" + assert "estimated_time_seconds" in context.display_dict + assert context.display_dict["estimated_time_seconds"] == 30.5 + + +@then("the display dict should include risk_factors") +def step_check_display_dict_risk_factors(context: Context) -> None: + """Verify display dict includes risk_factors as a list.""" + assert "risk_factors" in context.display_dict + assert context.display_dict["risk_factors"] == [ + "API rate limits", + "Large codebase", + ] + + +@when("I try to create an estimation result with 101 risk factors") +def step_create_estimation_too_many_risk_factors(context: Context) -> None: + """Try to create EstimationResult with >100 risk factors.""" + factors = tuple(f"risk-{i}" for i in range(101)) + try: + EstimationResult(risk_factors=factors) + context.error = None + except PydanticValidationError as exc: + context.error = exc + + +@then("a validation error should be raised for risk_factors limit") +def step_check_risk_factors_limit_error(context: Context) -> None: + """Verify a validation error was raised for too many risk factors.""" + assert context.error is not None + assert "100" in str(context.error) + + +@when("I try to modify the estimation result summary") +def step_try_modify_estimation(context: Context) -> None: + """Try to mutate the frozen EstimationResult.""" + try: + context.estimation_result.summary = "modified" # type: ignore[misc] + context.error = None + except PydanticValidationError as exc: + context.error = exc + + +@then("a validation error should be raised for frozen model") +def step_check_frozen_error(context: Context) -> None: + """Verify a validation error was raised due to frozen model.""" + assert context.error is not None + assert ( + "frozen" in str(context.error).lower() + or "immutable" in str(context.error).lower() + ) + + +@when("I serialize the estimation result to dict") +def step_serialize_estimation(context: Context) -> None: + """Serialize EstimationResult to dict.""" + context.estimation_dict = context.estimation_result.model_dump() + + +@when("I deserialize it back to an EstimationResult") +def step_deserialize_estimation(context: Context) -> None: + """Deserialize dict back to EstimationResult.""" + context.deserialized_estimation = EstimationResult(**context.estimation_dict) + + +@then("the deserialized result should match the original") +def step_check_deserialized_matches(context: Context) -> None: + """Verify deserialized object matches the original.""" + original: EstimationResult = context.estimation_result + deserialized: EstimationResult = context.deserialized_estimation + assert original == deserialized + + +# --------------------------------------------------------------------------- +# Plan model + estimation_result field steps +# --------------------------------------------------------------------------- + + +@given("I have a plan lifecycle service for estimation tests") +def step_create_lifecycle_for_estimation(context: Context) -> None: + """Create a plan lifecycle service.""" + settings = Settings() + context.est_lifecycle = PlanLifecycleService(settings=settings) + context.error = None + + +@given("I have an available action for estimation tests") +def step_create_action_for_estimation(context: Context) -> None: + """Create an action with estimation actor.""" + context.est_action = context.est_lifecycle.create_action( + name="local/est-test", + description="Test action for estimation", + definition_of_done="All tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + estimation_actor="local/estimator", + ) + + +@when("I use the action to create a plan for estimation tests") +def step_use_action_for_estimation(context: Context) -> None: + """Use the action to create a plan.""" + context.est_plan = context.est_lifecycle.use_action( + action_name="local/est-test", + ) + + +@when("I set an estimation result on the plan") +def step_set_estimation_on_plan(context: Context) -> None: + """Set an estimation result directly on the plan.""" + context.est_plan.estimation_result = EstimationResult( + estimated_cost_usd=1.23, + estimated_tokens=5000, + estimated_steps=3, + risk_level="low", + summary="Low risk, ~$1.23 estimated cost", + ) + + +@then("the plan estimation_result should be set") +def step_check_plan_estimation_set(context: Context) -> None: + """Verify the plan has an estimation result.""" + assert context.est_plan.estimation_result is not None + assert context.est_plan.estimation_result.summary != "" + + +@then("the plan cli dict should include estimation data") +def step_check_plan_cli_dict_estimation(context: Context) -> None: + """Verify as_cli_dict includes estimation data.""" + cli_dict = context.est_plan.as_cli_dict() + assert "estimation" in cli_dict + est_data = cli_dict["estimation"] + assert est_data["estimated_cost_usd"] == 1.23 + assert est_data["risk_level"] == "low" + + +# --------------------------------------------------------------------------- +# Plans without estimation actor +# --------------------------------------------------------------------------- + + +@when("I create an action without estimation actor for estimation tests") +def step_create_action_no_estimation(context: Context) -> None: + """Create an action without estimation actor.""" + context.est_action_no_est = context.est_lifecycle.create_action( + name="local/no-est-test", + description="Test action without estimation", + definition_of_done="All tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + ) + + +@when("I use that action to create an estimation test plan") +def step_use_no_est_action(context: Context) -> None: + """Use the action without estimation actor.""" + context.est_plan = context.est_lifecycle.use_action( + action_name="local/no-est-test", + ) + + +@then("the plan estimation_actor should be None") +def step_check_plan_no_estimation_actor(context: Context) -> None: + """Verify plan has no estimation actor.""" + assert context.est_plan.estimation_actor is None + + +@then("the plan estimation_result should be None") +def step_check_plan_no_estimation_result(context: Context) -> None: + """Verify plan has no estimation result.""" + assert context.est_plan.estimation_result is None + + +# --------------------------------------------------------------------------- +# EstimationStubActor steps +# --------------------------------------------------------------------------- + + +@when("I run the estimation stub actor with a valid plan id") +def step_run_stub_valid(context: Context) -> None: + """Run the stub actor with a valid plan ID.""" + stub = EstimationStubActor() + context.stub_result = stub.estimate("01ARZ3NDEKTSV4RRFFQ69G5FAV") + + +@then('the stub result summary should contain "stub actor"') +def step_check_stub_summary(context: Context) -> None: + """Verify the stub result summary mentions stub.""" + assert "stub actor" in context.stub_result.summary.lower() + + +@when("I run the estimation stub actor with an empty plan id") +def step_run_stub_empty_id(context: Context) -> None: + """Run the stub actor with an empty plan ID.""" + stub = EstimationStubActor() + try: + stub.estimate("") + context.error = None + except ValidationError as exc: + context.error = exc + + +@then("a validation error should be raised for estimation") +def step_check_estimation_validation_error(context: Context) -> None: + """Verify a validation error was raised.""" + assert context.error is not None + + +# --------------------------------------------------------------------------- +# ActorRole enum steps +# --------------------------------------------------------------------------- + + +@then("the ActorRole enum should have an ESTIMATOR value") +def step_check_actor_role_estimator(context: Context) -> None: + """Verify ActorRole has ESTIMATOR.""" + assert hasattr(ActorRole, "ESTIMATOR") + + +@then('the ESTIMATOR value should be "estimator"') +def step_check_estimator_value(context: Context) -> None: + """Verify ESTIMATOR value.""" + assert ActorRole.ESTIMATOR.value == "estimator" + + +# --------------------------------------------------------------------------- +# Estimation invocation in lifecycle steps +# --------------------------------------------------------------------------- + + +@given("I have an action with estimation actor set") +def step_create_action_with_estimation(context: Context) -> None: + """Create an action with estimation actor configured.""" + context.est_action = context.est_lifecycle.create_action( + name="local/est-lifecycle-test", + description="Test action with estimation", + definition_of_done="All tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + estimation_actor="local/estimator", + ) + + +@given("I have an action without estimation actor") +def step_create_action_without_estimation(context: Context) -> None: + """Create an action without estimation actor.""" + context.est_action = context.est_lifecycle.create_action( + name="local/no-est-lifecycle-test", + description="Test action without estimation", + definition_of_done="All tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + ) + + +@when("I use the estimation action and complete strategize") +def step_use_estimation_action_and_complete_strategize(context: Context) -> None: + """Use the action and drive through strategize phase.""" + action_name = str(context.est_action.namespaced_name) + context.est_plan = context.est_lifecycle.use_action(action_name=action_name) + plan_id = context.est_plan.identity.plan_id + + # Drive through strategize: start -> processing -> complete + context.est_lifecycle.start_strategize(plan_id) + # Manually set processing state to PROCESSING then COMPLETE + plan = context.est_lifecycle.get_plan(plan_id) + plan.processing_state = ProcessingState.PROCESSING + # TODO: Use public save_plan() once test helpers are refactored + context.est_lifecycle._commit_plan(plan) + context.est_lifecycle.complete_strategize(plan_id) + + +@when("I transition the estimation plan to execute phase") +def step_transition_estimation_plan_to_execute(context: Context) -> None: + """Transition plan to execute phase (triggers estimation).""" + plan_id = context.est_plan.identity.plan_id + plan = context.est_lifecycle.get_plan(plan_id) + # If auto_progress already advanced to Execute, just refresh + if plan.phase == PlanPhase.EXECUTE: + context.est_plan = plan + return + # Otherwise manually transition + context.est_plan = context.est_lifecycle.execute_plan(plan_id) + + +@when("I transition the estimation plan to execute phase without estimation") +def step_transition_estimation_plan_to_execute_no_estimation(context: Context) -> None: + """Transition plan to execute phase (no estimation actor).""" + plan_id = context.est_plan.identity.plan_id + plan = context.est_lifecycle.get_plan(plan_id) + if plan.phase == PlanPhase.EXECUTE: + context.est_plan = plan + return + context.est_plan = context.est_lifecycle.execute_plan(plan_id) diff --git a/robot/estimation_actor.robot b/robot/estimation_actor.robot new file mode 100644 index 000000000..5aadee731 --- /dev/null +++ b/robot/estimation_actor.robot @@ -0,0 +1,55 @@ +*** Settings *** +Documentation Integration tests for the estimation actor type. +... Covers EstimationResult model, EstimationStubActor, +... Plan estimation_result field, ActorRole.ESTIMATOR, +... and estimation invocation during plan lifecycle. +... Issue #890. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_estimation_actor.py + +*** Test Cases *** +EstimationResult Domain Model + [Documentation] Verify EstimationResult creation, defaults, serialization, and immutability + [Tags] estimation model critical + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-result-model cwd=${WORKSPACE} on_timeout=kill timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} estimation-result-model-ok + +EstimationStubActor Placeholder Output + [Documentation] Verify stub actor returns placeholder estimation with correct summary + [Tags] estimation stub critical + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-stub-actor cwd=${WORKSPACE} on_timeout=kill timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} estimation-stub-actor-ok + +ActorRole ESTIMATOR Enum Value + [Documentation] Verify ActorRole enum includes ESTIMATOR with value "estimator" + [Tags] estimation enum + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} actor-role-estimator cwd=${WORKSPACE} on_timeout=kill timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} actor-role-estimator-ok + +Plan Model Estimation Result Field + [Documentation] Verify Plan model accepts estimation_result and includes it in CLI dict + [Tags] estimation plan model + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} plan-estimation-field cwd=${WORKSPACE} on_timeout=kill timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} plan-estimation-field-ok + +Estimation Runs During Strategize To Execute Transition + [Documentation] Verify estimation actor is invoked when transitioning from Strategize to Execute + [Tags] estimation lifecycle critical + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-lifecycle-integration cwd=${WORKSPACE} on_timeout=kill timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} estimation-lifecycle-integration-ok + +Estimation Skipped When No Actor Configured + [Documentation] Verify estimation is skipped when no estimation_actor is set on the plan + [Tags] estimation lifecycle + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} no-estimation-lifecycle cwd=${WORKSPACE} on_timeout=kill timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} no-estimation-lifecycle-ok diff --git a/robot/helper_estimation_actor.py b/robot/helper_estimation_actor.py new file mode 100644 index 000000000..5f11cac1b --- /dev/null +++ b/robot/helper_estimation_actor.py @@ -0,0 +1,229 @@ +"""Helper utilities for estimation actor Robot integration tests. + +Covers: +- EstimationResult domain model creation and serialization +- EstimationStubActor placeholder output +- Plan model estimation_result field +- Estimation invocation during Strategize→Execute transition +- ActorRole.ESTIMATOR enum value + +Issue #890. +""" + +from __future__ import annotations + +import sys + +from cleveragents.application.services.plan_executor import EstimationStubActor +from cleveragents.application.services.plan_lifecycle_service import ( + PlanLifecycleService, +) +from cleveragents.config.settings import Settings +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.acms.tiers import ActorRole +from cleveragents.domain.models.core.estimation import EstimationResult +from cleveragents.domain.models.core.plan import ( + PlanPhase, + ProcessingState, +) + + +def _create_service() -> PlanLifecycleService: + settings = Settings() + return PlanLifecycleService(settings=settings) + + +def _estimation_result_model() -> None: + """Test EstimationResult domain model creation and serialization.""" + # All fields + result = EstimationResult( + estimated_cost_usd=0.05, + estimated_tokens=12000, + estimated_steps=5, + estimated_child_plans=2, + estimated_time_seconds=30.0, + risk_level="medium", + risk_factors=("API rate limits", "Large codebase"), + summary="Estimated 5 steps, ~$0.05 cost", + raw_output="raw output", + ) + assert result.estimated_cost_usd == 0.05 + assert result.estimated_tokens == 12000 + assert result.estimated_steps == 5 + assert result.estimated_child_plans == 2 + assert result.estimated_time_seconds == 30.0 + assert result.risk_level == "medium" + assert len(result.risk_factors) == 2 + assert result.summary == "Estimated 5 steps, ~$0.05 cost" + + # Defaults + default_result = EstimationResult() + assert default_result.estimated_cost_usd is None + assert default_result.estimated_tokens is None + assert default_result.summary == "" + assert default_result.risk_factors == () + + # Serialization round-trip + data = result.model_dump() + restored = EstimationResult(**data) + assert restored == result + + # Frozen check + try: + result.summary = "modified" # type: ignore[misc] + raise AssertionError("Should have raised ValidationError") + except Exception: + pass # Expected: frozen model + + print("estimation-result-model-ok") + + +def _estimation_stub_actor() -> None: + """Test EstimationStubActor placeholder output.""" + stub = EstimationStubActor() + + # Valid plan ID + result = stub.estimate("01ARZ3NDEKTSV4RRFFQ69G5FAV") + assert isinstance(result, EstimationResult) + assert "stub" in result.summary.lower() + + # Empty plan ID should raise + try: + stub.estimate("") + raise AssertionError("Should have raised ValidationError") + except ValidationError: + pass + + print("estimation-stub-actor-ok") + + +def _actor_role_estimator() -> None: + """Test ActorRole.ESTIMATOR enum value.""" + assert hasattr(ActorRole, "ESTIMATOR") + assert ActorRole.ESTIMATOR.value == "estimator" + assert ActorRole("estimator") == ActorRole.ESTIMATOR + print("actor-role-estimator-ok") + + +def _plan_estimation_field() -> None: + """Test Plan model estimation_result field.""" + service = _create_service() + + # Create action with estimation actor + service.create_action( + name="local/est-robot-test", + description="Robot test action", + definition_of_done="Tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + estimation_actor="local/estimator", + ) + + plan = service.use_action(action_name="local/est-robot-test") + assert plan.estimation_actor == "local/estimator" + assert plan.estimation_result is None + + # Set estimation result + plan.estimation_result = EstimationResult( + estimated_cost_usd=1.5, + summary="Test estimation", + ) + assert plan.estimation_result is not None + assert plan.estimation_result.estimated_cost_usd == 1.5 + + # CLI dict + cli_dict = plan.as_cli_dict() + assert "estimation" in cli_dict + assert cli_dict["estimation"]["estimated_cost_usd"] == 1.5 + + print("plan-estimation-field-ok") + + +def _estimation_lifecycle_integration() -> None: + """Test estimation runs during Strategize→Execute transition.""" + service = _create_service() + + # Action WITH estimation actor + service.create_action( + name="local/est-lifecycle-robot", + description="Lifecycle test", + definition_of_done="Tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + estimation_actor="local/estimator", + ) + + plan = service.use_action(action_name="local/est-lifecycle-robot") + plan_id = plan.identity.plan_id + + # Drive through strategize + service.start_strategize(plan_id) + p = service.get_plan(plan_id) + p.processing_state = ProcessingState.PROCESSING + service._commit_plan(p) + service.complete_strategize(plan_id) + + # Check the plan after complete_strategize — auto_progress might + # have advanced it to Execute already (triggering estimation). + p = service.get_plan(plan_id) + if p.phase == PlanPhase.STRATEGIZE: + service.execute_plan(plan_id) + p = service.get_plan(plan_id) + + assert p.estimation_result is not None, ( + "estimation_result should be set after execute_plan" + ) + assert "stub" in p.estimation_result.summary.lower() + + print("estimation-lifecycle-integration-ok") + + +def _no_estimation_lifecycle() -> None: + """Test estimation is skipped when no estimation_actor is set.""" + service = _create_service() + + # Action WITHOUT estimation actor + service.create_action( + name="local/no-est-lifecycle-robot", + description="No estimation test", + definition_of_done="Tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + ) + + plan = service.use_action(action_name="local/no-est-lifecycle-robot") + plan_id = plan.identity.plan_id + + # Drive through strategize + service.start_strategize(plan_id) + p = service.get_plan(plan_id) + p.processing_state = ProcessingState.PROCESSING + service._commit_plan(p) + service.complete_strategize(plan_id) + + p = service.get_plan(plan_id) + if p.phase == PlanPhase.STRATEGIZE: + service.execute_plan(plan_id) + p = service.get_plan(plan_id) + + assert p.estimation_result is None, ( + "estimation_result should be None when no estimation_actor is set" + ) + + print("no-estimation-lifecycle-ok") + + +_COMMANDS = { + "estimation-result-model": _estimation_result_model, + "estimation-stub-actor": _estimation_stub_actor, + "actor-role-estimator": _actor_role_estimator, + "plan-estimation-field": _plan_estimation_field, + "estimation-lifecycle-integration": _estimation_lifecycle_integration, + "no-estimation-lifecycle": _no_estimation_lifecycle, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>") + sys.exit(1) + _COMMANDS[sys.argv[1]]() diff --git a/src/cleveragents/application/services/plan_executor.py b/src/cleveragents/application/services/plan_executor.py index 792f76e18..1803385cb 100644 --- a/src/cleveragents/application/services/plan_executor.py +++ b/src/cleveragents/application/services/plan_executor.py @@ -30,6 +30,7 @@ from cleveragents.application.services.plan_execution_context import ( ) from cleveragents.core.exceptions import PlanError, ValidationError from cleveragents.domain.models.core.change import ChangeSetStore +from cleveragents.domain.models.core.estimation import EstimationResult from cleveragents.domain.models.core.plan import ( PlanInvariant, PlanPhase, @@ -250,6 +251,30 @@ class ExecuteStubActor: ) +class EstimationStubActor: + """Local estimation stub that produces a placeholder estimate. + + Used when no real estimation actor is configured. Returns an + ``EstimationResult`` with a summary indicating that estimation was + skipped. + """ + + def estimate(self, plan_id: str) -> EstimationResult: + """Produce a placeholder estimation result. + + Args: + plan_id: The plan ULID (used for logging/tracing). + + Returns: + An ``EstimationResult`` with a stub summary. + """ + if not plan_id: + raise ValidationError("plan_id must not be empty") + return EstimationResult( + summary="Estimation skipped (stub actor)", + ) + + # --------------------------------------------------------------------------- # Orchestrator # --------------------------------------------------------------------------- diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 3b0d22504..7e0097d00 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -281,6 +281,48 @@ class PlanLifecycleService: exc_info=True, ) + def _run_estimation(self, plan: Plan) -> None: + """Run the estimation actor if configured on the plan. + + Currently uses a stub implementation. When + ``plan.estimation_actor`` is set (and is not the + ``__optional__`` placeholder), invokes the estimation stub + actor to produce an ``EstimationResult``. The result is stored + on the plan and persisted. + + Estimation is informational only — failures are logged but + never block the Execute transition. + """ + actor_name = plan.estimation_actor + if not actor_name or actor_name == "__optional__": + return + + try: + from cleveragents.application.services.plan_executor import ( + EstimationStubActor, + ) + + # TODO: Replace EstimationStubActor with real actor dispatch + # via actor registry + stub = EstimationStubActor() + result = stub.estimate(plan.identity.plan_id) + plan.estimation_result = result + plan.timestamps.updated_at = datetime.now() + self._commit_plan(plan) + self._logger.info( + "Estimation completed", + plan_id=plan.identity.plan_id, + estimation_actor=actor_name, + summary=result.summary, + ) + except Exception: + self._logger.warning( + "estimation_actor_failed", + plan_id=plan.identity.plan_id, + estimation_actor=actor_name, + exc_info=True, + ) + def _persist_action_create(self, action: Action, ctx: Any) -> None: """Persist a new action via the repository. @@ -1043,6 +1085,9 @@ class PlanLifecycleService: plan_id, plan.phase, plan.state or ProcessingState.QUEUED ) + # Run estimation actor if configured (informational only) + self._run_estimation(plan) + # Layer 4: Consult Error Pattern Database for preventive guidance self._consult_error_patterns(plan) diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 64b669a3d..7712e067a 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -157,6 +157,8 @@ def _plan_spec_dict(plan: Any) -> dict[str, object]: {"text": inv.text, "source": inv.source.value} for inv in plan.invariants ] + if plan.estimation_result is not None: + result["estimation"] = plan.estimation_result.as_display_dict() if plan.validation_summary and plan.validation_summary.get("dod_evaluated"): result["dod_evaluation"] = { "all_passed": plan.validation_summary.get("dod_all_passed", False), @@ -1349,6 +1351,30 @@ def _print_lifecycle_plan(plan: Any, title: str = "Plan") -> None: if plan.invariant_actor: details += f"[bold]Invariant Actor:[/bold] {plan.invariant_actor}\n" + # Estimation results + # TODO: Migrate to EstimationResult.as_display_dict() to reduce duplication + if plan.estimation_result is not None: + est = plan.estimation_result + details += "[bold]Estimation:[/bold]\n" + if est.summary: + details += f" Summary: {est.summary}\n" + if est.estimated_cost_usd is not None: + details += f" Est. Cost: ${est.estimated_cost_usd:.4f}\n" + if est.estimated_tokens is not None: + details += f" Est. Tokens: {est.estimated_tokens}\n" + if est.estimated_steps is not None: + details += f" Est. Steps: {est.estimated_steps}\n" + if est.estimated_child_plans is not None: + details += f" Est. Child Plans: {est.estimated_child_plans}\n" + if est.estimated_time_seconds is not None: + details += f" Est. Time: {est.estimated_time_seconds:.1f}s\n" + if est.risk_level is not None: + details += f" Risk Level: {est.risk_level}\n" + if est.risk_factors: + details += " Risk Factors:\n" + for factor in est.risk_factors: + details += f" - {factor}\n" + # Arguments (ordered) if plan.arguments: details += "[bold]Arguments:[/bold]\n" diff --git a/src/cleveragents/domain/models/acms/tiers.py b/src/cleveragents/domain/models/acms/tiers.py index ca1792cce..9e1486f35 100644 --- a/src/cleveragents/domain/models/acms/tiers.py +++ b/src/cleveragents/domain/models/acms/tiers.py @@ -56,6 +56,7 @@ class ActorRole(StrEnum): STRATEGIST = "strategist" EXECUTOR = "executor" REVIEWER = "reviewer" + ESTIMATOR = "estimator" # --------------------------------------------------------------------------- @@ -169,6 +170,7 @@ _DEFAULT_VISIBLE_TIERS: dict[ActorRole, list[ContextTier]] = { ActorRole.STRATEGIST: [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD], ActorRole.EXECUTOR: [ContextTier.HOT, ContextTier.WARM], ActorRole.REVIEWER: [ContextTier.HOT], + ActorRole.ESTIMATOR: [ContextTier.HOT, ContextTier.WARM], } diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 8847edab4..c2ca761ef 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -129,6 +129,7 @@ from cleveragents.domain.models.core.escalation import ( HistoricalOutcome, OperationContext, ) +from cleveragents.domain.models.core.estimation import EstimationResult from cleveragents.domain.models.core.invariant import ( Invariant, InvariantEnforcementRecord, @@ -374,6 +375,7 @@ __all__ = [ "ErrorRecord", "ErrorRecoveryPolicy", "EscalationDecision", + "EstimationResult", "ExecutionEnvironment", "FileRecord", "FragmentProvenance", diff --git a/src/cleveragents/domain/models/core/estimation.py b/src/cleveragents/domain/models/core/estimation.py new file mode 100644 index 000000000..0f38c95f8 --- /dev/null +++ b/src/cleveragents/domain/models/core/estimation.py @@ -0,0 +1,117 @@ +"""Estimation result domain model for CleverAgents. + +Provides the ``EstimationResult`` value object returned by the +estimation actor after the Strategize phase completes and before +Execute begins. All fields are optional — the estimation actor +populates whichever fields it can compute. + +The result is informational only; it does **not** gate the +Execute phase. + +Based on ``docs/specification.md`` estimation actor sections and +issue #890. +""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +__all__ = ["EstimationResult"] + + +class EstimationResult(BaseModel): + """Cost, time, and risk estimation produced by the estimation actor. + + Every field is optional so that estimation actors can populate only + the metrics they are able to compute. The ``summary`` field + provides a human-readable overview and ``raw_output`` preserves the + unprocessed actor response for auditing. + """ + + estimated_cost_usd: float | None = Field( + default=None, + ge=0.0, + description="Estimated LLM cost in USD", + ) + estimated_tokens: int | None = Field( + default=None, + ge=0, + description="Estimated total token usage", + ) + estimated_steps: int | None = Field( + default=None, + ge=0, + description="Expected number of execution steps", + ) + estimated_child_plans: int | None = Field( + default=None, + description="Expected number of child plans", + ) + estimated_time_seconds: float | None = Field( + default=None, + ge=0.0, + description="Estimated execution time in seconds", + ) + risk_level: Literal["low", "medium", "high", "critical"] | None = Field( + default=None, + description="Risk assessment: low, medium, high, critical", + ) + risk_factors: tuple[str, ...] = Field( + default_factory=tuple, + description="Identified risk factors", + ) + summary: str = Field( + default="", + max_length=10_000, + description="Human-readable estimation summary", + ) + raw_output: str = Field( + default="", + max_length=100_000, + description="Raw actor output before parsing", + ) + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + @field_validator("risk_factors") + @classmethod + def _cap_risk_factors(cls, v: tuple[str, ...]) -> tuple[str, ...]: + if len(v) > 100: + raise ValueError("risk_factors must contain at most 100 items") + return v + + def as_display_dict(self) -> dict[str, Any]: + """Return a dict of non-empty fields suitable for display/CLI output. + + Only fields that are set (not ``None`` and not empty) are + included, keeping serialized output compact. + + .. note:: + + Three call-sites currently duplicate this logic and should + migrate to this method: + + - ``Plan.as_cli_dict()`` in ``plan.py`` + - ``_build_plan_summary()`` in ``cli/commands/plan.py`` + - ``_show_plan_detail()`` in ``cli/commands/plan.py`` + """ + d: dict[str, Any] = {} + if self.estimated_cost_usd is not None: + d["estimated_cost_usd"] = self.estimated_cost_usd + if self.estimated_tokens is not None: + d["estimated_tokens"] = self.estimated_tokens + if self.estimated_steps is not None: + d["estimated_steps"] = self.estimated_steps + if self.estimated_child_plans is not None: + d["estimated_child_plans"] = self.estimated_child_plans + if self.estimated_time_seconds is not None: + d["estimated_time_seconds"] = self.estimated_time_seconds + if self.risk_level is not None: + d["risk_level"] = self.risk_level + if self.risk_factors: + d["risk_factors"] = list(self.risk_factors) + if self.summary: + d["summary"] = self.summary + return d diff --git a/src/cleveragents/domain/models/core/plan.py b/src/cleveragents/domain/models/core/plan.py index c14b02cc8..d515b811b 100644 --- a/src/cleveragents/domain/models/core/plan.py +++ b/src/cleveragents/domain/models/core/plan.py @@ -64,6 +64,7 @@ from typing import Any, ClassVar from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from cleveragents.domain.models.core.cost_metadata import CostMetadata +from cleveragents.domain.models.core.estimation import EstimationResult from cleveragents.domain.models.core.multi_project import ( MultiProjectMetadata, ProjectScope, @@ -612,6 +613,10 @@ class Plan(BaseModel): default=None, description="Optional actor for cost/risk estimation", ) + estimation_result: EstimationResult | None = Field( + default=None, + description="Cost/risk estimation result from the estimation actor", + ) invariant_actor: str | None = Field( default=None, description="Optional actor for invariant reconciliation", @@ -987,6 +992,8 @@ class Plan(BaseModel): result["parent_plan_id"] = self.identity.parent_plan_id if self.has_subplans: result["subplan_count"] = len(self.subplan_statuses) + if self.estimation_result is not None: + result["estimation"] = self.estimation_result.as_display_dict() if self.cost_metadata is not None: result["cost"] = self.cost_metadata.as_display_dict() if self.skeleton_metadata is not None: diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 9efd7b69b..24bece56c 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -194,6 +194,7 @@ _SUPPORTED_SOURCES # noqa: B018, F821 # Plan executor — public API StrategizeStubActor # noqa: B018, F821 ExecuteStubActor # noqa: B018, F821 +EstimationStubActor # noqa: B018, F821 PlanExecutor # noqa: B018, F821 StrategyDecision # noqa: B018, F821 StrategizeResult # noqa: B018, F821 @@ -722,6 +723,20 @@ traverse # noqa: B018, F821 # ACMS Pipeline — SkeletonCompressor protocol parameter (required by interface) skeleton_budget # noqa: B018, F821 +# Estimation actor — public API (M5, issue #890) +EstimationResult # noqa: B018, F821 +estimation_result # noqa: B018, F821 +estimated_cost_usd # noqa: B018, F821 +estimated_tokens # noqa: B018, F821 +estimated_steps # noqa: B018, F821 +estimated_child_plans # noqa: B018, F821 +estimated_time_seconds # noqa: B018, F821 +risk_level # noqa: B018, F821 +risk_factors # noqa: B018, F821 +raw_output # noqa: B018, F821 +_run_estimation # noqa: B018, F821 +ESTIMATOR # noqa: B018, F821 + # Context Tiers — public API (M6 ACMS, issue #208) ContextTier # noqa: B018, F821 ActorRole # noqa: B018, F821 -- 2.52.0