Compare commits

...

1 Commits

Author SHA1 Message Date
aditya 3f6b255747 feat(estimation): add cost and risk estimation actor
CI / build (pull_request) Successful in 18s
CI / helm (pull_request) Successful in 21s
CI / benchmark-publish (pull_request) Has been skipped
CI / security (pull_request) Successful in 59s
CI / lint (pull_request) Successful in 3m18s
CI / typecheck (pull_request) Successful in 4m7s
CI / quality (pull_request) Successful in 4m13s
CI / unit_tests (pull_request) Failing after 6m5s
CI / docker (pull_request) Has been skipped
CI / coverage (pull_request) Successful in 9m14s
CI / e2e_tests (pull_request) Failing after 15m31s
CI / integration_tests (pull_request) Successful in 24m14s
CI / status-check (pull_request) Failing after 2s
CI / benchmark-regression (pull_request) Successful in 55m20s
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
2026-04-02 10:26:22 +00:00
18 changed files with 1145 additions and 11 deletions
+9
View File
@@ -2,6 +2,15 @@
## Unreleased
- Added cost and risk estimation actor with `EstimationReport` structured
domain model (cost ranges, expected steps/child plans, rollback risk,
confidence, rationale, historical basis), `EstimationSkipped` model for
opt-out/failure tracking, `ESTIMATION_PRODUCED` decision type,
`--no-estimate` CLI flag on `plan use`, and Alembic migration for
`estimation_report_json`/`estimation_skipped_json` columns. Estimation
failures now record `EstimationSkipped` instead of silently logging.
Includes Behave tests, Robot tests, ASV benchmarks, and reference
documentation. (#209)
- Added a context-sensitive TUI help panel overlay toggled by `F1`, with
help content that varies for main-screen, slash-command, reference, and
shell prompt modes. Updated Behave and Robot coverage for help-panel
@@ -0,0 +1,41 @@
"""Add estimation_report and estimation_skipped columns to v3_plans.
Adds TEXT columns for JSON-serialised ``EstimationReport`` and
``EstimationSkipped`` domain models.
Revision ID: m6_006_estimation_report_domain
Revises: m8_002_merge_profile_rename_and_corrections
Create Date: 2026-03-30 00:00:00
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "m6_006_estimation_report_domain"
down_revision: str | Sequence[str] | None = (
"m8_002_merge_profile_rename_and_corrections"
)
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Add estimation_report_json and estimation_skipped_json columns."""
with op.batch_alter_table("v3_plans") as batch_op:
batch_op.add_column(
sa.Column("estimation_report_json", sa.Text(), nullable=True)
)
batch_op.add_column(
sa.Column("estimation_skipped_json", sa.Text(), nullable=True)
)
def downgrade() -> None:
"""Remove estimation_report_json and estimation_skipped_json columns."""
with op.batch_alter_table("v3_plans") as batch_op:
batch_op.drop_column("estimation_skipped_json")
batch_op.drop_column("estimation_report_json")
+127
View File
@@ -0,0 +1,127 @@
"""ASV benchmarks for EstimationReport and EstimationSkipped models.
Measures the performance of:
- EstimationReport model construction (Pydantic validation)
- EstimationReport.model_dump() serialization
- EstimationReport.model_dump_json() JSON serialization
- EstimationReport.model_validate() deserialization
- EstimationSkipped model construction
- EstimationSkipped serialization round-trip
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.estimation import (
EstimationReport,
EstimationSkipped,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.estimation import (
EstimationReport,
EstimationSkipped,
)
def _make_full_report() -> EstimationReport:
"""Create a fully-populated estimation report for benchmarking."""
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 code review tasks in the last month.",
historical_basis=("plan-001", "plan-002", "plan-003"),
)
class EstimationReportConstructionSuite:
"""Benchmark EstimationReport construction."""
def time_minimal_construction(self) -> None:
"""Benchmark minimal EstimationReport creation."""
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="Minimal estimation.",
)
def time_full_construction(self) -> None:
"""Benchmark fully-populated EstimationReport creation."""
_make_full_report()
def time_with_historical_basis(self) -> None:
"""Benchmark EstimationReport with 50 historical basis entries."""
EstimationReport(
cost_range_usd_min=0.10,
cost_range_usd_max=1.00,
expected_steps=20,
expected_child_plans=5,
rollback_risk=0.3,
estimated_duration_minutes=15.0,
confidence=0.7,
rationale="Large task with extensive history.",
historical_basis=tuple(f"plan-{i:03d}" for i in range(50)),
)
class EstimationReportSerializationSuite:
"""Benchmark EstimationReport serialization."""
def setup(self) -> None:
"""Create objects for serialization benchmarks."""
self.report = _make_full_report()
self.report_dict = self.report.model_dump()
self.report_json = self.report.model_dump_json()
def time_model_dump(self) -> None:
"""Benchmark model_dump() serialization."""
self.report.model_dump()
def time_model_dump_json(self) -> None:
"""Benchmark model_dump_json() JSON serialization."""
self.report.model_dump_json()
def time_model_validate_dict(self) -> None:
"""Benchmark model_validate() from dict."""
EstimationReport.model_validate(self.report_dict)
def time_model_validate_json(self) -> None:
"""Benchmark model_validate_json() from JSON string."""
EstimationReport.model_validate_json(self.report_json)
class EstimationSkippedSuite:
"""Benchmark EstimationSkipped construction and serialization."""
def time_construction(self) -> None:
"""Benchmark EstimationSkipped creation."""
EstimationSkipped(
reason="User opted out via --no-estimate",
actor_name="local/estimation-actor",
)
def time_construction_minimal(self) -> None:
"""Benchmark minimal EstimationSkipped creation."""
EstimationSkipped(reason="Skipped")
def time_round_trip(self) -> None:
"""Benchmark EstimationSkipped serialization round-trip."""
skipped = EstimationSkipped(
reason="Estimation actor failed: timeout",
actor_name="local/estimation-actor",
)
data = skipped.model_dump()
EstimationSkipped.model_validate(data)
+86
View File
@@ -0,0 +1,86 @@
# Estimation Feature Reference
The estimation subsystem provides cost and risk estimation for plans
before the Execute phase begins. Estimation is **informational only** --
it does not gate execution.
## Overview
When a plan transitions from Strategize to Execute, the estimation actor
(if configured) produces an `EstimationReport` containing cost ranges,
expected work metrics, risk assessment, and rationale.
Users can opt out of estimation with the `--no-estimate` flag on
`plan use`, which records an `EstimationSkipped` record instead.
## EstimationReport Schema
| Field | Type | Constraints |
|----------------------------|-------------------|----------------------|
| `cost_range_usd_min` | `float` | >= 0.0 |
| `cost_range_usd_max` | `float` | >= 0.0, >= min |
| `expected_steps` | `int` | >= 0 |
| `expected_child_plans` | `int` | >= 0 |
| `rollback_risk` | `float` | 0.0 - 1.0 |
| `estimated_duration_minutes`| `float` | >= 0.0 |
| `confidence` | `float` | 0.0 - 1.0 |
| `rationale` | `str` | 1 - 10,000 chars |
| `historical_basis` | `tuple[str, ...]` | max 100 items |
The model is frozen (immutable) and rejects `inf`/`nan` values.
## EstimationSkipped Schema
| Field | Type | Description |
|--------------|------------------|-------------------------------------|
| `reason` | `str` | Why estimation was skipped |
| `timestamp` | `datetime` | When estimation was skipped |
| `actor_name` | `str` or `None` | Estimation actor name if applicable |
## Mutual Exclusion
A plan should have either `estimation_report` or `estimation_skipped`
set, but not both. This is enforced at the application layer:
- When `--no-estimate` is passed, `estimation_skipped` is set and the
estimation actor is cleared.
- When estimation succeeds, `estimation_report` is set.
- When estimation fails, `estimation_skipped` is set with the failure
reason.
## CLI Usage
### Skip estimation
```bash
agents plan use local/code-review my-project --no-estimate
```
When `--no-estimate` is passed:
- `estimation_skipped` is set with reason "User opted out via --no-estimate"
- The estimation actor is cleared from the plan
- No estimation is attempted during the Strategize-to-Execute transition
### View estimation data in plan status
```bash
agents plan status <plan_id> --format json | jq '.estimation'
```
The `plan status` command displays estimation data when available:
- `estimation` section shows the `EstimationReport` fields
- `estimation_skipped` section shows why estimation was skipped
## Database Storage
Both `EstimationReport` and `EstimationSkipped` are stored as
JSON-serialised TEXT columns on the `v3_plans` table:
- `estimation_report_json` -- serialised `EstimationReport`
- `estimation_skipped_json` -- serialised `EstimationSkipped`
## Legacy EstimationResult
The original `EstimationResult` model (all optional fields) is retained
for backward compatibility with the `EstimationStubActor`. New code
should prefer `EstimationReport` for structured estimation data.
+6 -4
View File
@@ -6,8 +6,8 @@ Feature: Consolidated Decision
# Feature: Decision domain model
# ============================================================
Scenario: All 11 decision types are defined
Then the DecisionType enum should have exactly 11 members
Scenario: All 12 decision types are defined
Then the DecisionType enum should have exactly 12 members
Scenario: Strategize-phase types are correctly classified
@@ -18,7 +18,8 @@ Feature: Consolidated Decision
And STRATEGIZE_TYPES should contain "subplan_spawn"
And STRATEGIZE_TYPES should contain "subplan_parallel_spawn"
And STRATEGIZE_TYPES should contain "user_intervention"
And STRATEGIZE_TYPES should have exactly 7 members
And STRATEGIZE_TYPES should contain "estimation_produced"
And STRATEGIZE_TYPES should have exactly 8 members
Scenario: Execute-phase types are correctly classified
@@ -290,7 +291,7 @@ Feature: Consolidated Decision
And as_cli_dict should contain key "parent"
# ------------------------------------------------------------------
# All 11 decision types can be instantiated
# All 12 decision types can be instantiated
# ------------------------------------------------------------------
@@ -312,6 +313,7 @@ Feature: Consolidated Decision
| error_recovery |
| validation_response |
| user_intervention |
| estimation_produced |
# ============================================================
+83
View File
@@ -0,0 +1,83 @@
Feature: Estimation skip behavior
Plans can skip estimation via --no-estimate or when the estimation actor fails.
# ------------------------------------------------------------------
# EstimationSkipped model
# ------------------------------------------------------------------
Scenario: Create an EstimationSkipped with required fields
When I create an EstimationSkipped with only reason "User opted out"
Then the EstimationSkipped should be created successfully
And the EstimationSkipped reason should be "User opted out"
And the EstimationSkipped actor_name should be None
Scenario: Create an EstimationSkipped with actor name
When I create an EstimationSkipped with reason "Actor failed" and actor "local/estimator"
Then the EstimationSkipped should be created successfully
And the EstimationSkipped actor_name should be "local/estimator"
Scenario: EstimationSkipped is frozen
When I create an EstimationSkipped with only reason "Frozen test"
Then modifying the EstimationSkipped reason should raise an error
Scenario: EstimationSkipped round-trips through serialization
When I create an EstimationSkipped with reason "Round-trip test" and actor "local/est"
Then the EstimationSkipped should round-trip through model_dump and model_validate
Scenario: EstimationSkipped requires non-empty reason
When I try to create an EstimationSkipped with empty reason
Then an estimation skip validation error should be raised
# ------------------------------------------------------------------
# EstimationReport model
# ------------------------------------------------------------------
Scenario: Create a valid EstimationReport
When I create a valid EstimationReport
Then the EstimationReport should be created successfully
And the EstimationReport confidence should be 0.85
Scenario: EstimationReport rejects max < min cost
When I try to create an EstimationReport with max cost less than min cost
Then an estimation report validation error should be raised
Scenario: EstimationReport rejects confidence above 1.0
When I try to create an EstimationReport with confidence 1.5
Then an estimation report validation error should be raised
Scenario: EstimationReport rejects negative rollback risk
When I try to create an EstimationReport with rollback risk -0.1
Then an estimation report validation error should be raised
Scenario: EstimationReport caps historical basis at 100
When I try to create an EstimationReport with 101 historical basis entries
Then an estimation report validation error should be raised
Scenario: EstimationReport is frozen
When I create a valid EstimationReport
Then modifying the EstimationReport rationale should raise an error
# ------------------------------------------------------------------
# Mutual exclusion on Plan
# ------------------------------------------------------------------
Scenario: Plan with estimation_skipped but no estimation_report
When I create a plan with estimation_skipped set
Then the plan estimation_skipped should not be None
And the plan estimation_report should be None
Scenario: Plan with estimation_report but no estimation_skipped
When I create a plan with estimation_report set
Then the plan estimation_report should not be None
And the plan estimation_skipped should be None
# ------------------------------------------------------------------
# --no-estimate CLI flag behavior
# ------------------------------------------------------------------
Scenario: --no-estimate sets estimation_skipped on plan
Given a lifecycle service with a test action
When I use the action with --no-estimate
Then the plan should have estimation_skipped set
And the estimation_skipped reason should mention "opted out"
And the plan estimation_actor should be None
+323
View File
@@ -0,0 +1,323 @@
"""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.
+51
View File
@@ -0,0 +1,51 @@
*** Settings ***
Documentation Integration tests for EstimationSkipped and --no-estimate behavior
Library Process
Suite Setup Log Estimation skip robot tests starting
*** Variables ***
${HELPER} ${CURDIR}/helper_estimation_skip.py
${PYTHONPATH} ${CURDIR}/../src
*** Test Cases ***
EstimationSkipped Model Creation
[Documentation] Verify EstimationSkipped model creation and validation
${result}= Run Process python ${HELPER} model-create
... env:PYTHONPATH=${PYTHONPATH} timeout=30s
Should Contain ${result.stdout} estimation-skipped-model-create-ok
Should Be Equal As Integers ${result.rc} 0
EstimationSkipped Frozen Check
[Documentation] Verify EstimationSkipped model is frozen (immutable)
${result}= Run Process python ${HELPER} frozen-check
... env:PYTHONPATH=${PYTHONPATH} timeout=30s
Should Contain ${result.stdout} estimation-skipped-frozen-ok
Should Be Equal As Integers ${result.rc} 0
EstimationSkipped Serialization Round-trip
[Documentation] Verify EstimationSkipped survives serialization round-trip
${result}= Run Process python ${HELPER} round-trip
... env:PYTHONPATH=${PYTHONPATH} timeout=30s
Should Contain ${result.stdout} estimation-skipped-round-trip-ok
Should Be Equal As Integers ${result.rc} 0
EstimationReport Model Creation
[Documentation] Verify EstimationReport model creation
${result}= Run Process python ${HELPER} report-create
... env:PYTHONPATH=${PYTHONPATH} timeout=30s
Should Contain ${result.stdout} estimation-report-create-ok
Should Be Equal As Integers ${result.rc} 0
EstimationReport Validation
[Documentation] Verify EstimationReport validation rules
${result}= Run Process python ${HELPER} report-validation
... env:PYTHONPATH=${PYTHONPATH} timeout=30s
Should Contain ${result.stdout} estimation-report-validation-ok
Should Be Equal As Integers ${result.rc} 0
No Estimate CLI Integration
[Documentation] Verify --no-estimate sets estimation_skipped on plan
${result}= Run Process python ${HELPER} no-estimate-integration
... env:PYTHONPATH=${PYTHONPATH} timeout=30s
Should Contain ${result.stdout} no-estimate-integration-ok
Should Be Equal As Integers ${result.rc} 0
+3 -2
View File
@@ -55,7 +55,7 @@ def _test_create_child() -> None:
def _test_enum_values() -> None:
"""Verify all 11 decision type enum values."""
"""Verify all 12 decision type enum values."""
expected = {
"prompt_definition",
"invariant_enforced",
@@ -68,10 +68,11 @@ def _test_enum_values() -> None:
"error_recovery",
"validation_response",
"user_intervention",
"estimation_produced",
}
actual = {dt.value for dt in DecisionType}
assert actual == expected, f"Mismatch: {actual.symmetric_difference(expected)}"
assert len(DecisionType) == 11
assert len(DecisionType) == 12
print("decision-enums-ok")
+214
View File
@@ -0,0 +1,214 @@
"""Helper utilities for estimation skip Robot integration tests.
Covers:
- EstimationSkipped domain model creation, validation, serialization
- EstimationReport domain model creation and validation
- --no-estimate CLI integration behavior
Issue #209.
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from pydantic import ValidationError
from cleveragents.domain.models.core.estimation import (
EstimationReport,
EstimationSkipped,
)
def _model_create() -> None:
"""Test EstimationSkipped model creation."""
# With all fields
skipped = EstimationSkipped(
reason="User opted out via --no-estimate",
actor_name="local/estimation-actor",
)
assert skipped.reason == "User opted out via --no-estimate"
assert skipped.actor_name == "local/estimation-actor"
assert skipped.timestamp is not None
# Minimal (no actor)
minimal = EstimationSkipped(reason="Skipped")
assert minimal.actor_name is None
# Empty reason should fail
try:
EstimationSkipped(reason="")
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
print("estimation-skipped-model-create-ok")
def _frozen_check() -> None:
"""Test EstimationSkipped is frozen."""
skipped = EstimationSkipped(reason="Frozen test")
try:
skipped.reason = "modified" # type: ignore[misc]
raise AssertionError("Should have raised error for frozen model")
except (ValidationError, TypeError, AttributeError):
pass
print("estimation-skipped-frozen-ok")
def _round_trip() -> None:
"""Test EstimationSkipped serialization round-trip."""
skipped = EstimationSkipped(
reason="Round-trip test",
actor_name="local/est-actor",
)
data = skipped.model_dump()
restored = EstimationSkipped.model_validate(data)
assert restored.reason == skipped.reason
assert restored.actor_name == skipped.actor_name
# JSON round-trip
json_str = skipped.model_dump_json()
restored_json = EstimationSkipped.model_validate_json(json_str)
assert restored_json.reason == skipped.reason
print("estimation-skipped-round-trip-ok")
def _report_create() -> None:
"""Test EstimationReport model creation."""
report = 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"),
)
assert report.cost_range_usd_min == 0.01
assert report.cost_range_usd_max == 0.50
assert report.expected_steps == 10
assert report.confidence == 0.85
assert len(report.historical_basis) == 2
# Round-trip
data = report.model_dump()
restored = EstimationReport.model_validate(data)
assert restored == report
print("estimation-report-create-ok")
def _report_validation() -> None:
"""Test EstimationReport validation rules."""
# max < min should fail
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.",
)
raise AssertionError("Should have rejected max < min")
except ValidationError:
pass
# confidence > 1.0 should fail
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=1.5,
rationale="Bad confidence.",
)
raise AssertionError("Should have rejected confidence > 1.0")
except ValidationError:
pass
# Too many historical_basis entries
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 entries.",
historical_basis=tuple(f"plan-{i}" for i in range(101)),
)
raise AssertionError("Should have rejected > 100 historical_basis")
except ValidationError:
pass
print("estimation-report-validation-ok")
def _no_estimate_integration() -> None:
"""Test --no-estimate integration with plan lifecycle."""
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
settings = Settings()
service = PlanLifecycleService(settings=settings)
# Create action with estimation actor
service.create_action(
name="local/test-no-est",
description="Test action for no-estimate",
definition_of_done="All tests pass",
strategy_actor="local/mock-strategy",
execution_actor="local/mock-execute",
estimation_actor="local/mock-estimation",
)
# Use the action
plan = service.use_action(action_name="local/test-no-est")
# Simulate --no-estimate behavior (as CLI does)
plan.estimation_skipped = EstimationSkipped(
reason="User opted out via --no-estimate",
actor_name=plan.estimation_actor,
)
plan.estimation_actor = None
assert plan.estimation_skipped is not None
assert "opted out" in plan.estimation_skipped.reason
assert plan.estimation_actor is None
assert plan.estimation_report is None
print("no-estimate-integration-ok")
_COMMANDS = {
"model-create": _model_create,
"frozen-check": _frozen_check,
"round-trip": _round_trip,
"report-create": _report_create,
"report-validation": _report_validation,
"no-estimate-integration": _no_estimate_integration,
}
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]]()
@@ -320,7 +320,15 @@ class PlanLifecycleService:
estimation_actor=actor_name,
summary=result.summary,
)
except Exception:
except Exception as exc:
from cleveragents.domain.models.core.estimation import EstimationSkipped
plan.estimation_skipped = EstimationSkipped(
reason=f"Estimation actor failed: {exc}",
actor_name=actor_name,
)
plan.timestamps.updated_at = datetime.now()
self._commit_plan(plan)
self._logger.warning(
"estimation_actor_failed",
plan_id=plan.identity.plan_id,
+20
View File
@@ -1600,6 +1600,13 @@ def use_action(
help="Priority semantics: fallback (default) or override",
),
] = None,
no_estimate: Annotated[
bool,
typer.Option(
"--no-estimate",
help="Skip cost/risk estimation for this plan",
),
] = False,
fmt: Annotated[
str,
typer.Option(
@@ -1618,6 +1625,7 @@ def use_action(
Examples:
agents plan use local/code-coverage proj-1 proj-2 --arg target_coverage=80
agents plan use local/lint --project proj-1 --invariant "No new warnings"
agents plan use local/code-review proj-1 --no-estimate
"""
from cleveragents.application.services.plan_lifecycle_service import (
ActionNotAvailableError,
@@ -1707,6 +1715,18 @@ def use_action(
# re-persist the plan when necessary.
has_overrides = False
# Handle --no-estimate: skip estimation and record skipped
if no_estimate:
from cleveragents.domain.models.core.estimation import EstimationSkipped
plan.estimation_skipped = EstimationSkipped(
reason="User opted out via --no-estimate",
actor_name=plan.estimation_actor,
)
# Clear estimation actor so lifecycle service skips estimation
plan.estimation_actor = None
has_overrides = True
# Preserve explicit CLI override locally as well so CLI output,
# persistence refresh, and mocked lifecycle-service tests all
# observe the plan-level profile consistently.
@@ -139,7 +139,11 @@ from cleveragents.domain.models.core.escalation import (
HistoricalOutcome,
OperationContext,
)
from cleveragents.domain.models.core.estimation import EstimationResult
from cleveragents.domain.models.core.estimation import (
EstimationReport,
EstimationResult,
EstimationSkipped,
)
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
@@ -397,7 +401,9 @@ __all__ = [
"ErrorRecord",
"ErrorRecoveryPolicy",
"EscalationDecision",
"EstimationReport",
"EstimationResult",
"EstimationSkipped",
"ExecutionEnvironment",
"ExecutionEnvironmentPreference",
"FileRecord",
@@ -105,6 +105,7 @@ class DecisionType(StrEnum):
ERROR_RECOVERY = "error_recovery"
VALIDATION_RESPONSE = "validation_response"
USER_INTERVENTION = "user_intervention"
ESTIMATION_PRODUCED = "estimation_produced"
#: Decision types permitted during the Strategize phase.
@@ -132,6 +133,7 @@ STRATEGIZE_TYPES: frozenset[DecisionType] = frozenset(
DecisionType.SUBPLAN_SPAWN,
DecisionType.SUBPLAN_PARALLEL_SPAWN,
DecisionType.USER_INTERVENTION,
DecisionType.ESTIMATION_PRODUCED,
}
)
@@ -14,11 +14,12 @@ issue #890.
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
__all__ = ["EstimationResult"]
__all__ = ["EstimationReport", "EstimationResult", "EstimationSkipped"]
class EstimationResult(BaseModel):
@@ -115,3 +116,103 @@ class EstimationResult(BaseModel):
if self.summary:
d["summary"] = self.summary
return d
class EstimationReport(BaseModel):
"""Structured estimation report produced by the estimation actor.
Contains cost ranges, expected work metrics, risk assessment, and
rationale. Identical to the model defined in issue #649 to allow
clean merge of the companion PR.
All numeric fields have non-negative constraints. Probability
fields (``rollback_risk``, ``confidence``) are clamped to [0, 1].
"""
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
cost_range_usd_min: float = Field(
...,
ge=0.0,
description="Minimum estimated cost in USD",
)
cost_range_usd_max: float = Field(
...,
ge=0.0,
description="Maximum estimated cost in USD",
)
expected_steps: int = Field(
...,
ge=0,
description="Expected number of execution steps",
)
expected_child_plans: int = Field(
...,
ge=0,
description="Expected number of child plans",
)
rollback_risk: float = Field(
...,
ge=0.0,
le=1.0,
description="Probability that a rollback will be needed (0-1)",
)
estimated_duration_minutes: float = Field(
...,
ge=0.0,
description="Estimated wall-clock duration in minutes",
)
confidence: float = Field(
...,
ge=0.0,
le=1.0,
description="Confidence in the estimation (0-1)",
)
rationale: str = Field(
...,
min_length=1,
max_length=10_000,
description="Human-readable rationale for the estimation",
)
historical_basis: tuple[str, ...] = Field(
default_factory=tuple,
description="References to historical plans used as basis",
)
@field_validator("historical_basis")
@classmethod
def _cap_historical_basis(cls, v: tuple[str, ...]) -> tuple[str, ...]:
if len(v) > 100:
raise ValueError("historical_basis must contain at most 100 items")
return v
@model_validator(mode="after")
def _cost_range_ordered(self) -> EstimationReport:
"""Ensure cost_range_usd_max >= cost_range_usd_min."""
if self.cost_range_usd_max < self.cost_range_usd_min:
raise ValueError("cost_range_usd_max must be >= cost_range_usd_min")
return self
class EstimationSkipped(BaseModel):
"""Record indicating that estimation was skipped for a plan.
Created when the user passes ``--no-estimate`` or when the
estimation actor fails.
"""
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
reason: str = Field(
...,
min_length=1,
description="Why estimation was skipped",
)
timestamp: datetime = Field(
default_factory=datetime.now,
description="When estimation was skipped",
)
actor_name: str | None = Field(
default=None,
description="Estimation actor name (if applicable)",
)
+13 -1
View File
@@ -65,7 +65,11 @@ 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.estimation import (
EstimationReport,
EstimationResult,
EstimationSkipped,
)
from cleveragents.domain.models.core.multi_project import (
MultiProjectMetadata,
ProjectScope,
@@ -639,6 +643,14 @@ class Plan(BaseModel):
default=None,
description="Cost/risk estimation result from the estimation actor",
)
estimation_report: EstimationReport | None = Field(
default=None,
description="Structured estimation report from the estimation actor",
)
estimation_skipped: EstimationSkipped | None = Field(
default=None,
description="Record indicating estimation was skipped or failed",
)
invariant_actor: str | None = Field(
default=None,
description="Optional actor for invariant reconciliation",
@@ -669,6 +669,10 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
# Inputs schema JSON (optional; copied from action)
inputs_schema_json = Column(Text, nullable=True)
# Estimation report and skipped records (JSON-serialized domain models)
estimation_report_json = Column(Text, nullable=True)
estimation_skipped_json = Column(Text, nullable=True)
# Execution environment override (ADR-043)
execution_environment = Column(String(255), nullable=True)
execution_env_priority = Column(String(20), nullable=True)
@@ -959,6 +963,26 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
json.loads(raw_err_details) if raw_err_details else None
)
# Deserialize estimation_report
from cleveragents.domain.models.core.estimation import (
EstimationReport,
EstimationSkipped,
)
estimation_report_obj: EstimationReport | None = None
raw_est_report = cast("str | None", self.estimation_report_json)
if raw_est_report:
estimation_report_obj = EstimationReport.model_validate(
json.loads(raw_est_report)
)
estimation_skipped_obj: EstimationSkipped | None = None
raw_est_skipped = cast("str | None", self.estimation_skipped_json)
if raw_est_skipped:
estimation_skipped_obj = EstimationSkipped.model_validate(
json.loads(raw_est_skipped)
)
return Plan(
identity=PlanIdentity(
plan_id=cast(str, self.plan_id),
@@ -980,6 +1004,8 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
apply_actor=cast("str | None", self.apply_actor),
estimation_actor=cast("str | None", self.estimation_actor),
invariant_actor=cast("str | None", self.invariant_actor),
estimation_report=estimation_report_obj,
estimation_skipped=estimation_skipped_obj,
project_links=project_links_list,
invariants=invariants_list,
arguments=arguments_dict,
@@ -1077,6 +1103,16 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
apply_actor=getattr(plan, "apply_actor", None),
estimation_actor=getattr(plan, "estimation_actor", None),
invariant_actor=getattr(plan, "invariant_actor", None),
estimation_report_json=(
plan.estimation_report.model_dump_json()
if getattr(plan, "estimation_report", None) is not None
else None
),
estimation_skipped_json=(
plan.estimation_skipped.model_dump_json()
if getattr(plan, "estimation_skipped", None) is not None
else None
),
automation_profile=automation_profile_json,
effective_profile_snapshot=plan.effective_profile_snapshot,
reusable=plan.reusable,
@@ -1371,6 +1371,18 @@ class LifecyclePlanRepository:
row.estimation_actor = getattr(plan, "estimation_actor", None) # type: ignore[assignment]
row.invariant_actor = getattr(plan, "invariant_actor", None) # type: ignore[assignment]
# Serialize estimation report and skipped
row.estimation_report_json = ( # type: ignore[assignment]
plan.estimation_report.model_dump_json()
if getattr(plan, "estimation_report", None) is not None
else None
)
row.estimation_skipped_json = ( # type: ignore[assignment]
plan.estimation_skipped.model_dump_json()
if getattr(plan, "estimation_skipped", None) is not None
else None
)
# Serialize automation profile — column is NOT NULL, default 'balanced'
row.automation_profile = LifecyclePlanModel._serialize_automation_profile( # type: ignore[assignment]
plan.automation_profile,