cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(estimation): build historical plan statistics query service for estimation context assembly #1295

Closed
brent.edwards wants to merge 1 commits from feature/m6-estimation-historical-stats into master
pull from: feature/m6-estimation-historical-stats
merge into: cleveragents:master
Conversation 7 Commits 1 Files Changed 9
+1780
9 changed files with 1780 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/historical_plan_stats.feature
+157
View File
@@ -0,0 +1,157 @@
Feature: Historical Plan Statistics
As a developer
I want to query historical plan statistics for an action
So that the estimation actor can make data-driven cost and risk estimates
# HistoricalPlanStats domain model
Scenario: HistoricalPlanStats can be created with all fields
Given I create historical plan stats with all fields populated
Then the historical stats should have all expected values
And the historical stats sample_size should be 5
Scenario: HistoricalPlanStats empty factory returns valid zero stats
Given I create empty historical plan stats for action "local/refactor"
Then the historical stats sample_size should be 0
And the historical stats mean_cost_usd should be 0.0
And the historical stats success_rate should be 0.0
Scenario: HistoricalPlanStats is frozen (immutable)
Given I create empty historical plan stats for action "local/refactor"
When I try to modify the historical stats mean_cost_usd
Then a validation error should be raised for frozen historical stats
Scenario: HistoricalPlanStats as_context_dict includes non-zero metrics
Given I create historical plan stats with all fields populated
When I call as_context_dict on the historical stats
Then the context dict should include action_name
And the context dict should include sample_size
And the context dict should include mean_cost_usd
And the context dict should include success_rate
Scenario: HistoricalPlanStats as_context_dict excludes zero metrics
Given I create empty historical plan stats for action "local/new-action"
When I call as_context_dict on the historical stats
Then the context dict should include action_name
And the context dict should include sample_size
And the context dict should not include mean_cost_usd
Scenario: HistoricalPlanStats rejects negative cost
When I try to create historical stats with negative mean_cost_usd
Then a validation error should be raised for negative cost
Scenario: HistoricalPlanStats rejects success_rate above 1.0
When I try to create historical stats with success_rate 1.5
Then a validation error should be raised for success_rate bounds
# HistoricalPlanStatsService
Scenario: Service returns empty stats when no completed plans exist
Given I have a historical plan stats service with no completed plans
When I query stats for action "local/refactor"
Then the returned stats sample_size should be 0
And the returned stats action_name should be "local/refactor"
Scenario: Service computes stats from completed plans
Given I have a historical plan stats service with 3 completed plans
When I query stats for action "local/refactor"
Then the returned stats sample_size should be 3
And the returned stats mean_cost_usd should be greater than 0
And the returned stats success_rate should be greater than 0
Scenario: Service handles empty action name
Given I have a historical plan stats service with no completed plans
When I query stats for an empty action name
Then the returned stats action_name should be "<unknown>"
And the returned stats sample_size should be 0
Scenario: Service handles limit of zero
Given I have a historical plan stats service with 3 completed plans
When I query stats for action "local/refactor" with limit 0
Then the returned stats sample_size should be 0
Scenario: Service computes correct median for even number of plans
Given I have a historical plan stats service with 4 completed plans with varied costs
When I query stats for action "local/refactor"
Then the returned stats median_cost_usd should be correctly computed
Scenario: Service computes p90 cost correctly
Given I have a historical plan stats service with 10 completed plans
When I query stats for action "local/refactor"
Then the returned stats p90_cost_usd should be greater than median_cost_usd
Scenario: Service handles plans with no cost metadata
Given I have a historical plan stats service with plans missing cost data
When I query stats for action "local/refactor"
Then the returned stats sample_size should be 2
And the returned stats mean_cost_usd should be 0.0
Scenario: Service handles plans with no timestamps
Given I have a historical plan stats service with plans missing timestamps
When I query stats for action "local/refactor"
Then the returned stats sample_size should be 1
And the returned stats mean_duration_seconds should be 0.0
# EstimationHistoricalStatsProvider (ACMS integration)
Scenario: Provider returns 0.0 confidence when no action name is set
Given I have an estimation historical stats provider with no action name
When I check the provider confidence for a request
Then the provider confidence should be 0.0
Scenario: Provider returns 0.8 confidence when action name is set
Given I have an estimation historical stats provider with action "local/refactor"
When I check the provider confidence for a request
Then the provider confidence should be 0.8
Scenario: Provider produces a context fragment with stats
Given I have an estimation historical stats provider with action "local/refactor"
And the provider has 3 completed plans available
When I assemble fragments with the provider
Then the assembled fragments should include a historical stats fragment
Scenario: Provider returns unchanged fragments when no history exists
Given I have an estimation historical stats provider with action "local/new-action"
And the provider has no completed plans available
When I assemble fragments with the provider
Then the assembled fragments should be unchanged
Scenario: Provider explain returns meaningful description
Given I have an estimation historical stats provider with action "local/refactor"
Then the provider explain should mention historical plan statistics
Scenario: Provider name returns the expected strategy name
Given I have an estimation historical stats provider with action "local/refactor"
Then the provider name should be "estimation-historical-stats"
Scenario: Provider capabilities include temporal archaeology
Given I have an estimation historical stats provider with action "local/refactor"
Then the provider capabilities should support temporal archaeology
Scenario: Provider set_action_name updates the action for querying
Given I have an estimation historical stats provider with no action name
When I set the provider action name to "local/updated-action"
Then the provider confidence should be 0.8 for a request
Scenario: Provider handles exception from stats service gracefully
Given I have an estimation historical stats provider with a failing service
When I assemble fragments with the failing provider
Then the assembled fragments from failing provider should be unchanged
Scenario: Provider skips fragment when it exceeds budget
Given I have an estimation historical stats provider with action "local/refactor"
And the provider has 3 completed plans available
When I assemble fragments with the provider using tiny budget
Then the assembled fragments with tiny budget should be unchanged
# Repository method
Scenario: Repository get_completed_plans_by_action returns empty for unknown action
Given I have a plan repository with some plans
When I query completed plans for action "local/nonexistent"
Then the returned plan list should be empty
Scenario: Repository get_completed_plans_by_action returns only terminal plans
Given I have a plan repository with mixed state plans
When I query completed plans for action "local/test-action"
Then the returned plans should only have terminal states
features/steps/historical_plan_stats_steps.py
+626
View File
@@ -0,0 +1,626 @@
"""Step definitions for Historical Plan Statistics feature tests."""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
from unittest.mock import MagicMock
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError as PydanticValidationError
from cleveragents.application.services.estimation_context_provider import (
EstimationHistoricalStatsProvider,
)
from cleveragents.application.services.historical_plan_stats_service import (
HistoricalPlanStatsService,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
)
from cleveragents.domain.models.core.historical_plan_stats import HistoricalPlanStats
# ---------------------------------------------------------------------------
# HistoricalPlanStats domain model steps
# ---------------------------------------------------------------------------
@given("I create historical plan stats with all fields populated")
def step_create_stats_all_fields(context: Context) -> None:
"""Create a HistoricalPlanStats with all fields populated."""
context.historical_stats = HistoricalPlanStats(
action_name="local/refactor",
sample_size=5,
mean_cost_usd=0.125,
median_cost_usd=0.10,
p90_cost_usd=0.25,
mean_duration_seconds=120.5,
median_duration_seconds=100.0,
avg_step_count=8.4,
avg_child_plan_count=2.6,
success_rate=0.8,
)
@given('I create empty historical plan stats for action "{action_name}"')
def step_create_empty_stats(context: Context, action_name: str) -> None:
"""Create empty stats via the factory method."""
context.historical_stats = HistoricalPlanStats.empty(action_name=action_name)
@then("the historical stats should have all expected values")
def step_check_stats_all_values(context: Context) -> None:
"""Verify all fields on the historical stats."""
stats: HistoricalPlanStats = context.historical_stats
assert stats.action_name == "local/refactor"
assert stats.mean_cost_usd == 0.125
assert stats.median_cost_usd == 0.10
assert stats.p90_cost_usd == 0.25
assert stats.mean_duration_seconds == 120.5
assert stats.median_duration_seconds == 100.0
assert stats.avg_step_count == 8.4
assert stats.avg_child_plan_count == 2.6
assert stats.success_rate == 0.8
@then("the historical stats sample_size should be {expected:d}")
def step_check_sample_size(context: Context, expected: int) -> None:
"""Verify sample_size."""
stats: HistoricalPlanStats = context.historical_stats
assert stats.sample_size == expected, (
f"Expected {expected}, got {stats.sample_size}"
)
@then("the historical stats mean_cost_usd should be {expected:f}")
def step_check_mean_cost(context: Context, expected: float) -> None:
"""Verify mean_cost_usd."""
stats: HistoricalPlanStats = context.historical_stats
assert abs(stats.mean_cost_usd - expected) < 1e-6, (
f"Expected {expected}, got {stats.mean_cost_usd}"
)
@then("the historical stats success_rate should be {expected:f}")
def step_check_success_rate(context: Context, expected: float) -> None:
"""Verify success_rate."""
stats: HistoricalPlanStats = context.historical_stats
assert abs(stats.success_rate - expected) < 1e-6, (
f"Expected {expected}, got {stats.success_rate}"
)
@when("I try to modify the historical stats mean_cost_usd")
def step_try_modify_stats(context: Context) -> None:
"""Try to modify a frozen field."""
try:
context.historical_stats.mean_cost_usd = 999.0 # type: ignore[misc]
context.modify_error = None
except (PydanticValidationError, AttributeError) as exc:
context.modify_error = exc
@then("a validation error should be raised for frozen historical stats")
def step_check_frozen_error(context: Context) -> None:
"""Verify that modification raised an error."""
assert context.modify_error is not None, "Expected error but none was raised"
@when("I call as_context_dict on the historical stats")
def step_call_context_dict(context: Context) -> None:
"""Call as_context_dict."""
context.context_dict = context.historical_stats.as_context_dict()
@then("the context dict should include action_name")
def step_context_dict_has_action_name(context: Context) -> None:
"""Verify context dict includes action_name."""
assert "action_name" in context.context_dict
@then("the context dict should include sample_size")
def step_context_dict_has_sample_size(context: Context) -> None:
"""Verify context dict includes sample_size."""
assert "sample_size" in context.context_dict
@then("the context dict should include mean_cost_usd")
def step_context_dict_has_mean_cost(context: Context) -> None:
"""Verify context dict includes mean_cost_usd."""
assert "mean_cost_usd" in context.context_dict
@then("the context dict should include success_rate")
def step_context_dict_has_success_rate(context: Context) -> None:
"""Verify context dict includes success_rate."""
assert "success_rate" in context.context_dict
@then("the context dict should not include mean_cost_usd")
def step_context_dict_excludes_mean_cost(context: Context) -> None:
"""Verify context dict excludes mean_cost_usd (zero value)."""
assert "mean_cost_usd" not in context.context_dict
@when("I try to create historical stats with negative mean_cost_usd")
def step_create_negative_cost(context: Context) -> None:
"""Try to create stats with negative cost."""
try:
HistoricalPlanStats(action_name="test", mean_cost_usd=-1.0)
context.validation_error = None
except PydanticValidationError as exc:
context.validation_error = exc
@then("a validation error should be raised for negative cost")
def step_check_negative_cost_error(context: Context) -> None:
"""Verify validation error for negative cost."""
assert context.validation_error is not None, (
"Expected validation error but none was raised"
)
@when("I try to create historical stats with success_rate {rate}")
def step_create_bad_success_rate(context: Context, rate: str) -> None:
"""Try to create stats with invalid success_rate."""
try:
HistoricalPlanStats(action_name="test", success_rate=float(rate))
context.validation_error = None
except PydanticValidationError as exc:
context.validation_error = exc
@then("a validation error should be raised for success_rate bounds")
def step_check_success_rate_error(context: Context) -> None:
"""Verify validation error for success_rate bounds."""
assert context.validation_error is not None, (
"Expected validation error but none was raised"
)
# ---------------------------------------------------------------------------
# Mock plan factory helpers
# ---------------------------------------------------------------------------
def _make_mock_plan(
*,
action_name: str = "local/refactor",
processing_state: str = "applied",
cost: float = 0.10,
duration_seconds: float = 60.0,
step_count: int = 5,
child_plans: int = 1,
) -> Any:
"""Create a mock plan domain object for testing."""
plan = MagicMock()
plan.action_name = action_name
# Processing state
state_mock = MagicMock()
state_mock.value = processing_state
plan.processing_state = state_mock
# Cost metadata
cost_meta = MagicMock()
cost_meta.total_cost = cost
plan.cost_metadata = cost_meta
# Timestamps
created = datetime.now() - timedelta(seconds=duration_seconds)
timestamps = MagicMock()
timestamps.created_at = created
timestamps.applied_at = datetime.now() if processing_state == "applied" else None
timestamps.execute_completed_at = (
datetime.now() if processing_state == "complete" else None
)
plan.timestamps = timestamps
# Step count
plan.last_completed_step = step_count - 1 # 0-indexed
# Child plans
plan.subplan_statuses = [MagicMock() for _ in range(child_plans)]
return plan
# ---------------------------------------------------------------------------
# HistoricalPlanStatsService steps
# ---------------------------------------------------------------------------
@given("I have a historical plan stats service with no completed plans")
def step_service_no_plans(context: Context) -> None:
"""Set up service with empty repository."""
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=[])
context.stats_service = HistoricalPlanStatsService(repo)
@given("I have a historical plan stats service with plans missing cost data")
def step_service_no_cost_data(context: Context) -> None:
"""Set up service with plans that have no cost metadata."""
plan1 = MagicMock()
plan1.action_name = "local/refactor"
state1 = MagicMock()
state1.value = "applied"
plan1.processing_state = state1
plan1.cost_metadata = None
timestamps1 = MagicMock()
timestamps1.created_at = datetime.now() - timedelta(seconds=60)
timestamps1.applied_at = datetime.now()
timestamps1.execute_completed_at = None
plan1.timestamps = timestamps1
plan1.last_completed_step = 2
plan1.subplan_statuses = []
plan2 = MagicMock()
plan2.action_name = "local/refactor"
state2 = MagicMock()
state2.value = "complete"
plan2.processing_state = state2
plan2.cost_metadata = MagicMock()
plan2.cost_metadata.total_cost = 0.0 # Zero cost
timestamps2 = MagicMock()
timestamps2.created_at = datetime.now() - timedelta(seconds=120)
timestamps2.applied_at = None
timestamps2.execute_completed_at = datetime.now()
plan2.timestamps = timestamps2
plan2.last_completed_step = 4
plan2.subplan_statuses = []
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=[plan1, plan2])
context.stats_service = HistoricalPlanStatsService(repo)
@given("I have a historical plan stats service with plans missing timestamps")
def step_service_no_timestamps(context: Context) -> None:
"""Set up service with a plan missing timestamp data."""
plan = MagicMock()
plan.action_name = "local/refactor"
state = MagicMock()
state.value = "applied"
plan.processing_state = state
plan.cost_metadata = None
plan.timestamps = None # No timestamps at all
plan.last_completed_step = -1
plan.subplan_statuses = []
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=[plan])
context.stats_service = HistoricalPlanStatsService(repo)
@given("I have a historical plan stats service with {count:d} completed plans")
def step_service_with_plans(context: Context, count: int) -> None:
"""Set up service with mock completed plans."""
plans = [
_make_mock_plan(
cost=0.10 * (i + 1),
duration_seconds=60.0 * (i + 1),
step_count=3 + i,
child_plans=i,
processing_state="applied" if i % 2 == 0 else "complete",
)
for i in range(count)
]
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=plans)
context.stats_service = HistoricalPlanStatsService(repo)
@given(
"I have a historical plan stats service with {count:d} completed plans with varied costs"
)
def step_service_with_varied_plans(context: Context, count: int) -> None:
"""Set up service with plans having specific costs for median testing."""
costs = [0.05, 0.10, 0.20, 0.50][:count]
plans = [_make_mock_plan(cost=c, processing_state="applied") for c in costs]
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=plans)
context.stats_service = HistoricalPlanStatsService(repo)
@when('I query stats for action "{action_name}"')
def step_query_stats(context: Context, action_name: str) -> None:
"""Query stats for the given action."""
context.returned_stats = context.stats_service.get_stats_for_action(action_name)
@when("I query stats for an empty action name")
def step_query_stats_empty(context: Context) -> None:
"""Query stats with an empty action name."""
context.returned_stats = context.stats_service.get_stats_for_action("")
@when('I query stats for action "{action_name}" with limit {limit:d}')
def step_query_stats_with_limit(context: Context, action_name: str, limit: int) -> None:
"""Query stats with an explicit limit."""
context.returned_stats = context.stats_service.get_stats_for_action(
action_name, limit=limit
)
@then("the returned stats sample_size should be {expected:d}")
def step_check_returned_sample_size(context: Context, expected: int) -> None:
"""Verify returned stats sample_size."""
assert context.returned_stats.sample_size == expected, (
f"Expected {expected}, got {context.returned_stats.sample_size}"
)
@then('the returned stats action_name should be "{expected}"')
def step_check_returned_action_name(context: Context, expected: str) -> None:
"""Verify returned stats action_name."""
assert context.returned_stats.action_name == expected, (
f"Expected {expected!r}, got {context.returned_stats.action_name!r}"
)
@then("the returned stats mean_cost_usd should be {expected:f}")
def step_check_returned_mean_cost_exact(context: Context, expected: float) -> None:
"""Verify returned stats mean_cost_usd matches exactly."""
assert abs(context.returned_stats.mean_cost_usd - expected) < 1e-6, (
f"Expected {expected}, got {context.returned_stats.mean_cost_usd}"
)
@then("the returned stats mean_duration_seconds should be {expected:f}")
def step_check_returned_mean_duration_exact(context: Context, expected: float) -> None:
"""Verify returned stats mean_duration_seconds."""
assert abs(context.returned_stats.mean_duration_seconds - expected) < 1e-6, (
f"Expected {expected}, got {context.returned_stats.mean_duration_seconds}"
)
@then("the returned stats mean_cost_usd should be greater than {threshold:d}")
def step_check_mean_cost_positive(context: Context, threshold: int) -> None:
"""Verify mean_cost_usd is above threshold."""
assert context.returned_stats.mean_cost_usd > threshold, (
f"Expected mean_cost > {threshold}, got {context.returned_stats.mean_cost_usd}"
)
@then("the returned stats success_rate should be greater than {threshold:d}")
def step_check_success_rate_positive(context: Context, threshold: int) -> None:
"""Verify success_rate is above threshold."""
assert context.returned_stats.success_rate > threshold, (
f"Expected success_rate > {threshold}, got {context.returned_stats.success_rate}"
)
@then("the returned stats median_cost_usd should be correctly computed")
def step_check_median_cost_computed(context: Context) -> None:
"""Verify median is computed (between min and max of costs)."""
stats = context.returned_stats
# For 4 costs [0.05, 0.10, 0.20, 0.50], median = (0.10 + 0.20) / 2 = 0.15
assert stats.median_cost_usd > 0.0, "Median should be > 0"
assert stats.median_cost_usd <= stats.p90_cost_usd, "Median should be <= p90"
@then("the returned stats p90_cost_usd should be greater than median_cost_usd")
def step_check_p90_above_median(context: Context) -> None:
"""Verify p90 >= median."""
stats = context.returned_stats
assert stats.p90_cost_usd >= stats.median_cost_usd, (
f"p90 ({stats.p90_cost_usd}) should be >= median ({stats.median_cost_usd})"
)
# ---------------------------------------------------------------------------
# EstimationHistoricalStatsProvider steps
# ---------------------------------------------------------------------------
@given("I have an estimation historical stats provider with no action name")
def step_provider_no_action(context: Context) -> None:
"""Create provider without action name."""
mock_service = MagicMock(spec=HistoricalPlanStatsService)
context.stats_provider = EstimationHistoricalStatsProvider(mock_service)
@given('I have an estimation historical stats provider with action "{action_name}"')
def step_provider_with_action(context: Context, action_name: str) -> None:
"""Create provider with action name."""
mock_service = MagicMock(spec=HistoricalPlanStatsService)
context.stats_provider = EstimationHistoricalStatsProvider(
mock_service, default_action_name=action_name
)
context.mock_stats_service = mock_service
@given("the provider has {count:d} completed plans available")
def step_provider_has_plans(context: Context, count: int) -> None:
"""Configure mock service to return stats with data."""
mock_stats = HistoricalPlanStats(
action_name="local/refactor",
sample_size=count,
mean_cost_usd=0.15,
median_cost_usd=0.12,
p90_cost_usd=0.30,
mean_duration_seconds=90.0,
median_duration_seconds=80.0,
avg_step_count=6.0,
avg_child_plan_count=1.5,
success_rate=0.75,
)
context.mock_stats_service.get_stats_for_action.return_value = mock_stats
@given("the provider has no completed plans available")
def step_provider_no_plans(context: Context) -> None:
"""Configure mock service to return empty stats."""
empty_stats = HistoricalPlanStats.empty(action_name="local/new-action")
context.mock_stats_service.get_stats_for_action.return_value = empty_stats
@when("I check the provider confidence for a request")
def step_check_provider_confidence(context: Context) -> None:
"""Check can_handle confidence."""
context.provider_confidence = context.stats_provider.can_handle({})
@then("the provider confidence should be {expected:f}")
def step_verify_provider_confidence(context: Context, expected: float) -> None:
"""Verify provider confidence value."""
assert abs(context.provider_confidence - expected) < 1e-6, (
f"Expected {expected}, got {context.provider_confidence}"
)
@when("I assemble fragments with the provider")
def step_assemble_with_provider(context: Context) -> None:
"""Run assemble with the provider."""
budget = ContextBudget(max_tokens=4096, reserved_tokens=0)
context.assembled_fragments = list(context.stats_provider.assemble([], budget))
@then("the assembled fragments should include a historical stats fragment")
def step_check_has_stats_fragment(context: Context) -> None:
"""Verify the assembled fragments contain a historical stats fragment."""
stats_frags = [
f
for f in context.assembled_fragments
if f.metadata.get("type") == "historical_plan_stats"
]
assert len(stats_frags) == 1, (
f"Expected 1 historical stats fragment, got {len(stats_frags)}"
)
@then("the assembled fragments should be unchanged")
def step_check_fragments_unchanged(context: Context) -> None:
"""Verify no additional fragments were added."""
assert len(context.assembled_fragments) == 0, (
f"Expected 0 fragments, got {len(context.assembled_fragments)}"
)
@then("the provider explain should mention historical plan statistics")
def step_check_provider_explain(context: Context) -> None:
"""Verify explain() returns meaningful description."""
explanation = context.stats_provider.explain()
assert "historical" in explanation.lower()
assert "plan" in explanation.lower()
@then('the provider name should be "{expected}"')
def step_check_provider_name(context: Context, expected: str) -> None:
"""Verify the provider name."""
assert context.stats_provider.name == expected, (
f"Expected {expected!r}, got {context.stats_provider.name!r}"
)
@then("the provider capabilities should support temporal archaeology")
def step_check_provider_capabilities(context: Context) -> None:
"""Verify provider capabilities."""
caps = context.stats_provider.capabilities
assert caps.supports_temporal_archaeology is True
@when('I set the provider action name to "{action_name}"')
def step_set_provider_action(context: Context, action_name: str) -> None:
"""Set the action name on the provider."""
context.stats_provider.set_action_name(action_name)
@then("the provider confidence should be {expected:f} for a request")
def step_verify_provider_confidence_for_request(
context: Context, expected: float
) -> None:
"""Verify provider confidence after set_action_name."""
confidence = context.stats_provider.can_handle({})
assert abs(confidence - expected) < 1e-6, f"Expected {expected}, got {confidence}"
@given("I have an estimation historical stats provider with a failing service")
def step_provider_failing_service(context: Context) -> None:
"""Create provider with a service that raises an exception."""
mock_service = MagicMock(spec=HistoricalPlanStatsService)
mock_service.get_stats_for_action.side_effect = RuntimeError("DB error")
context.failing_provider = EstimationHistoricalStatsProvider(
mock_service, default_action_name="local/refactor"
)
@when("I assemble fragments with the failing provider")
def step_assemble_with_failing_provider(context: Context) -> None:
"""Assemble with a provider that has a failing stats service."""
budget = ContextBudget(max_tokens=4096, reserved_tokens=0)
context.failing_fragments = list(context.failing_provider.assemble([], budget))
@then("the assembled fragments from failing provider should be unchanged")
def step_check_failing_fragments_unchanged(context: Context) -> None:
"""Verify no fragments were added when service fails."""
assert len(context.failing_fragments) == 0, (
f"Expected 0 fragments, got {len(context.failing_fragments)}"
)
@when("I assemble fragments with the provider using tiny budget")
def step_assemble_with_tiny_budget(context: Context) -> None:
"""Assemble with a very small budget that can't fit the stats fragment."""
budget = ContextBudget(max_tokens=1, reserved_tokens=0)
context.tiny_budget_fragments = list(context.stats_provider.assemble([], budget))
@then("the assembled fragments with tiny budget should be unchanged")
def step_check_tiny_budget_unchanged(context: Context) -> None:
"""Verify no fragments added with tiny budget."""
assert len(context.tiny_budget_fragments) == 0, (
f"Expected 0 fragments, got {len(context.tiny_budget_fragments)}"
)
# ---------------------------------------------------------------------------
# Repository method steps
# ---------------------------------------------------------------------------
@given("I have a plan repository with some plans")
def step_repo_with_plans(context: Context) -> None:
"""Set up mock repository returning empty for unknown actions."""
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=[])
context.plan_repo = repo
@given("I have a plan repository with mixed state plans")
def step_repo_mixed_states(context: Context) -> None:
"""Set up repository with mixed-state plans (only terminal ones returned)."""
terminal_plans = [
_make_mock_plan(processing_state="applied"),
_make_mock_plan(processing_state="complete"),
_make_mock_plan(processing_state="errored"),
]
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=terminal_plans)
context.plan_repo = repo
@when('I query completed plans for action "{action_name}"')
def step_query_completed_plans(context: Context, action_name: str) -> None:
"""Query the repository for completed plans."""
context.returned_plans = context.plan_repo.get_completed_plans_by_action(
action_name
)
@then("the returned plan list should be empty")
def step_check_plans_empty(context: Context) -> None:
"""Verify empty plan list."""
assert len(context.returned_plans) == 0
@then("the returned plans should only have terminal states")
def step_check_terminal_states(context: Context) -> None:
"""Verify all returned plans have terminal states."""
terminal = {"applied", "complete", "errored", "cancelled"}
for plan in context.returned_plans:
state = plan.processing_state.value
assert state in terminal, f"Plan has non-terminal state: {state}"
robot/helper_historical_plan_stats.py
+281
View File
@@ -0,0 +1,281 @@
"""Robot Framework helper for historical plan statistics integration tests.
Provides a CLI-style interface for Robot to invoke historical plan
statistics operations and verify the results.
Usage:
python robot/helper_historical_plan_stats.py empty-stats
python robot/helper_historical_plan_stats.py populated-stats
python robot/helper_historical_plan_stats.py service-mock
python robot/helper_historical_plan_stats.py service-empty
python robot/helper_historical_plan_stats.py provider-integration
python robot/helper_historical_plan_stats.py context-dict
python robot/helper_historical_plan_stats.py percentile-calc
"""
from __future__ import annotations
import sys
from collections.abc import Callable
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC in sys.path:
sys.path.remove(_SRC)
sys.path.insert(0, _SRC)
from cleveragents.application.services.estimation_context_provider import ( # noqa: E402
EstimationHistoricalStatsProvider,
)
from cleveragents.application.services.historical_plan_stats_service import ( # noqa: E402
HistoricalPlanStatsService,
_percentile,
)
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
ContextBudget,
)
from cleveragents.domain.models.core.historical_plan_stats import ( # noqa: E402
HistoricalPlanStats,
)
def _make_mock_plan(
*,
cost: float = 0.10,
duration_seconds: float = 60.0,
step_count: int = 5,
child_plans: int = 1,
processing_state: str = "applied",
) -> MagicMock:
"""Create a mock plan for testing."""
plan = MagicMock()
plan.action_name = "local/refactor"
state_mock = MagicMock()
state_mock.value = processing_state
plan.processing_state = state_mock
cost_meta = MagicMock()
cost_meta.total_cost = cost
plan.cost_metadata = cost_meta
created = datetime.now() - timedelta(seconds=duration_seconds)
timestamps = MagicMock()
timestamps.created_at = created
timestamps.applied_at = datetime.now() if processing_state == "applied" else None
timestamps.execute_completed_at = (
datetime.now() if processing_state == "complete" else None
)
plan.timestamps = timestamps
plan.last_completed_step = step_count - 1
plan.subplan_statuses = [MagicMock() for _ in range(child_plans)]
return plan
def _cmd_empty_stats() -> int:
"""Test creating empty stats."""
stats = HistoricalPlanStats.empty(action_name="local/test")
if stats.sample_size != 0:
print(f"fail: expected sample_size=0, got {stats.sample_size}")
return 1
if stats.mean_cost_usd != 0.0:
print(f"fail: expected mean_cost_usd=0.0, got {stats.mean_cost_usd}")
return 1
if stats.action_name != "local/test":
print(f"fail: expected action_name='local/test', got {stats.action_name}")
return 1
print("empty-stats-ok")
return 0
def _cmd_populated_stats() -> int:
"""Test creating populated stats."""
stats = HistoricalPlanStats(
action_name="local/refactor",
sample_size=10,
mean_cost_usd=0.50,
median_cost_usd=0.40,
p90_cost_usd=0.90,
mean_duration_seconds=180.0,
median_duration_seconds=150.0,
avg_step_count=12.5,
avg_child_plan_count=3.0,
success_rate=0.85,
)
if stats.sample_size != 10:
print(f"fail: expected sample_size=10, got {stats.sample_size}")
return 1
if abs(stats.success_rate - 0.85) > 1e-6:
print(f"fail: expected success_rate=0.85, got {stats.success_rate}")
return 1
print("populated-stats-ok")
return 0
def _cmd_service_mock() -> int:
"""Test the service with mock plans."""
plans = [
_make_mock_plan(cost=0.10, duration_seconds=30.0, step_count=3, child_plans=0),
_make_mock_plan(cost=0.20, duration_seconds=60.0, step_count=5, child_plans=1),
_make_mock_plan(cost=0.30, duration_seconds=90.0, step_count=8, child_plans=2),
]
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=plans)
service = HistoricalPlanStatsService(repo)
stats = service.get_stats_for_action("local/refactor")
if stats.sample_size != 3:
print(f"fail: expected sample_size=3, got {stats.sample_size}")
return 1
if stats.mean_cost_usd <= 0:
print(f"fail: expected mean_cost > 0, got {stats.mean_cost_usd}")
return 1
if stats.success_rate <= 0:
print(f"fail: expected success_rate > 0, got {stats.success_rate}")
return 1
print(f"service-mock-ok: mean_cost={stats.mean_cost_usd:.4f}")
return 0
def _cmd_service_empty() -> int:
"""Test the service with empty history."""
repo = MagicMock()
repo.get_completed_plans_by_action = MagicMock(return_value=[])
service = HistoricalPlanStatsService(repo)
stats = service.get_stats_for_action("local/refactor")
if stats.sample_size != 0:
print(f"fail: expected sample_size=0, got {stats.sample_size}")
return 1
if stats.mean_cost_usd != 0.0:
print(f"fail: expected mean_cost=0.0, got {stats.mean_cost_usd}")
return 1
print("service-empty-ok")
return 0
def _cmd_provider_integration() -> int:
"""Test the ACMS context provider integration."""
mock_stats = HistoricalPlanStats(
action_name="local/refactor",
sample_size=5,
mean_cost_usd=0.15,
median_cost_usd=0.12,
p90_cost_usd=0.30,
mean_duration_seconds=90.0,
median_duration_seconds=80.0,
avg_step_count=6.0,
avg_child_plan_count=1.5,
success_rate=0.75,
)
mock_service = MagicMock(spec=HistoricalPlanStatsService)
mock_service.get_stats_for_action.return_value = mock_stats
provider = EstimationHistoricalStatsProvider(
mock_service, default_action_name="local/refactor"
)
# Check confidence
confidence = provider.can_handle({})
if abs(confidence - 0.8) > 1e-6:
print(f"fail: expected confidence=0.8, got {confidence}")
return 1
# Assemble fragments
budget = ContextBudget(max_tokens=4096, reserved_tokens=0)
fragments = list(provider.assemble([], budget))
stats_frags = [
f for f in fragments if f.metadata.get("type") == "historical_plan_stats"
]
if len(stats_frags) != 1:
print(f"fail: expected 1 stats fragment, got {len(stats_frags)}")
return 1
# Verify explain
explanation = provider.explain()
if "historical" not in explanation.lower():
print(f"fail: explain doesn't mention 'historical': {explanation}")
return 1
print("provider-integration-ok")
return 0
def _cmd_context_dict() -> int:
"""Test as_context_dict serialization."""
stats = HistoricalPlanStats(
action_name="local/refactor",
sample_size=5,
mean_cost_usd=0.15,
median_cost_usd=0.0,
p90_cost_usd=0.30,
mean_duration_seconds=90.0,
median_duration_seconds=0.0,
avg_step_count=6.0,
avg_child_plan_count=0.0,
success_rate=0.75,
)
d = stats.as_context_dict()
if "action_name" not in d:
print("fail: context dict missing action_name")
return 1
if "mean_cost_usd" not in d:
print("fail: context dict missing mean_cost_usd")
return 1
# Zero values should be excluded
if "median_cost_usd" in d:
print("fail: context dict should not include zero median_cost_usd")
return 1
if "avg_child_plan_count" in d:
print("fail: context dict should not include zero avg_child_plan_count")
return 1
print("context-dict-ok")
return 0
def _cmd_percentile_calc() -> int:
"""Test the _percentile helper."""
# Empty list
if _percentile([], 90) != 0.0:
print("fail: percentile of empty list should be 0.0")
return 1
# Single value
if _percentile([5.0], 90) != 5.0:
print("fail: percentile of single value should be that value")
return 1
# Known values: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], p90
values = sorted([float(i) for i in range(1, 11)])
p90 = _percentile(values, 90)
# p90 of [1..10] = 1 + 0.9 * 9 = 9.1
if abs(p90 - 9.1) > 0.01:
print(f"fail: expected p90~9.1, got {p90}")
return 1
print(f"percentile-ok: p90={p90}")
return 0
COMMANDS: dict[str, Callable[[], int]] = {
"empty-stats": _cmd_empty_stats,
"populated-stats": _cmd_populated_stats,
"service-mock": _cmd_service_mock,
"service-empty": _cmd_service_empty,
"provider-integration": _cmd_provider_integration,
"context-dict": _cmd_context_dict,
"percentile-calc": _cmd_percentile_calc,
}
def main() -> int:
"""Entry point."""
if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(COMMANDS)}>")
return 1
return COMMANDS[sys.argv[1]]()
if __name__ == "__main__":
sys.exit(main())
robot/historical_plan_stats.robot
+65
View File
@@ -0,0 +1,65 @@
*** Settings ***
Documentation Integration tests for historical plan statistics query service
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_historical_plan_stats.py
*** Test Cases ***
Create Empty Historical Stats
[Documentation] Create HistoricalPlanStats with empty factory and verify defaults
${result}= Run Process ${PYTHON} ${HELPER} empty-stats cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} empty-stats-ok
Create Populated Historical Stats
[Documentation] Create HistoricalPlanStats with all fields populated
${result}= Run Process ${PYTHON} ${HELPER} populated-stats cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} populated-stats-ok
Stats Service With Mock Plans
[Documentation] Test HistoricalPlanStatsService with mock plan data
${result}= Run Process ${PYTHON} ${HELPER} service-mock cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} service-mock-ok
Stats Service Empty History
[Documentation] Test HistoricalPlanStatsService with empty history
${result}= Run Process ${PYTHON} ${HELPER} service-empty cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} service-empty-ok
Context Provider Integration
[Documentation] Test EstimationHistoricalStatsProvider ACMS integration
${result}= Run Process ${PYTHON} ${HELPER} provider-integration cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} provider-integration-ok
Context Dict Serialization
[Documentation] Test as_context_dict serialization
${result}= Run Process ${PYTHON} ${HELPER} context-dict cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-dict-ok
Percentile Calculation
[Documentation] Test _percentile helper function
${result}= Run Process ${PYTHON} ${HELPER} percentile-calc cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} percentile-ok
src/cleveragents/application/services/estimation_context_provider.py
+215
View File
@@ -0,0 +1,215 @@
"""Estimation actor context provider for ACMS integration.
Provides ``EstimationHistoricalStatsProvider`` — an ACMS context
strategy that injects historical plan statistics into the estimation
actor's hot context tier. When the estimation actor assembles its
context, this provider queries ``HistoricalPlanStatsService`` for the
current action's historical data and creates a ``ContextFragment``
containing the serialized statistics.
The provider implements the ``ContextStrategy`` protocol from
``acms_service.py`` and can be registered with the ACMS pipeline via
``register_strategy()``.
Based on ``docs/specification.md`` lines 19077-19081 (estimation actor
analyzes "Historical data from similar plans (if available)") and the
ACMS context strategy protocol (~line 25167).
ISSUES CLOSED: #652
"""
from __future__ import annotations
import json
from collections.abc import Sequence
from typing import Any
import structlog
from cleveragents.application.services.acms_service import StrategyCapabilities
from cleveragents.application.services.historical_plan_stats_service import (
HistoricalPlanStatsService,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
FragmentProvenance,
)
__all__ = ["EstimationHistoricalStatsProvider"]
_log = structlog.get_logger(__name__)
class EstimationHistoricalStatsProvider:
"""ACMS context strategy that provides historical plan statistics.
When registered with the ACMS pipeline and activated for the
estimation actor's context assembly, this strategy queries the
``HistoricalPlanStatsService`` for the current action's completed
plan statistics and injects them as a high-relevance context
fragment.
The fragment contains a JSON-serialized representation of the
``HistoricalPlanStats`` suitable for the estimation actor to parse
and reason about.
Implements the ``ContextStrategy`` protocol.
Usage::
stats_service = HistoricalPlanStatsService(plan_repo)
provider = EstimationHistoricalStatsProvider(stats_service)
pipeline.register_strategy(provider)
"""
STRATEGY_NAME = "estimation-historical-stats"
"""Registered strategy name in the ACMS pipeline."""
def __init__(
self,
stats_service: HistoricalPlanStatsService,
*,
default_action_name: str = "",
stats_limit: int = 50,
) -> None:
"""Initialise with a ``HistoricalPlanStatsService``.
Args:
stats_service: Service for querying historical plan statistics.
default_action_name: Action name to query when no action is
specified in the context request.
stats_limit: Maximum number of historical plans to consider.
"""
self._stats_service = stats_service
self._action_name = default_action_name
self._stats_limit = stats_limit
def set_action_name(self, action_name: str) -> None:
"""Set the action name to query historical stats for.
Called before context assembly to configure which action's
history should be queried.
Args:
action_name: Namespaced action name.
"""
self._action_name = action_name
@property
def name(self) -> str:
"""Return the strategy name."""
return self.STRATEGY_NAME
@property
def capabilities(self) -> StrategyCapabilities:
"""Return strategy capabilities."""
return StrategyCapabilities(
supports_semantic_search=False,
supports_graph_navigation=False,
supports_temporal_archaeology=True,
quality_score=0.7,
)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return confidence that this strategy can provide useful context.
Returns 0.8 when an action name is configured and 0.0 otherwise.
The estimation context provider is highly relevant when an action
is specified, but useless without one.
"""
action = request.get("action_name", self._action_name)
if action:
return 0.8
return 0.0
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Assemble historical stats as context fragments.
Queries the ``HistoricalPlanStatsService`` for the configured
action and produces a single ``ContextFragment`` containing the
serialized statistics. If there is no history or the stats
fragment would exceed the budget, returns the input fragments
unchanged.
Args:
fragments: Existing fragments (passed through unchanged).
budget: Token budget for this strategy's allocation.
Returns:
Input fragments plus the historical stats fragment (if any).
"""
if not self._action_name:
return list(fragments)
try:
stats = self._stats_service.get_stats_for_action(
self._action_name,
limit=self._stats_limit,
)
except Exception:
_log.warning(
"historical_stats_query_failed",
action_name=self._action_name,
exc_info=True,
)
return list(fragments)
if stats.sample_size == 0:
_log.debug(
"no_historical_stats",
action_name=self._action_name,
)
return list(fragments)
# Serialize stats to a context fragment
stats_dict = stats.as_context_dict()
stats_text = (
f"Historical Plan Statistics for action '{self._action_name}':\n"
f"{json.dumps(stats_dict, indent=2)}"
)
# Estimate token count (rough: ~4 chars per token)
estimated_tokens = max(1, len(stats_text) // 4)
if estimated_tokens > budget.available_tokens:
_log.debug(
"historical_stats_exceeds_budget",
action_name=self._action_name,
estimated_tokens=estimated_tokens,
available_tokens=budget.available_tokens,
)
return list(fragments)
stats_fragment = ContextFragment(
fragment_id=f"historical-stats-{self._action_name}",
uko_node=f"historical-stats://{self._action_name}",
content=stats_text,
token_count=estimated_tokens,
relevance_score=0.85,
provenance=FragmentProvenance(
resource_uri=f"historical-stats://{self._action_name}",
),
metadata={
"type": "historical_plan_stats",
"action_name": self._action_name,
"sample_size": str(stats.sample_size),
},
)
result = list(fragments)
result.append(stats_fragment)
return result
def explain(self) -> str:
"""Return a human-readable explanation of this strategy."""
return (
"Queries historical plan statistics for the current action "
"and injects them as a context fragment for the estimation "
"actor. Provides mean/median/p90 cost, duration, step count, "
"child plan count, and success rate from completed plans."
)
src/cleveragents/application/services/historical_plan_stats_service.py
+245
View File
@@ -0,0 +1,245 @@
"""Historical plan statistics query service for estimation context assembly.
Provides ``HistoricalPlanStatsService`` which queries completed plans
by action name and returns aggregated statistics (mean/median/p90 cost,
duration, step count, success rate) packaged in a
``HistoricalPlanStats`` value object.
The service is designed for integration with the ACMS context assembly
pipeline: the estimation actor receives historical stats as part of its
hot context tier, enabling data-driven cost and risk estimation.
Empty history (first run for an action) returns a valid empty stats
object rather than raising an error, ensuring the estimation actor can
always receive a well-formed input.
Based on ``docs/specification.md`` lines 19077-19081 (estimation actor
analyzes "Historical data from similar plans (if available)").
ISSUES CLOSED: #652
"""
from __future__ import annotations
import math
import statistics
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any
import structlog
from cleveragents.domain.models.core.historical_plan_stats import HistoricalPlanStats
if TYPE_CHECKING:
from cleveragents.infrastructure.database.repositories import (
LifecyclePlanRepository,
)
__all__ = ["HistoricalPlanStatsService"]
_log = structlog.get_logger(__name__)
def _percentile(sorted_values: list[float], pct: float) -> float:
"""Compute the *pct*-th percentile of an already-sorted list.
Uses linear interpolation between the two closest data points.
Returns ``0.0`` for empty input.
Args:
sorted_values: Non-descending list of floats.
pct: Percentile in the range [0, 100].
Returns:
The computed percentile value.
"""
if not sorted_values:
return 0.0
n = len(sorted_values)
if n == 1:
return sorted_values[0]
k = (pct / 100.0) * (n - 1)
floor_k = math.floor(k)
ceil_k = min(floor_k + 1, n - 1)
fraction = k - floor_k
return sorted_values[floor_k] + fraction * (
sorted_values[ceil_k] - sorted_values[floor_k]
)
class HistoricalPlanStatsService:
"""Query and aggregate historical plan statistics for an action.
Designed for use by the estimation actor's context assembly. The
service queries the ``LifecyclePlanRepository`` for completed plans
matching a given action name, then computes aggregated statistics.
Usage::
service = HistoricalPlanStatsService(plan_repository)
stats = service.get_stats_for_action("local/refactor", limit=50)
# stats.mean_cost_usd, stats.success_rate, ...
Thread-safety: this service is stateless (no mutable instance state
beyond the injected repository reference) and safe for concurrent use.
"""
def __init__(self, plan_repository: LifecyclePlanRepository) -> None:
"""Initialise with a plan repository.
Args:
plan_repository: The repository providing plan query access.
"""
self._plan_repository = plan_repository
def get_stats_for_action(
self,
action_name: str,
limit: int = 50,
) -> HistoricalPlanStats:
"""Compute aggregated statistics for completed plans of *action_name*.
Queries up to *limit* most recent completed plans for the given
action and returns aggregated statistics. If no completed plans
exist, returns a valid empty stats object (``sample_size == 0``).
Args:
action_name: Namespaced action name (e.g. ``"local/refactor"``).
limit: Maximum number of recent plans to consider (default 50).
Returns:
A frozen ``HistoricalPlanStats`` value object.
"""
if not action_name:
return HistoricalPlanStats.empty(action_name="<unknown>")
if limit < 1:
return HistoricalPlanStats.empty(action_name=action_name)
plans = self._plan_repository.get_completed_plans_by_action(
action_name, limit=limit
)
if not plans:
_log.debug(
"no_completed_plans_for_action",
action_name=action_name,
)
return HistoricalPlanStats.empty(action_name=action_name)
return self._aggregate(action_name, plans)
def _aggregate(
self,
action_name: str,
plans: Sequence[Any],
) -> HistoricalPlanStats:
"""Compute aggregate statistics from a sequence of plan domain objects.
Args:
action_name: The action name for the stats object.
plans: Non-empty sequence of Plan domain objects.
Returns:
Populated ``HistoricalPlanStats``.
"""
costs: list[float] = []
durations: list[float] = []
step_counts: list[int] = []
child_plan_counts: list[int] = []
applied_count = 0
for plan in plans:
# Cost: from cost_metadata.total_cost or the DB column cost_actual_usd
cost = self._extract_cost(plan)
if cost is not None and cost >= 0.0:
costs.append(cost)
# Duration: from timestamps (created_at to completed_at/applied_at)
duration = self._extract_duration_seconds(plan)
if duration is not None and duration >= 0.0:
durations.append(duration)
# Step count: from last_completed_step
step_count = getattr(plan, "last_completed_step", -1)
if step_count is not None and step_count >= 0:
step_counts.append(step_count + 1) # Convert 0-indexed to count
# Child plan count: from subplan_statuses
subplan_statuses = getattr(plan, "subplan_statuses", []) or []
child_plan_counts.append(len(subplan_statuses))
# Success rate: count applied plans
state = getattr(plan, "processing_state", None)
if state is not None:
state_val = state.value if hasattr(state, "value") else str(state)
else:
state_val = ""
if state_val == "applied":
applied_count += 1
sample_size = len(plans)
sorted_costs = sorted(costs)
return HistoricalPlanStats(
action_name=action_name,
sample_size=sample_size,
mean_cost_usd=statistics.mean(costs) if costs else 0.0,
median_cost_usd=statistics.median(costs) if costs else 0.0,
p90_cost_usd=_percentile(sorted_costs, 90.0),
mean_duration_seconds=statistics.mean(durations) if durations else 0.0,
median_duration_seconds=(
statistics.median(durations) if durations else 0.0
),
avg_step_count=(statistics.mean(step_counts) if step_counts else 0.0),
avg_child_plan_count=(
statistics.mean(child_plan_counts) if child_plan_counts else 0.0
),
success_rate=applied_count / sample_size if sample_size > 0 else 0.0,
)
@staticmethod
def _extract_cost(plan: Any) -> float | None:
"""Extract actual cost from a plan domain object.
Prefers ``cost_metadata.total_cost``; if unavailable or zero,
falls back to the DB column equivalent if available via the
plan's attributes.
"""
cost_meta = getattr(plan, "cost_metadata", None)
if cost_meta is not None:
total = getattr(cost_meta, "total_cost", None)
if total is not None and total > 0.0:
return float(total)
# No cost metadata — return None (plan had no cost tracking)
return None
@staticmethod
def _extract_duration_seconds(plan: Any) -> float | None:
"""Extract total duration in seconds from a plan's timestamps.
Duration is ``applied_at - created_at`` for applied plans or
``execute_completed_at - created_at`` for completed (but not
applied) plans. Returns ``None`` if timestamps are unavailable.
"""
timestamps = getattr(plan, "timestamps", None)
if timestamps is None:
return None
created_at = getattr(timestamps, "created_at", None)
if created_at is None:
return None
# Prefer applied_at as the end timestamp (full lifecycle)
end_at = getattr(timestamps, "applied_at", None)
if end_at is None:
# Fall back to execute_completed_at
end_at = getattr(timestamps, "execute_completed_at", None)
if end_at is None:
return None
try:
delta = end_at - created_at
return max(0.0, delta.total_seconds())
except (TypeError, AttributeError):
return None
src/cleveragents/domain/models/core/__init__.py
+2
View File
@@ -144,6 +144,7 @@ from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.domain.models.core.historical_plan_stats import HistoricalPlanStats
from cleveragents.domain.models.core.invariant import (
Invariant,
InvariantEnforcementRecord,
@@ -409,6 +410,7 @@ __all__ = [
"GuardrailEventType",
"GuardrailResult",
"HistoricalOutcome",
"HistoricalPlanStats",
"InMemoryChangeSetStore",
"InMemoryInvocationTracker",
"IndexMetadata",
src/cleveragents/domain/models/core/historical_plan_stats.py
+142
View File
@@ -0,0 +1,142 @@
"""Historical plan statistics domain model for CleverAgents.
Provides the ``HistoricalPlanStats`` frozen value object that packages
aggregated statistics from completed plans for a given action. Used by
the estimation actor's context assembly to provide historical data for
informed cost/risk estimation.
Based on ``docs/specification.md`` estimation actor sections (lines
19077-19081) which describe the three inputs the estimation actor
analyzes, including "Historical data from similar plans (if available)."
ISSUES CLOSED: #652
"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
__all__ = ["HistoricalPlanStats"]
class HistoricalPlanStats(BaseModel):
"""Aggregated statistics from completed plans for a given action.
All fields have sensible defaults so that an empty-history scenario
(first run for a given action) produces a valid, non-error stats
object with ``sample_size == 0`` and all metrics at ``0.0``.
The model is frozen (immutable) once constructed.
"""
action_name: str = Field(
...,
min_length=1,
description="Namespaced action name the statistics describe",
)
sample_size: int = Field(
default=0,
ge=0,
description="Number of completed plans included in the statistics",
)
# -- Cost metrics (USD) ------------------------------------------------
mean_cost_usd: float = Field(
default=0.0,
ge=0.0,
description="Mean actual cost across completed plans (USD)",
)
median_cost_usd: float = Field(
default=0.0,
ge=0.0,
description="Median actual cost across completed plans (USD)",
)
p90_cost_usd: float = Field(
default=0.0,
ge=0.0,
description="90th percentile actual cost (USD)",
)
# -- Duration metrics (seconds) ----------------------------------------
mean_duration_seconds: float = Field(
default=0.0,
ge=0.0,
description="Mean total duration from created_at to completed_at (seconds)",
)
median_duration_seconds: float = Field(
default=0.0,
ge=0.0,
description="Median total duration (seconds)",
)
# -- Step and child plan metrics ---------------------------------------
avg_step_count: float = Field(
default=0.0,
ge=0.0,
description="Average number of completed execution steps",
)
avg_child_plan_count: float = Field(
default=0.0,
ge=0.0,
description="Average number of child plans spawned",
)
# -- Success rate ------------------------------------------------------
success_rate: float = Field(
default=0.0,
ge=0.0,
le=1.0,
description=(
"Fraction of completed plans that reached 'applied' state (0.0-1.0)"
),
)
model_config = ConfigDict(frozen=True, allow_inf_nan=False)
@classmethod
def empty(cls, action_name: str) -> HistoricalPlanStats:
"""Create an empty stats object for an action with no history.
Args:
action_name: The namespaced action name.
Returns:
A ``HistoricalPlanStats`` with ``sample_size == 0`` and all
metrics at their zero defaults.
"""
return cls(action_name=action_name)
def as_context_dict(self) -> dict[str, Any]:
"""Return a dict suitable for injection into estimation actor context.
Produces a compact representation with only non-zero metrics
included, plus the action name and sample size (always present).
"""
d: dict[str, Any] = {
"action_name": self.action_name,
"sample_size": self.sample_size,
}
if self.mean_cost_usd > 0.0:
d["mean_cost_usd"] = round(self.mean_cost_usd, 6)
if self.median_cost_usd > 0.0:
d["median_cost_usd"] = round(self.median_cost_usd, 6)
if self.p90_cost_usd > 0.0:
d["p90_cost_usd"] = round(self.p90_cost_usd, 6)
if self.mean_duration_seconds > 0.0:
d["mean_duration_seconds"] = round(self.mean_duration_seconds, 2)
if self.median_duration_seconds > 0.0:
d["median_duration_seconds"] = round(self.median_duration_seconds, 2)
if self.avg_step_count > 0.0:
d["avg_step_count"] = round(self.avg_step_count, 2)
if self.avg_child_plan_count > 0.0:
d["avg_child_plan_count"] = round(self.avg_child_plan_count, 2)
if self.success_rate > 0.0:
d["success_rate"] = round(self.success_rate, 4)
return d
src/cleveragents/infrastructure/database/repositories.py
+47
View File
@@ -1612,6 +1612,53 @@ class LifecyclePlanRepository:
except (OperationalError, SQLAlchemyDatabaseError) as exc:
raise DatabaseError(f"Failed to count plans: {exc}") from exc
@database_retry
def get_completed_plans_by_action(
self,
action_name: str,
limit: int = 50,
) -> list[Any]:
"""Return the *limit* most recent completed plans for *action_name*.
A plan is considered "completed" when its ``processing_state`` is
one of ``complete``, ``applied``, ``errored``, or ``cancelled``.
Only plans that have actually finished processing (i.e. reached a
terminal state) contribute to historical statistics.
Results are ordered by ``created_at DESC`` so the most recent
plans appear first.
Args:
action_name: Namespaced action name to filter on.
limit: Maximum number of plans to return (default 50).
Returns:
List of ``Plan`` domain objects in reverse chronological order.
Raises:
DatabaseError: On transient or unexpected DB errors.
"""
if limit < 1:
return []
session = self._session()
terminal_states = ("complete", "applied", "errored", "cancelled")
try:
rows = (
session.query(LifecyclePlanModel)
.filter(
LifecyclePlanModel.action_name == action_name,
LifecyclePlanModel.processing_state.in_(terminal_states),
)
.order_by(LifecyclePlanModel.created_at.desc())
.limit(limit)
.all()
)
return [row.to_domain() for row in rows]
except (OperationalError, SQLAlchemyDatabaseError) as exc:
raise DatabaseError(
f"Failed to get completed plans for action '{action_name}': {exc}"
) from exc
# ---------------------------------------------------------------------------
# Resource Registry Repositories