From 2aa0ff64bb6158bd419373c8f5ab8863cc1e07ed Mon Sep 17 00:00:00 2001 From: "Brent E. Edwards" Date: Thu, 2 Apr 2026 09:10:49 +0000 Subject: [PATCH] feat(estimation): build historical plan statistics query service for estimation context assembly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the historical plan statistics infrastructure for the estimation actor's context assembly, as specified in docs/specification.md lines 19077-19081 (estimation actor analyzes "Historical data from similar plans (if available)"). Components added: 1. HistoricalPlanStats — Frozen Pydantic v2 value object in domain/models/core/ packaging mean/median/p90 cost, mean/median duration, average step count, average child plan count, and success rate. Includes empty() factory for first-run scenarios and as_context_dict() for ACMS context fragment serialization. 2. LifecyclePlanRepository.get_completed_plans_by_action() — Repository method filtering on terminal states (complete, applied, errored, cancelled) ordered by created_at DESC with retry logic. 3. HistoricalPlanStatsService — Stateless application service that queries the repository and computes aggregate statistics using stdlib statistics module. Handles missing cost/timestamp/step data gracefully. 4. EstimationHistoricalStatsProvider — ACMS ContextStrategy implementation that injects historical stats as a high-relevance context fragment into the estimation actor's hot context tier. Handles service failures, empty history, and budget constraints. Testing: 27 Behave scenarios covering model validation, service logic (empty/normal/edge cases), ACMS provider integration, and repository methods. 7 Robot integration tests with helper script. All quality gates pass: lint, typecheck (0 errors), unit_tests (556 features, 13798 scenarios), integration_tests (1875 tests), e2e_tests, and coverage_report (98.6%, threshold 97%). ISSUES CLOSED: #652 --- features/historical_plan_stats.feature | 157 +++++ features/steps/historical_plan_stats_steps.py | 626 ++++++++++++++++++ robot/helper_historical_plan_stats.py | 281 ++++++++ robot/historical_plan_stats.robot | 65 ++ .../services/estimation_context_provider.py | 215 ++++++ .../services/historical_plan_stats_service.py | 245 +++++++ .../domain/models/core/__init__.py | 2 + .../models/core/historical_plan_stats.py | 142 ++++ .../infrastructure/database/repositories.py | 47 ++ 9 files changed, 1780 insertions(+) create mode 100644 features/historical_plan_stats.feature create mode 100644 features/steps/historical_plan_stats_steps.py create mode 100644 robot/helper_historical_plan_stats.py create mode 100644 robot/historical_plan_stats.robot create mode 100644 src/cleveragents/application/services/estimation_context_provider.py create mode 100644 src/cleveragents/application/services/historical_plan_stats_service.py create mode 100644 src/cleveragents/domain/models/core/historical_plan_stats.py diff --git a/features/historical_plan_stats.feature b/features/historical_plan_stats.feature new file mode 100644 index 00000000..d3adfee4 --- /dev/null +++ b/features/historical_plan_stats.feature @@ -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 "" + 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 diff --git a/features/steps/historical_plan_stats_steps.py b/features/steps/historical_plan_stats_steps.py new file mode 100644 index 00000000..93437237 --- /dev/null +++ b/features/steps/historical_plan_stats_steps.py @@ -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}" diff --git a/robot/helper_historical_plan_stats.py b/robot/helper_historical_plan_stats.py new file mode 100644 index 00000000..aa56b41a --- /dev/null +++ b/robot/helper_historical_plan_stats.py @@ -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()) diff --git a/robot/historical_plan_stats.robot b/robot/historical_plan_stats.robot new file mode 100644 index 00000000..8cab8d94 --- /dev/null +++ b/robot/historical_plan_stats.robot @@ -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 diff --git a/src/cleveragents/application/services/estimation_context_provider.py b/src/cleveragents/application/services/estimation_context_provider.py new file mode 100644 index 00000000..c073ca6e --- /dev/null +++ b/src/cleveragents/application/services/estimation_context_provider.py @@ -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." + ) diff --git a/src/cleveragents/application/services/historical_plan_stats_service.py b/src/cleveragents/application/services/historical_plan_stats_service.py new file mode 100644 index 00000000..2f1ce2ca --- /dev/null +++ b/src/cleveragents/application/services/historical_plan_stats_service.py @@ -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="") + + 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 diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 1bdc61ec..c56884d8 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -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", diff --git a/src/cleveragents/domain/models/core/historical_plan_stats.py b/src/cleveragents/domain/models/core/historical_plan_stats.py new file mode 100644 index 00000000..12d8b015 --- /dev/null +++ b/src/cleveragents/domain/models/core/historical_plan_stats.py @@ -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 diff --git a/src/cleveragents/infrastructure/database/repositories.py b/src/cleveragents/infrastructure/database/repositories.py index b193719b..deae0202 100644 --- a/src/cleveragents/infrastructure/database/repositories.py +++ b/src/cleveragents/infrastructure/database/repositories.py @@ -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