diff --git a/src/cleveragents/application/services/plan_executor.py b/src/cleveragents/application/services/plan_executor.py index 8952b1636..fcf95cedd 100644 --- a/src/cleveragents/application/services/plan_executor.py +++ b/src/cleveragents/application/services/plan_executor.py @@ -1,78 +1,80 @@ -"""Plan executor service: stub actors for Strategize and Execute phases. - -Provides local-only (no LLM) stub actors for M1 that integrate with the -``PlanLifecycleService`` to drive plans through the Strategize and Execute -phases. When a ``PlanExecutionContext`` is provided, the execute phase -delegates to ``RuntimeExecuteActor`` for full changeset capture. - -Updated in M4 to add optional checkpoint hooks via ``CheckpointManager``. -Updated in M5 to wire ``SubplanService`` and ``SubplanExecutionService`` -into the Execute phase so that ``subplan_spawn`` and -``subplan_parallel_spawn`` decisions are realised as actual child plan -executions. -Updated in M6 to wire StrategyActor decisions through to Execute phase. -""" - -from __future__ import annotations - import json import time import traceback -from collections.abc import Callable -from datetime import UTC, datetime -from typing import TYPE_CHECKING, Any +from datetime import datetime, timezone +from typing import Any, Callable, List, Optional import structlog -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, ValidationError from ulid import ULID +from cleveragents.application.actors import ( + BaseActor, + ActorExecuteError, + ActorToolCall, + ActorParseError, + RuntimeExecuteActor, +) +from cleveragents.application.commands import ( + ExecutePlan, + RecordResultFromHistory, + RecordResult, + MarkPlanFailed, +) +from cleveragents.application.events import ( + PlanEvent, + PlanExecutionStarted, + PlanExecutionCompleted, + PlanExecutionError, +) +from cleveragents.application.exceptions import PlanValidationError, PlanError +from cleveragents.application.models import ( + Plan, + PlanPhase, + ProcessingState, + PlanExecutionContext, + PlanInvariant, + Policy, + SubplanConfig, + StrategyTree, + ChangeSet, + ChangeSetStore, + EstimationResult, + RuntimeExecuteResult, +) +from cleveragents.application.models.plan_execution import OperationalMetricKey +from cleveragents.application.repositories import ( + PlanPersistenceProtocol, +) from cleveragents.application.services.autonomy_guardrail_service import ( AutonomyGuardrailService, ) -from cleveragents.application.services.fix_then_revalidate import ( - FixThenRevalidateOrchestrator, -) -from cleveragents.application.services.plan_execution_context import ( - PlanExecutionContext, - RuntimeExecuteActor, - RuntimeExecuteResult, -) -from cleveragents.application.services.strategy_models import StrategyTree -from cleveragents.core.exceptions import PlanError, ValidationError -from cleveragents.domain.models.core.change import ChangeSetStore -from cleveragents.domain.models.core.estimation import EstimationResult -from cleveragents.domain.models.core.plan import ( - PlanInvariant, - PlanPhase, - ProcessingState, - SubplanConfig, -) -from cleveragents.domain.models.observability.metrics import ( - MetricCollector, - OperationalMetricKey, -) -from cleveragents.infrastructure.sandbox.checkpoint import ( +from cleveragents.application.services.checkpoint_manager import ( CheckpointManager, SandboxCheckpoint, ) -from cleveragents.tool.builtins.changeset import ChangeSet, ChangeSetCapture -from cleveragents.tool.runner import ToolRunner +from cleveragents.application.services.error_recovery_service import ( + ErrorRecoveryService, +) +from cleveragents.application.services.fix_revalidate_orchestrator import ( + FixThenRevalidateOrchestrator, +) +from cleveragents.application.services.lifecycle_service import PlanLifecycleService +from cleveragents.application.services.subplan_execution_service import ( + SubplanExecutionResult, + SubplanExecutionService, +) +from cleveragents.application.services.subplan_service import ( + SpawnResult, + SubplanService, +) +from cleveragents.application.services.tool_runner import ToolRunner +from cleveragents.infrastructure.filesystems import ChangeSetCapture +from cleveragents.infrastructure.observability.metrics_emitter import ( + MetricsEmitter, +) -if TYPE_CHECKING: - from cleveragents.application.services.error_recovery_service import ( - ErrorRecoveryService, - ) - from cleveragents.application.services.subplan_execution_service import ( - SubplanExecutionResult, - SubplanExecutionService, - ) - from cleveragents.application.services.subplan_service import ( - SpawnResult, - SubplanService, - ) - from cleveragents.infrastructure.observability.metrics_emitter import ( - MetricsEmitter, - ) +UTC = timezone.utc logger = structlog.get_logger(__name__) @@ -821,7 +823,7 @@ class PlanExecutor: raise ValidationError("plan_id must not be empty") if self._execution_context is not None: return self._run_execute_with_runtime(plan_id, stream_callback) - return self._run_execute_with_stub(plan_id, stream_callback) + return self._run_execute_with_actor(plan_id, stream_callback) def _build_decisions(self, plan: Any) -> list[StrategyDecision]: """Build decisions for the Execute phase. @@ -953,12 +955,12 @@ class PlanExecutor: plan = self._lifecycle.get_plan(plan_id) plan.changeset_id = result.changeset_id plan.sandbox_refs = result.sandbox_refs - plan.error_details = { - "tool_call_count": str(result.tool_call_count), - "decisions_processed": str(len(result.decision_ids_processed)), - "execution_duration_ms": str(result.execution_duration_ms), - "mode": "runtime", - } + existing = dict(plan.error_details or {}) + existing["tool_call_count"] = str(result.tool_call_count) + existing["decisions_processed"] = str(len(result.decision_ids_processed)) + existing["execution_duration_ms"] = str(result.execution_duration_ms) + existing["mode"] = type(self._execute_actor).__name__ + plan.error_details = existing plan.timestamps.updated_at = datetime.now(tz=UTC) # Spawn and execute child subplans from spawn decisions @@ -996,21 +998,25 @@ class PlanExecutor: self._try_rollback_to_last_checkpoint(plan_id) error_msg = f"{type(exc).__name__}: {exc}" plan = self._lifecycle.get_plan(plan_id) - plan.error_details = { - "exception_type": type(exc).__name__, - "traceback": traceback.format_exc(), - "mode": "runtime", - } + existing = dict(plan.error_details or {}) + existing["exception_type"] = type(exc).__name__ + existing["traceback"] = traceback.format_exc() + existing["mode"] = type(self._execute_actor).__name__ + plan.error_details = existing self._lifecycle._commit_plan(plan) self._lifecycle.fail_execute(plan_id, error_msg) raise - def _run_execute_with_stub( + def _run_execute_with_actor( self, plan_id: str, stream_callback: StreamCallback | None = None, ) -> ExecuteResult: - """Execute using the legacy ExecuteStubActor with optional retry.""" + """Execute using the actor with optional retry. + + Generic executor wrapper — dispatches to whatever ``_execute_actor`` + has been configured (stub, LLM, etc.). + """ plan = self._guard_execute(plan_id) decisions = self._build_decisions(plan) @@ -1043,11 +1049,11 @@ class PlanExecutor: plan = self._lifecycle.get_plan(plan_id) plan.changeset_id = result.changeset_id plan.sandbox_refs = result.sandbox_refs - plan.error_details = { - "tool_calls_count": str(result.tool_calls_count), - "sandbox_refs_count": str(len(result.sandbox_refs)), - "mode": "stub", - } + existing = dict(plan.error_details or {}) + existing["tool_calls_count"] = str(result.tool_calls_count) + existing["sandbox_refs_count"] = str(len(result.sandbox_refs)) + existing["mode"] = type(self._execute_actor).__name__ + plan.error_details = existing plan.timestamps.updated_at = datetime.now(tz=UTC) # Spawn and execute child subplans from spawn decisions @@ -1116,11 +1122,11 @@ class PlanExecutor: self._try_rollback_to_last_checkpoint(plan_id) error_msg = f"{type(last_exc).__name__}: {last_exc}" plan = self._lifecycle.get_plan(plan_id) - plan.error_details = { - "exception_type": type(last_exc).__name__, - "traceback": traceback.format_exc(), - "mode": "stub", - } + existing = dict(plan.error_details or {}) + existing["exception_type"] = type(last_exc).__name__ + existing["traceback"] = traceback.format_exc() + existing["mode"] = type(self._execute_actor).__name__ + plan.error_details = existing self._lifecycle._commit_plan(plan) self._lifecycle.fail_execute(plan_id, error_msg) raise last_exc