feat(actor): implement estimation actor type (#962)
CI / build (push) Successful in 24s
CI / lint (push) Successful in 3m16s
CI / quality (push) Successful in 3m45s
CI / typecheck (push) Successful in 3m52s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 4m1s
CI / unit_tests (push) Successful in 7m8s
CI / integration_tests (push) Successful in 7m1s
CI / docker (push) Successful in 1m6s
CI / e2e_tests (push) Successful in 9m49s
CI / coverage (push) Successful in 12m22s
CI / status-check (push) Successful in 1s
CI / benchmark-publish (push) Successful in 19m34s

## Summary

Implement the estimation actor as a functional actor type. The estimation actor provides cost, time, and resource estimates for plan operations, running after Strategize completes (before Execute). Estimation is informational only and optional — plans work without an estimation actor configured.

### Changes

**New files:**
- `src/cleveragents/domain/models/core/estimation.py` — `EstimationResult` frozen Pydantic model with fields for cost (USD), tokens, steps, child plans, time, risk level/factors, summary
- `features/estimation_actor.feature` — 12 Behave test scenarios (model validation, serialization, stub actor, plan integration, optional behavior)
- `features/steps/estimation_actor_steps.py` — Step definitions
- `robot/estimation_actor.robot` — 6 Robot integration tests
- `robot/helper_estimation_actor.py` — Robot test helper

**Modified files:**
- `domain/models/acms/tiers.py` — Added `ESTIMATOR` to `ActorRole` enum
- `domain/models/core/plan.py` — Added `estimation_result: EstimationResult | None` field, estimation data in `as_cli_dict()`
- `application/services/plan_executor.py` — Added `EstimationStubActor` class
- `application/services/plan_lifecycle_service.py` — Added `_run_estimation()` method, invoked in `execute_plan()`
- `cli/commands/plan.py` — Display estimation results in plan status
- `vulture_whitelist.py` — Added 12 new public symbols

### Quality Gates

| Session | Result |
|---|---|
| `nox -s lint` | PASS |
| `nox -s typecheck` | PASS (0 errors) |
| `nox -s unit_tests` | PASS (10,818 scenarios) |
| `nox -s integration_tests` | PASS (1,512 tests) |
| `nox -s coverage_report` | 98% (>= 97%) |

Closes #890

Reviewed-on: #962
Co-authored-by: Brent Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent Edwards <brent.edwards@cleverthis.com>
This commit was merged in pull request #962.
This commit is contained in:
2026-03-21 01:40:58 +00:00
committed by Forgejo
parent 95d3e09925
commit 6531440431
14 changed files with 1017 additions and 4 deletions
+2 -2
View File
@@ -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 ----
+95
View File
@@ -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
+3 -2
View File
@@ -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]
# ---------------------------------------------------------------------------
+394
View File
@@ -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)
+55
View File
@@ -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
+229
View File
@@ -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 StrategizeExecute 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]]()
@@ -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
# ---------------------------------------------------------------------------
@@ -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)
+26
View File
@@ -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"
@@ -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],
}
@@ -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.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
@@ -379,6 +380,7 @@ __all__ = [
"ErrorRecord",
"ErrorRecoveryPolicy",
"EscalationDecision",
"EstimationResult",
"ExecutionEnvironment",
"ExecutionEnvironmentPreference",
"FileRecord",
@@ -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
@@ -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:
+15
View File
@@ -216,6 +216,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
@@ -750,6 +751,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