Compare commits

...

1 Commits

Author SHA1 Message Date
aditya 218c1c248b feat(estimation): implement EstimationReport domain model and estimation_produced decision type
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 18s
CI / helm (pull_request) Successful in 34s
CI / typecheck (pull_request) Successful in 52s
CI / security (pull_request) Successful in 55s
CI / lint (pull_request) Successful in 3m16s
CI / quality (pull_request) Successful in 3m44s
CI / unit_tests (pull_request) Failing after 6m14s
CI / docker (pull_request) Has been skipped
CI / coverage (pull_request) Successful in 12m49s
CI / e2e_tests (pull_request) Failing after 16m8s
CI / integration_tests (pull_request) Successful in 24m13s
CI / status-check (pull_request) Failing after 1s
CI / benchmark-regression (pull_request) Successful in 55m1s
Add the spec-aligned EstimationReport Pydantic domain model with
multi-dimensional estimation output: cost range (min/max USD),
expected steps and child plans, rollback risk (0.0-1.0), estimated
duration in minutes, confidence score (0.0-1.0), rationale, and
optional historical basis (plan IDs).

Add estimation_produced to the DecisionType enum and STRATEGIZE_TYPES
frozenset for recording estimation results in the decision tree.

Add Alembic migration m6_006 that adds the estimation_report TEXT
column to v3_plans and updates the ck_decisions_type CHECK constraint
to include the new decision type.

Update Plan domain model and database ORM layer (models.py and
repositories.py) to serialize/deserialize EstimationReport via JSON.

Update existing consolidated_decision.feature and decision model
Robot helper to expect 12 decision types (was 11).

Add comprehensive BDD tests (20 scenarios in estimation_report.feature)
and Robot Framework integration tests (6 test cases) covering creation,
serialization round-trip, validation constraints, immutability, and
DecisionType membership.

ISSUES CLOSED: #649
2026-04-02 10:22:49 +00:00
14 changed files with 927 additions and 13 deletions
+10
View File
@@ -2,6 +2,16 @@
## Unreleased
- Added `EstimationReport` Pydantic domain model with spec-aligned
multi-dimensional estimation 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 (plan IDs). Added `estimation_produced`
to `DecisionType` enum and `STRATEGIZE_TYPES`. Alembic migration
`m6_006` adds `estimation_report` TEXT column to `v3_plans` and
updates the `ck_decisions_type` CHECK constraint. Plan domain model
and ORM layer updated for JSON serialization/deserialization.
Includes 20 BDD scenarios and 6 Robot integration tests. (#649)
- 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,86 @@
"""Add estimation_report JSON column and estimation_produced decision type.
Adds the ``estimation_report`` TEXT column to the ``v3_plans`` table for
persisting structured ``EstimationReport`` domain objects alongside the
existing scalar ``cost_estimate_usd``.
Updates the ``ck_decisions_type`` CHECK constraint to include the new
``estimation_produced`` decision type used to record estimation results
in the decision tree.
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
import sqlalchemy as sa
from alembic import op
# 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
# Decision type enum values (original 11 + new estimation_produced)
_OLD_DECISION_TYPES = (
"'prompt_definition', 'invariant_enforced', "
"'strategy_choice', 'implementation_choice', "
"'resource_selection', 'subplan_spawn', "
"'subplan_parallel_spawn', 'tool_invocation', "
"'error_recovery', 'validation_response', "
"'user_intervention'"
)
_NEW_DECISION_TYPES = (
"'prompt_definition', 'invariant_enforced', "
"'strategy_choice', 'implementation_choice', "
"'resource_selection', 'subplan_spawn', "
"'subplan_parallel_spawn', 'tool_invocation', "
"'error_recovery', 'validation_response', "
"'user_intervention', 'estimation_produced'"
)
def upgrade() -> None:
"""Add estimation_report column and update decision type constraint."""
# 1. Add estimation_report JSON column to v3_plans
op.add_column(
"v3_plans",
sa.Column("estimation_report", sa.Text(), nullable=True),
)
# 2. Update the decision type CHECK constraint to include estimation_produced
# SQLite does not support ALTER CONSTRAINT, so we use batch mode.
with op.batch_alter_table("decisions", schema=None) as batch_op:
batch_op.drop_constraint("ck_decisions_type", type_="check")
batch_op.create_check_constraint(
"ck_decisions_type",
f"decision_type IN ({_NEW_DECISION_TYPES})",
)
def downgrade() -> None:
"""Remove estimation_report column and revert decision type constraint.
.. warning::
The downgrade will fail if any ``decisions`` rows contain
``estimation_produced`` as their ``decision_type``. Delete
those rows before downgrading.
"""
# 1. Revert the decision type CHECK constraint
with op.batch_alter_table("decisions", schema=None) as batch_op:
batch_op.drop_constraint("ck_decisions_type", type_="check")
batch_op.create_check_constraint(
"ck_decisions_type",
f"decision_type IN ({_OLD_DECISION_TYPES})",
)
# 2. Drop the estimation_report column
op.drop_column("v3_plans", "estimation_report")
+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 |
# ============================================================
+109
View File
@@ -0,0 +1,109 @@
Feature: EstimationReport domain model
As a developer
I want a structured EstimationReport domain model
So that estimation actors can produce spec-aligned multi-dimensional output
with cost ranges, rollback risk, confidence, and historical basis
# Basic creation and field validation
Scenario: EstimationReport can be created with all required fields
Given I create an estimation report with valid fields
Then the estimation report should have the expected cost range
And the estimation report should have the expected steps and child plans
And the estimation report should have the expected rollback risk and confidence
Scenario: EstimationReport created with historical_basis
Given I create an estimation report with historical basis
Then the estimation report historical_basis should have 2 entries
Scenario: EstimationReport created without historical_basis defaults to empty tuple
Given I create an estimation report with valid fields
Then the estimation report historical_basis should be empty
# Range and probability constraints
Scenario: EstimationReport rejects cost_range_usd_max less than cost_range_usd_min
When I try to create an estimation report with max cost less than min cost
Then a validation error should be raised for cost range ordering
Scenario: EstimationReport rejects negative cost_range_usd_min
When I try to create an estimation report with negative min cost
Then a validation error should be raised for cost range
Scenario: EstimationReport rejects rollback_risk above 1.0
When I try to create an estimation report with rollback_risk of 1.5
Then a validation error should be raised for rollback risk
Scenario: EstimationReport rejects rollback_risk below 0.0
When I try to create an estimation report with rollback_risk of -0.1
Then a validation error should be raised for rollback risk
Scenario: EstimationReport rejects confidence above 1.0
When I try to create an estimation report with confidence of 1.5
Then a validation error should be raised for confidence
Scenario: EstimationReport rejects confidence below 0.0
When I try to create an estimation report with confidence of -0.1
Then a validation error should be raised for confidence
Scenario: EstimationReport accepts boundary value 0.0 for rollback_risk
Given I create an estimation report with rollback_risk of 0.0
Then the estimation report rollback_risk should be 0.0
Scenario: EstimationReport accepts boundary value 1.0 for rollback_risk
Given I create an estimation report with rollback_risk of 1.0
Then the estimation report rollback_risk should be 1.0
Scenario: EstimationReport accepts boundary value 0.0 for confidence
Given I create an estimation report with confidence of 0.0
Then the estimation report confidence should be 0.0
Scenario: EstimationReport accepts boundary value 1.0 for confidence
Given I create an estimation report with confidence of 1.0
Then the estimation report confidence should be 1.0
# Immutability
Scenario: EstimationReport is frozen (immutable)
Given I create an estimation report with valid fields
When I try to modify the estimation report rationale
Then a validation error should be raised for frozen model
# Serialization round-trip
Scenario: EstimationReport model_dump and model_validate round-trip
Given I create an estimation report with all optional fields
When I serialize the estimation report via model_dump
And I deserialize it back via model_validate
Then the deserialized estimation report should match the original
Scenario: EstimationReport JSON serialization round-trip
Given I create an estimation report with all optional fields
When I serialize the estimation report to JSON
And I deserialize it back from JSON
Then the deserialized estimation report should match the original
# historical_basis constraints
Scenario: EstimationReport rejects more than 100 historical_basis entries
When I try to create an estimation report with 101 historical basis entries
Then a validation error should be raised for historical basis limit
# as_display_dict
Scenario: EstimationReport as_display_dict includes all fields
Given I create an estimation report with all optional fields
When I call as_display_dict on the estimation report
Then the display dict should include cost_range_usd with min and max
And the display dict should include expected_steps
And the display dict should include rollback_risk
And the display dict should include confidence
And the display dict should include rationale
And the display dict should include historical_basis
# DecisionType includes estimation_produced
Scenario: DecisionType enum includes estimation_produced
Then the DecisionType enum should have an ESTIMATION_PRODUCED value
And the ESTIMATION_PRODUCED value should be "estimation_produced"
And estimation_produced should be in STRATEGIZE_TYPES
+288
View File
@@ -0,0 +1,288 @@
"""Step definitions for EstimationReport domain model scenarios (issue #649)."""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.decision import (
STRATEGIZE_TYPES,
DecisionType,
)
from cleveragents.domain.models.core.estimation import EstimationReport
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_VALID_REPORT_KWARGS: dict[str, object] = {
"cost_range_usd_min": 0.50,
"cost_range_usd_max": 2.50,
"expected_steps": 10,
"expected_child_plans": 3,
"rollback_risk": 0.15,
"estimated_duration_minutes": 45.0,
"confidence": 0.85,
"rationale": "Based on similar refactoring tasks in the past.",
}
def _make_report(**overrides: object) -> EstimationReport:
kwargs = {**_VALID_REPORT_KWARGS, **overrides}
return EstimationReport(**kwargs) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("I create an estimation report with valid fields")
def step_create_valid_report(context: Context) -> None:
context.estimation_report = _make_report()
@given("I create an estimation report with historical basis")
def step_create_report_with_history(context: Context) -> None:
context.estimation_report = _make_report(
historical_basis=("01PLAN_A", "01PLAN_B"),
)
@given("I create an estimation report with all optional fields")
def step_create_report_with_all_optional(context: Context) -> None:
context.estimation_report = _make_report(
historical_basis=("01PLAN_X",),
)
@given("I create an estimation report with rollback_risk of {value}")
def step_create_report_rollback_risk(context: Context, value: str) -> None:
context.estimation_report = _make_report(rollback_risk=float(value))
@given("I create an estimation report with confidence of {value}")
def step_create_report_confidence(context: Context, value: str) -> None:
context.estimation_report = _make_report(confidence=float(value))
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("I try to create an estimation report with max cost less than min cost")
def step_try_max_lt_min(context: Context) -> None:
try:
_make_report(cost_range_usd_min=5.0, cost_range_usd_max=1.0)
context.validation_error = None
except ValidationError as exc:
context.validation_error = exc
@when("I try to create an estimation report with negative min cost")
def step_try_negative_min_cost(context: Context) -> None:
try:
_make_report(cost_range_usd_min=-1.0)
context.validation_error = None
except ValidationError as exc:
context.validation_error = exc
@when("I try to create an estimation report with rollback_risk of {value}")
def step_try_invalid_rollback_risk(context: Context, value: str) -> None:
try:
_make_report(rollback_risk=float(value))
context.validation_error = None
except ValidationError as exc:
context.validation_error = exc
@when("I try to create an estimation report with confidence of {value}")
def step_try_invalid_confidence(context: Context, value: str) -> None:
try:
_make_report(confidence=float(value))
context.validation_error = None
except ValidationError as exc:
context.validation_error = exc
@when("I try to modify the estimation report rationale")
def step_try_modify_rationale(context: Context) -> None:
try:
context.estimation_report.rationale = "tampered" # type: ignore[misc]
context.error = None
except ValidationError as exc:
context.error = exc
@when("I try to create an estimation report with 101 historical basis entries")
def step_try_101_historical_basis(context: Context) -> None:
entries = tuple(f"PLAN_{i:04d}" for i in range(101))
try:
_make_report(historical_basis=entries)
context.validation_error = None
except ValidationError as exc:
context.validation_error = exc
@when("I serialize the estimation report via model_dump")
def step_serialize_model_dump(context: Context) -> None:
context.report_dict = context.estimation_report.model_dump()
@when("I deserialize it back via model_validate")
def step_deserialize_model_validate(context: Context) -> None:
context.deserialized_report = EstimationReport.model_validate(context.report_dict)
@when("I serialize the estimation report to JSON")
def step_serialize_json(context: Context) -> None:
context.report_json = context.estimation_report.model_dump_json()
@when("I deserialize it back from JSON")
def step_deserialize_json(context: Context) -> None:
context.deserialized_report = EstimationReport.model_validate_json(
context.report_json,
)
@when("I call as_display_dict on the estimation report")
def step_call_display_dict(context: Context) -> None:
context.display_dict = context.estimation_report.as_display_dict()
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the estimation report should have the expected cost range")
def step_check_cost_range(context: Context) -> None:
rpt = context.estimation_report
assert rpt.cost_range_usd_min == 0.50, f"min={rpt.cost_range_usd_min}"
assert rpt.cost_range_usd_max == 2.50, f"max={rpt.cost_range_usd_max}"
@then("the estimation report should have the expected steps and child plans")
def step_check_steps(context: Context) -> None:
rpt = context.estimation_report
assert rpt.expected_steps == 10, f"steps={rpt.expected_steps}"
assert rpt.expected_child_plans == 3, f"plans={rpt.expected_child_plans}"
@then("the estimation report should have the expected rollback risk and confidence")
def step_check_risk_confidence(context: Context) -> None:
rpt = context.estimation_report
assert rpt.rollback_risk == 0.15, f"risk={rpt.rollback_risk}"
assert rpt.confidence == 0.85, f"conf={rpt.confidence}"
@then("the estimation report historical_basis should have {count:d} entries")
def step_check_historical_count(context: Context, count: int) -> None:
assert len(context.estimation_report.historical_basis) == count
@then("the estimation report historical_basis should be empty")
def step_check_historical_empty(context: Context) -> None:
assert context.estimation_report.historical_basis == ()
@then("a validation error should be raised for cost range ordering")
def step_assert_cost_range_ordering_error(context: Context) -> None:
assert context.validation_error is not None, "Expected a ValidationError"
assert "cost_range_usd_max" in str(context.validation_error)
@then("a validation error should be raised for cost range")
def step_assert_cost_range_error(context: Context) -> None:
assert context.validation_error is not None, "Expected a ValidationError"
@then("a validation error should be raised for rollback risk")
def step_assert_rollback_risk_error(context: Context) -> None:
assert context.validation_error is not None, "Expected a ValidationError"
@then("a validation error should be raised for confidence")
def step_assert_confidence_error(context: Context) -> None:
assert context.validation_error is not None, "Expected a ValidationError"
@then("the estimation report rollback_risk should be {value}")
def step_check_rollback_value(context: Context, value: str) -> None:
assert context.estimation_report.rollback_risk == float(value)
@then("the estimation report confidence should be {value}")
def step_check_confidence_value(context: Context, value: str) -> None:
assert context.estimation_report.confidence == float(value)
# Note: "a validation error should be raised for frozen model" is defined in
# estimation_actor_steps.py and is reused here. The When step above sets
# context.error to be compatible with that shared step.
@then("the deserialized estimation report should match the original")
def step_check_roundtrip_match(context: Context) -> None:
assert context.deserialized_report == context.estimation_report
@then("a validation error should be raised for historical basis limit")
def step_assert_historical_limit_error(context: Context) -> None:
assert context.validation_error is not None, "Expected a ValidationError"
assert "historical_basis" in str(context.validation_error)
@then("the display dict should include cost_range_usd with min and max")
def step_check_display_cost(context: Context) -> None:
d = context.display_dict
assert "cost_range_usd" in d
assert "min" in d["cost_range_usd"]
assert "max" in d["cost_range_usd"]
@then("the display dict should include expected_steps")
def step_check_display_steps(context: Context) -> None:
assert "expected_steps" in context.display_dict
@then("the display dict should include rollback_risk")
def step_check_display_rollback(context: Context) -> None:
assert "rollback_risk" in context.display_dict
@then("the display dict should include confidence")
def step_check_display_confidence(context: Context) -> None:
assert "confidence" in context.display_dict
@then("the display dict should include rationale")
def step_check_display_rationale(context: Context) -> None:
assert "rationale" in context.display_dict
@then("the display dict should include historical_basis")
def step_check_display_historical(context: Context) -> None:
assert "historical_basis" in context.display_dict
# --- DecisionType checks ---
@then("the DecisionType enum should have an ESTIMATION_PRODUCED value")
def step_check_decision_type_exists(context: Context) -> None:
assert hasattr(DecisionType, "ESTIMATION_PRODUCED")
@then('the ESTIMATION_PRODUCED value should be "estimation_produced"')
def step_check_decision_type_value(context: Context) -> None:
assert DecisionType.ESTIMATION_PRODUCED.value == "estimation_produced"
@then("estimation_produced should be in STRATEGIZE_TYPES")
def step_check_in_strategize_types(context: Context) -> None:
assert DecisionType.ESTIMATION_PRODUCED in STRATEGIZE_TYPES
+55
View File
@@ -0,0 +1,55 @@
*** Settings ***
Documentation Integration tests for the EstimationReport domain model.
... Covers EstimationReport creation, serialization round-trip,
... validation constraints, DecisionType.ESTIMATION_PRODUCED,
... and Plan model integration.
... Issue #649.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_estimation_report.py
*** Test Cases ***
EstimationReport Domain Model
[Documentation] Verify EstimationReport creation, field values, and defaults
[Tags] estimation model critical
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-report-model cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} estimation-report-model-ok
EstimationReport Serialization Round-Trip
[Documentation] Verify model_dump/model_validate and JSON round-trip
[Tags] estimation serialization critical
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-report-roundtrip cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} estimation-report-roundtrip-ok
EstimationReport Validation Constraints
[Documentation] Verify range constraints for rollback_risk, confidence, and cost ordering
[Tags] estimation validation critical
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-report-validation cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} estimation-report-validation-ok
EstimationReport Immutability
[Documentation] Verify EstimationReport is frozen (immutable)
[Tags] estimation model
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} estimation-report-frozen cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} estimation-report-frozen-ok
DecisionType Estimation Produced
[Documentation] Verify DecisionType.ESTIMATION_PRODUCED exists and is in STRATEGIZE_TYPES
[Tags] estimation decision critical
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} decision-type-estimation cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-type-estimation-ok
Plan Model Estimation Report Field
[Documentation] Verify Plan model accepts estimation_report field
[Tags] estimation plan model
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} plan-estimation-report-field cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-estimation-report-field-ok
+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")
+197
View File
@@ -0,0 +1,197 @@
"""Helper utilities for EstimationReport Robot integration tests.
Covers:
- EstimationReport domain model creation and field validation
- Serialization / deserialization round-trip
- Validation constraints (rollback_risk, confidence, cost ordering)
- Immutability (frozen model)
- DecisionType.ESTIMATION_PRODUCED enum and STRATEGIZE_TYPES membership
- Plan model estimation_report field
Issue #649.
"""
from __future__ import annotations
import sys
from pydantic import ValidationError
from cleveragents.domain.models.core.decision import (
STRATEGIZE_TYPES,
DecisionType,
)
from cleveragents.domain.models.core.estimation import EstimationReport
def _make_report(**overrides: object) -> EstimationReport:
"""Create a valid EstimationReport with sensible defaults."""
defaults: dict[str, object] = {
"cost_range_usd_min": 0.50,
"cost_range_usd_max": 2.50,
"expected_steps": 10,
"expected_child_plans": 3,
"rollback_risk": 0.15,
"estimated_duration_minutes": 45.0,
"confidence": 0.85,
"rationale": "Based on similar refactoring tasks.",
}
defaults.update(overrides)
return EstimationReport(**defaults) # type: ignore[arg-type]
def _estimation_report_model() -> None:
"""Test EstimationReport creation and field values."""
report = _make_report(historical_basis=("PLAN_A", "PLAN_B"))
assert report.cost_range_usd_min == 0.50
assert report.cost_range_usd_max == 2.50
assert report.expected_steps == 10
assert report.expected_child_plans == 3
assert report.rollback_risk == 0.15
assert report.estimated_duration_minutes == 45.0
assert report.confidence == 0.85
assert "refactoring" in report.rationale
assert len(report.historical_basis) == 2
# Defaults
report2 = _make_report()
assert report2.historical_basis == ()
print("estimation-report-model-ok")
def _estimation_report_roundtrip() -> None:
"""Test model_dump/model_validate and JSON round-trip."""
report = _make_report(historical_basis=("PLAN_X",))
# model_dump -> model_validate
d = report.model_dump()
restored = EstimationReport.model_validate(d)
assert restored == report, "model_dump round-trip failed"
# JSON round-trip
json_str = report.model_dump_json()
restored_json = EstimationReport.model_validate_json(json_str)
assert restored_json == report, "JSON round-trip failed"
print("estimation-report-roundtrip-ok")
def _estimation_report_validation() -> None:
"""Test validation constraints."""
# cost max < min
try:
_make_report(cost_range_usd_min=5.0, cost_range_usd_max=1.0)
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
# negative min cost
try:
_make_report(cost_range_usd_min=-1.0)
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
# rollback_risk > 1.0
try:
_make_report(rollback_risk=1.5)
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
# rollback_risk < 0.0
try:
_make_report(rollback_risk=-0.1)
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
# confidence > 1.0
try:
_make_report(confidence=1.5)
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
# confidence < 0.0
try:
_make_report(confidence=-0.1)
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
# Boundary values should work
report_0 = _make_report(rollback_risk=0.0, confidence=0.0)
assert report_0.rollback_risk == 0.0
assert report_0.confidence == 0.0
report_1 = _make_report(rollback_risk=1.0, confidence=1.0)
assert report_1.rollback_risk == 1.0
assert report_1.confidence == 1.0
# historical_basis > 100
try:
_make_report(historical_basis=tuple(f"PLAN_{i:04d}" for i in range(101)))
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
print("estimation-report-validation-ok")
def _estimation_report_frozen() -> None:
"""Test that EstimationReport is immutable."""
report = _make_report()
try:
report.rationale = "tampered" # type: ignore[misc]
raise AssertionError("Should have raised ValidationError")
except ValidationError:
pass
print("estimation-report-frozen-ok")
def _decision_type_estimation() -> None:
"""Test DecisionType.ESTIMATION_PRODUCED and STRATEGIZE_TYPES."""
assert hasattr(DecisionType, "ESTIMATION_PRODUCED")
assert DecisionType.ESTIMATION_PRODUCED.value == "estimation_produced"
assert DecisionType.ESTIMATION_PRODUCED in STRATEGIZE_TYPES
# Verify total count is 12 (11 original + estimation_produced)
assert len(DecisionType) == 12, f"Expected 12 members, got {len(DecisionType)}"
print("decision-type-estimation-ok")
def _plan_estimation_report_field() -> None:
"""Test Plan model accepts estimation_report field."""
from cleveragents.domain.models.core.plan import Plan
# Verify the field exists on Plan
assert "estimation_report" in Plan.model_fields, (
"Plan should have estimation_report field"
)
# Verify default is None
field_info = Plan.model_fields["estimation_report"]
assert field_info.default is None
print("plan-estimation-report-field-ok")
_COMMANDS = {
"estimation-report-model": _estimation_report_model,
"estimation-report-roundtrip": _estimation_report_roundtrip,
"estimation-report-validation": _estimation_report_validation,
"estimation-report-frozen": _estimation_report_frozen,
"decision-type-estimation": _decision_type_estimation,
"plan-estimation-report-field": _plan_estimation_report_field,
}
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]]()
@@ -139,7 +139,10 @@ 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,
)
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
@@ -397,6 +400,7 @@ __all__ = [
"ErrorRecord",
"ErrorRecoveryPolicy",
"EscalationDecision",
"EstimationReport",
"EstimationResult",
"ExecutionEnvironment",
"ExecutionEnvironmentPreference",
@@ -48,6 +48,9 @@ Decision types
* - ``user_intervention``
- Any
- User-provided guidance / correction
* - ``estimation_produced``
- Strategize
- Estimation actor produced a cost/risk report
Context snapshots
-----------------
@@ -105,6 +108,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 +136,7 @@ STRATEGIZE_TYPES: frozenset[DecisionType] = frozenset(
DecisionType.SUBPLAN_SPAWN,
DecisionType.SUBPLAN_PARALLEL_SPAWN,
DecisionType.USER_INTERVENTION,
DecisionType.ESTIMATION_PRODUCED,
}
)
@@ -1,15 +1,20 @@
"""Estimation result domain model for CleverAgents.
"""Estimation domain models 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.
``EstimationReport`` is the spec-aligned, multi-dimensional estimation
payload (issue #649) that captures cost ranges, rollback risk, confidence,
rationale, and historical basis. It supplements ``EstimationResult``
which remains in use by the existing estimation stub actor.
The result is informational only; it does **not** gate the
Execute phase.
Based on ``docs/specification.md`` estimation actor sections and
issue #890.
Based on ``docs/specification.md`` estimation actor sections,
issue #890, and issue #649.
"""
from __future__ import annotations
@@ -18,7 +23,7 @@ from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
__all__ = ["EstimationResult"]
__all__ = ["EstimationReport", "EstimationResult"]
class EstimationResult(BaseModel):
@@ -115,3 +120,124 @@ class EstimationResult(BaseModel):
if self.summary:
d["summary"] = self.summary
return d
class EstimationReport(BaseModel):
"""Spec-aligned, multi-dimensional estimation report (issue #649).
Captures the full structured output of the estimation actor as
described in ``docs/specification.md`` lines 19068-19085:
* LLM tokens/cost range (min/max USD)
* Expected steps and child plans
* Rollback risk (0.0-1.0)
* Estimated execution time
* Confidence score (0.0-1.0)
* Rationale and optional historical basis
Unlike ``EstimationResult`` (which uses optional scalar fields),
this model requires most fields and uses range-based costs for
production-grade estimation output.
"""
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
# Cost range
cost_range_usd_min: float = Field(
...,
ge=0.0,
description="Minimum estimated LLM cost in USD",
)
cost_range_usd_max: float = Field(
...,
ge=0.0,
description="Maximum estimated LLM cost in USD",
)
# Effort estimates
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",
)
# Risk
rollback_risk: float = Field(
...,
ge=0.0,
le=1.0,
description="Probability of needing a rollback (0.0-1.0)",
)
# Duration
estimated_duration_minutes: float = Field(
...,
ge=0.0,
description="Estimated execution time in minutes",
)
# Confidence
confidence: float = Field(
...,
ge=0.0,
le=1.0,
description="Confidence in the estimation (0.0-1.0)",
)
# Rationale
rationale: str = Field(
...,
min_length=1,
max_length=10_000,
description="Human-readable rationale for the estimation",
)
# Historical basis (optional)
historical_basis: tuple[str, ...] = Field(
default_factory=tuple,
description="Plan IDs used as historical reference for the estimate",
)
# --- Validators ---
@field_validator("cost_range_usd_max")
@classmethod
def _cost_max_gte_min(cls, v: float, info: Any) -> float:
"""Ensure max cost is not less than min cost."""
min_val = info.data.get("cost_range_usd_min")
if min_val is not None and v < min_val:
raise ValueError(
f"cost_range_usd_max ({v}) must be >= cost_range_usd_min ({min_val})"
)
return v
@field_validator("historical_basis")
@classmethod
def _cap_historical_basis(cls, v: tuple[str, ...]) -> tuple[str, ...]:
"""Limit historical basis entries to a reasonable count."""
if len(v) > 100:
raise ValueError("historical_basis must contain at most 100 items")
return v
def as_display_dict(self) -> dict[str, Any]:
"""Return a dict of fields suitable for display/CLI output."""
d: dict[str, Any] = {
"cost_range_usd": {
"min": self.cost_range_usd_min,
"max": self.cost_range_usd_max,
},
"expected_steps": self.expected_steps,
"expected_child_plans": self.expected_child_plans,
"rollback_risk": self.rollback_risk,
"estimated_duration_minutes": self.estimated_duration_minutes,
"confidence": self.confidence,
"rationale": self.rationale,
}
if self.historical_basis:
d["historical_basis"] = list(self.historical_basis)
return d
+8 -1
View File
@@ -65,7 +65,10 @@ 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,
)
from cleveragents.domain.models.core.multi_project import (
MultiProjectMetadata,
ProjectScope,
@@ -639,6 +642,10 @@ class Plan(BaseModel):
default=None,
description="Cost/risk estimation result from the estimation actor",
)
estimation_report: EstimationReport | None = Field(
default=None,
description="Structured multi-dimensional estimation report (issue #649)",
)
invariant_actor: str | None = Field(
default=None,
description="Optional actor for invariant reconciliation",
@@ -683,6 +683,9 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
error_message = Column(Text, nullable=True)
error_details_json = Column(Text, nullable=True)
# Estimation report (structured JSON -- issue #649)
estimation_report = Column(Text, nullable=True)
# Cost/token tracking
cost_estimate_usd = Column(Float, nullable=True)
cost_actual_usd = Column(Float, nullable=True)
@@ -959,6 +962,16 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
json.loads(raw_err_details) if raw_err_details else None
)
# Deserialize estimation report (issue #649)
from cleveragents.domain.models.core.estimation import EstimationReport
estimation_report_obj: EstimationReport | None = None
raw_estimation_report = cast("str | None", self.estimation_report)
if raw_estimation_report:
estimation_report_obj = EstimationReport.model_validate_json(
raw_estimation_report
)
return Plan(
identity=PlanIdentity(
plan_id=cast(str, self.plan_id),
@@ -979,6 +992,7 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
review_actor=cast("str | None", self.review_actor),
apply_actor=cast("str | None", self.apply_actor),
estimation_actor=cast("str | None", self.estimation_actor),
estimation_report=estimation_report_obj,
invariant_actor=cast("str | None", self.invariant_actor),
project_links=project_links_list,
invariants=invariants_list,
@@ -1076,6 +1090,11 @@ class LifecyclePlanModel(Base): # type: ignore[misc]
review_actor=getattr(plan, "review_actor", None),
apply_actor=getattr(plan, "apply_actor", None),
estimation_actor=getattr(plan, "estimation_actor", None),
estimation_report=(
plan.estimation_report.model_dump_json()
if getattr(plan, "estimation_report", None) is not None
else None
),
invariant_actor=getattr(plan, "invariant_actor", None),
automation_profile=automation_profile_json,
effective_profile_snapshot=plan.effective_profile_snapshot,
@@ -2731,7 +2750,7 @@ class DecisionModel(Base): # type: ignore[misc]
"'resource_selection', 'subplan_spawn', "
"'subplan_parallel_spawn', 'tool_invocation', "
"'error_recovery', 'validation_response', "
"'user_intervention')",
"'user_intervention', 'estimation_produced')",
name="ck_decisions_type",
),
CheckConstraint(
@@ -1369,6 +1369,11 @@ class LifecyclePlanRepository:
row.review_actor = getattr(plan, "review_actor", None) # type: ignore[assignment]
row.apply_actor = getattr(plan, "apply_actor", None) # type: ignore[assignment]
row.estimation_actor = getattr(plan, "estimation_actor", None) # type: ignore[assignment]
row.estimation_report = ( # type: ignore[assignment]
plan.estimation_report.model_dump_json()
if getattr(plan, "estimation_report", None) is not None
else None
)
row.invariant_actor = getattr(plan, "invariant_actor", None) # type: ignore[assignment]
# Serialize automation profile — column is NOT NULL, default 'balanced'