diff --git a/benchmarks/plan_execute_bench.py b/benchmarks/plan_execute_bench.py new file mode 100644 index 00000000..575f4fdd --- /dev/null +++ b/benchmarks/plan_execute_bench.py @@ -0,0 +1,215 @@ +"""Airspeed Velocity benchmarks for plan execute runtime integration. + +Measures construction, changeset operations, RuntimeExecuteActor +dispatch, and PlanExecutor mode selection overhead. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +from ulid import ULID + +from cleveragents.application.services.plan_execution_context import ( + PlanExecutionContext, + RuntimeExecuteActor, + RuntimeExecuteResult, +) +from cleveragents.application.services.plan_executor import ( + PlanExecutor, + StrategyDecision, +) +from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + InMemoryChangeSetStore, +) +from cleveragents.tool.registry import ToolRegistry +from cleveragents.tool.runner import ToolRunner + +_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6EA00" +_RESOURCE_ID = "01HGZ6FE0AQDYTR4BXVQZ6EB00" + + +def _make_runner() -> ToolRunner: + return ToolRunner(registry=ToolRegistry()) + + +def _make_decisions(count: int) -> list[StrategyDecision]: + root_id = str(ULID()) + return [ + StrategyDecision( + decision_id=root_id if i == 0 else str(ULID()), + step_text=f"Step {i + 1}", + sequence=i, + parent_id=root_id if i > 0 else None, + ) + for i in range(count) + ] + + +def _make_entry(plan_id: str, idx: int = 0) -> ChangeEntry: + return ChangeEntry( + plan_id=plan_id, + resource_id=_RESOURCE_ID, + tool_name="bench/tool", + operation=ChangeOperation.MODIFY, + path=f"src/bench_{idx}.py", + ) + + +class PlanExecutionContextSuite: + """Benchmark PlanExecutionContext construction and operations.""" + + def time_minimal_context_creation(self) -> None: + """Time creating context with only plan_id.""" + PlanExecutionContext(plan_id=_PLAN_ID) + + def time_full_context_creation(self) -> None: + """Time creating context with all optional fields.""" + PlanExecutionContext( + plan_id=_PLAN_ID, + decision_root_id=_RESOURCE_ID, + sandbox_root="/tmp/sandbox", + automation_profile="trusted", + project_resources={"repo": {"path": "/code"}}, + changeset_store=InMemoryChangeSetStore(), + ) + + def time_start_changeset(self) -> None: + """Time starting a new changeset.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + ctx.start_changeset() + + def time_record_change_single(self) -> None: + """Time recording a single change entry.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + ctx.start_changeset() + ctx.record_change(_make_entry(_PLAN_ID)) + + def time_record_change_batch_10(self) -> None: + """Time recording 10 change entries.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + ctx.start_changeset() + for i in range(10): + ctx.record_change(_make_entry(_PLAN_ID, i)) + + def time_summarize_empty(self) -> None: + """Time summarize with no changesets.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + ctx.summarize() + + def time_summarize_with_changeset(self) -> None: + """Time summarize with one changeset and 5 entries.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + ctx.start_changeset() + for i in range(5): + ctx.record_change(_make_entry(_PLAN_ID, i)) + ctx.summarize() + + +class RuntimeExecuteResultSuite: + """Benchmark RuntimeExecuteResult model creation.""" + + def time_result_creation_minimal(self) -> None: + """Time creating result with only required fields.""" + RuntimeExecuteResult(changeset_id=str(ULID())) + + def time_result_creation_full(self) -> None: + """Time creating result with all fields populated.""" + RuntimeExecuteResult( + changeset_id=str(ULID()), + tool_call_count=42, + sandbox_refs=["/tmp/sb1", "/tmp/sb2"], + decision_ids_processed=[str(ULID()) for _ in range(5)], + execution_duration_ms=1500.0, + ) + + +class RuntimeExecuteActorSuite: + """Benchmark RuntimeExecuteActor execution.""" + + def setup(self) -> None: + self.store = InMemoryChangeSetStore() + self.runner = _make_runner() + + def time_single_decision_execution(self) -> None: + """Time executing a single decision.""" + ctx = PlanExecutionContext( + plan_id=_PLAN_ID, + changeset_store=self.store, + ) + actor = RuntimeExecuteActor( + tool_runner=self.runner, + execution_context=ctx, + ) + actor.execute(decisions=_make_decisions(1)) + + def time_five_decision_execution(self) -> None: + """Time executing five decisions.""" + ctx = PlanExecutionContext( + plan_id=_PLAN_ID, + changeset_store=self.store, + ) + actor = RuntimeExecuteActor( + tool_runner=self.runner, + execution_context=ctx, + ) + actor.execute(decisions=_make_decisions(5)) + + def time_execution_with_callback(self) -> None: + """Time executing with a stream callback.""" + ctx = PlanExecutionContext( + plan_id=_PLAN_ID, + changeset_store=self.store, + ) + actor = RuntimeExecuteActor( + tool_runner=self.runner, + execution_context=ctx, + ) + events: list[tuple[str, dict]] = [] + actor.execute( + decisions=_make_decisions(3), + stream_callback=lambda t, d: events.append((t, d)), + ) + + +class PlanExecutorModeSuite: + """Benchmark PlanExecutor construction with different modes.""" + + def time_stub_mode_construction(self) -> None: + """Time constructing PlanExecutor in stub mode.""" + PlanExecutor(lifecycle_service=MagicMock()) + + def time_runtime_mode_construction(self) -> None: + """Time constructing PlanExecutor in runtime mode.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + PlanExecutor( + lifecycle_service=MagicMock(), + tool_runner=_make_runner(), + execution_context=ctx, + ) + + def time_has_runtime_check_stub(self) -> None: + """Time has_runtime property check in stub mode.""" + executor = PlanExecutor(lifecycle_service=MagicMock()) + _ = executor.has_runtime + + def time_has_runtime_check_runtime(self) -> None: + """Time has_runtime property check in runtime mode.""" + ctx = PlanExecutionContext(plan_id=_PLAN_ID) + executor = PlanExecutor( + lifecycle_service=MagicMock(), + execution_context=ctx, + ) + _ = executor.has_runtime + + def time_changeset_store_access(self) -> None: + """Time changeset_store property access in runtime mode.""" + store = InMemoryChangeSetStore() + ctx = PlanExecutionContext(plan_id=_PLAN_ID, changeset_store=store) + executor = PlanExecutor( + lifecycle_service=MagicMock(), + execution_context=ctx, + ) + _ = executor.changeset_store diff --git a/docs/reference/plan_execute.md b/docs/reference/plan_execute.md index 5f17b086..135ec4db 100644 --- a/docs/reference/plan_execute.md +++ b/docs/reference/plan_execute.md @@ -7,133 +7,189 @@ drive plans through the **Strategize** and **Execute** phases. In M1, these actors are local-only stubs (no LLM calls); future milestones will integrate real AI providers. +When a `PlanExecutionContext` is provided, the execute phase delegates to +`RuntimeExecuteActor` for full tool-calling runtime integration with +changeset capture through `ChangeSetStore`. + ## Architecture ``` PlanExecutor -├── StrategizeStubActor (read-only, produces decision tree) -├── ExecuteStubActor (sandbox + ToolRunner + ChangeSet capture) -└── PlanLifecycleService (phase transitions, persistence) +├── StrategizeStubActor (read-only, produces decision tree) +├── ExecuteStubActor (legacy stub: sandbox + ChangeSetCapture) +├── RuntimeExecuteActor (runtime: ToolRunner + ChangeSetStore) +├── PlanExecutionContext (plan metadata + resource bindings) +└── PlanLifecycleService (phase transitions, persistence) ``` +## Execution Modes + +| Mode | Actor | Trigger | Output | +|---------|---------------------|-----------------------------------|------------------------| +| Stub | ExecuteStubActor | No `execution_context` | `ExecuteResult` | +| Runtime | RuntimeExecuteActor | `execution_context` is provided | `RuntimeExecuteResult` | + +## PlanExecutionContext + +The `PlanExecutionContext` bridges plan metadata into the tool runtime: + +```python +from cleveragents.application.services.plan_execution_context import ( + PlanExecutionContext, +) +from cleveragents.domain.models.core.change import InMemoryChangeSetStore + +ctx = PlanExecutionContext( + plan_id="01HGZ...", + decision_root_id="01HGZ...", + sandbox_root="/tmp/sandbox", + automation_profile="trusted", + project_resources={"repo": {"path": "/code"}}, + changeset_store=InMemoryChangeSetStore(), +) + +# Start a changeset for tracking mutations +changeset_id = ctx.start_changeset() + +# Record changes during execution +ctx.record_change(entry) + +# Retrieve changeset +cs = ctx.get_changeset(changeset_id) + +# Summarize context state +summary = ctx.summarize() +``` + +### Fields + +| Field | Type | Required | Description | +|---------------------|---------------------------|----------|------------------------------------| +| `plan_id` | `str` | Yes | ULID of the plan | +| `decision_root_id` | `str \| None` | No | Root decision from strategize | +| `sandbox_root` | `str \| None` | No | Sandbox filesystem path | +| `automation_profile` | `str \| None` | No | Automation profile name | +| `project_resources` | `dict[str, Any]` | No | Project resource metadata | +| `resource_bindings` | `dict[str, BoundResource]`| No | Resolved resource bindings | +| `changeset_store` | `ChangeSetStore` | No | Defaults to InMemoryChangeSetStore | + +## RuntimeExecuteActor + +Wraps `ToolRunner` to execute strategy decisions with changeset capture: + +```python +from cleveragents.application.services.plan_execution_context import ( + RuntimeExecuteActor, + RuntimeExecuteResult, +) + +actor = RuntimeExecuteActor( + tool_runner=runner, + execution_context=ctx, +) +result: RuntimeExecuteResult = actor.execute(decisions) +``` + +### RuntimeExecuteResult Fields + +| Field | Type | Description | +|--------------------------|---------------|----------------------------------| +| `changeset_id` | `str` | ULID of the produced changeset | +| `tool_call_count` | `int` | Number of tool calls made | +| `sandbox_refs` | `list[str]` | Sandbox reference paths | +| `decision_ids_processed` | `list[str]` | Processed decision node IDs | +| `execution_duration_ms` | `float` | Wall-clock execution time (ms) | + +## PlanExecutor Runtime Mode + +The `PlanExecutor` auto-selects runtime vs stub mode: + +```python +from cleveragents.application.services.plan_executor import PlanExecutor + +# Stub mode (no execution_context) +executor = PlanExecutor(lifecycle_service=lifecycle, tool_runner=runner) +assert not executor.has_runtime + +# Runtime mode (with execution_context) +executor = PlanExecutor( + lifecycle_service=lifecycle, + tool_runner=runner, + execution_context=ctx, +) +assert executor.has_runtime +assert executor.changeset_store is not None + +# Execute auto-dispatches to RuntimeExecuteActor +result = executor.run_execute(plan_id) +``` + +### Properties + +| Property | Type | Description | +|---------------------|-----------------------------|--------------------------------------| +| `has_runtime` | `bool` | True if execution_context is set | +| `changeset_store` | `ChangeSetStore \| None` | Store from execution context | +| `execution_context` | `PlanExecutionContext \| None` | The execution context | + +## ChangeSetStore Wiring + +The `ChangeSetStore` protocol defines the interface for changeset persistence: + +```python +class ChangeSetStore(Protocol): + def start(self, plan_id: str) -> str: ... + def record(self, changeset_id: str, entry: ChangeEntry) -> None: ... + def get(self, changeset_id: str) -> SpecChangeSet | None: ... + def get_for_plan(self, plan_id: str) -> list[SpecChangeSet]: ... + def summarize(self, changeset_id: str) -> dict[str, Any]: ... +``` + +`InMemoryChangeSetStore` is the default for M1. Database-backed +implementations will be added in D1 milestone. + ## Phase Lifecycle -| Phase | Actor | Mode | Output | -|-------------|---------------------|-----------|---------------------------------| -| Strategize | StrategizeStubActor | Read-only | Decision tree, invariant records| -| Execute | ExecuteStubActor | Sandbox | ChangeSet, execution metadata | - -## Strategize Phase - -The strategize phase is **read-only**: it produces a decision tree from the -action's `definition_of_done` without modifying any resources. - -### Decision Tree - -The stub actor parses `definition_of_done` into discrete steps, each -represented as a `StrategyDecision` node with a ULID identifier: - -```python -from cleveragents.application.services.plan_executor import ( - PlanExecutor, - StrategizeResult, -) - -executor = PlanExecutor(lifecycle_service=lifecycle, tool_runner=runner) -result: StrategizeResult = executor.run_strategize(plan_id) - -# result.decision_root_id -> ULID of root node -# result.decisions -> list[StrategyDecision] -# result.invariant_records -> list[dict] (stub enforcement records) -``` - -### Invariant Propagation - -Project and action invariants are propagated into the strategize context. -In M1, enforcement is stubbed (all invariants are accepted). Full -reconciliation via the Invariant Reconciliation Actor lands in D2. - -## Execute Phase - -The execute phase uses sandbox resources with tool calls routed through -`ToolRunner` and captured by `ChangeSetCapture`. - -### ChangeSet Capture - -All tool mutations during execute are recorded in a `ChangeSet`: - -```python -result: ExecuteResult = executor.run_execute(plan_id) - -# result.changeset_id -> ULID of the changeset -# result.changeset -> ChangeSet with entries -# result.tool_calls_count -> int -# result.sandbox_refs -> list[str] -``` - -### Metadata Persistence - -After execute completes, the following metadata is persisted on the Plan: - -- `changeset_id`: The ChangeSet identifier -- `sandbox_refs`: List of sandbox reference paths -- `error_details`: Tool call count and sandbox ref count - -## Phase Guards - -- **Execute requires Strategize COMPLETE**: The executor validates that the - plan has completed strategize (has a `decision_root_id`) before allowing - execute to proceed. -- **Phase validation**: Both `run_strategize()` and `run_execute()` verify - the plan is in the correct phase before proceeding. +| Phase | Actor | Mode | Output | +|------------|------------------------|-----------|----------------------------------| +| Strategize | StrategizeStubActor | Read-only | Decision tree, invariant records | +| Execute | RuntimeExecuteActor | Runtime | ChangeSet via ChangeSetStore | +| Execute | ExecuteStubActor | Stub | ChangeSet via ChangeSetCapture | ## Error Handling Failures in either phase are captured with full error context: - `error_message`: The exception message string -- `error_details`: Dict containing `exception_type` and `traceback` +- `error_details`: Dict with `exception_type`, `traceback`, and `mode` - The plan transitions to `ERRORED` processing state -### Retry Guidance - -Plans in `ERRORED` state can be retried by: - -1. Resetting the plan's processing state back to `QUEUED` -2. Re-running the failed phase via the executor - -Full retry automation is planned for D1b (Phase Reversion & Error Recovery). - ## Streaming Hooks -Both phases accept an optional `stream_callback` parameter for real-time -status updates (used by the `--stream` CLI flag): - -```python -def my_callback(event_type: str, data: dict) -> None: - print(f"[{event_type}] {data}") - -executor.run_strategize(plan_id, stream_callback=my_callback) -``` +Both phases accept an optional `stream_callback`: ### Event Types -| Event | Phase | Description | -|------------------------|------------|---------------------------------| -| `strategize_started` | Strategize | Phase processing began | -| `strategize_decisions` | Strategize | Decisions produced | -| `strategize_complete` | Strategize | Phase completed successfully | -| `execute_started` | Execute | Phase processing began | -| `execute_step` | Execute | Individual decision being executed | -| `execute_complete` | Execute | Phase completed successfully | +| Event | Phase | Actor | Description | +|----------------------------|----------|---------|---------------------------------| +| `strategize_started` | Strat. | Stub | Phase processing began | +| `strategize_decisions` | Strat. | Stub | Decisions produced | +| `strategize_complete` | Strat. | Stub | Phase completed | +| `execute_started` | Execute | Stub | Stub execute began | +| `execute_step` | Execute | Stub | Stub decision step | +| `execute_complete` | Execute | Stub | Stub execute completed | +| `runtime_execute_started` | Execute | Runtime | Runtime execute began | +| `runtime_execute_step` | Execute | Runtime | Runtime decision step | +| `runtime_execute_complete` | Execute | Runtime | Runtime execute completed | ## Module Reference -- **`cleveragents.application.services.plan_executor`**: Core module +- **`cleveragents.application.services.plan_execution_context`** + - `PlanExecutionContext`: Execution context bridging plan to runtime + - `RuntimeExecuteActor`: Tool-calling execute actor + - `RuntimeExecuteResult`: Runtime execution output model +- **`cleveragents.application.services.plan_executor`** + - `PlanExecutor`: Orchestrator with runtime/stub mode selection - `StrategizeStubActor`: Local-only strategize actor - `ExecuteStubActor`: Local-only execute actor - - `PlanExecutor`: Orchestrator connecting lifecycle service to actors - - `StrategyDecision`: Decision node model - - `StrategizeResult`: Strategize output model - - `ExecuteResult`: Execute output model - - `StreamCallback`: Type alias for streaming callbacks + - `StrategyDecision`, `StrategizeResult`, `ExecuteResult`: Data models diff --git a/features/plan_execute_runtime.feature b/features/plan_execute_runtime.feature new file mode 100644 index 00000000..35217b30 --- /dev/null +++ b/features/plan_execute_runtime.feature @@ -0,0 +1,221 @@ +@phase2 @plan @execute @runtime +Feature: Plan Execute Runtime Integration + As a system operator + I want the plan executor to dispatch through the tool-calling actor runtime + And capture changesets during execution + So that plan execution is fully traceable and auditable + + Background: + Given a fresh in-memory changeset store + And a valid plan ID + + # -- PlanExecutionContext creation -- + + @context + Scenario: Create PlanExecutionContext with required plan_id + When I create a PlanExecutionContext with a valid plan_id + Then the context plan_id should match the provided value + + @context + Scenario: Create PlanExecutionContext with all optional fields + When I create a PlanExecutionContext with all optional fields + Then the context should have the correct decision_root_id + And the context should have the correct sandbox_root + And the context should have the correct automation_profile + + @context @error + Scenario: PlanExecutionContext rejects empty plan_id + When I attempt to create a PlanExecutionContext with empty plan_id + Then the plan execution context should raise a ValidationError containing "plan_id" + + @context @error + Scenario: PlanExecutionContext rejects None plan_id + When I attempt to create a PlanExecutionContext with None plan_id + Then the plan execution context should raise a ValidationError containing "plan_id" + + @context + Scenario: PlanExecutionContext defaults to InMemoryChangeSetStore + When I create a PlanExecutionContext without specifying a changeset store + Then the context changeset_store should be an InMemoryChangeSetStore + + @context + Scenario: PlanExecutionContext accepts custom changeset store + When I create a PlanExecutionContext with a custom changeset store + Then the context changeset_store should be the provided store + + @context + Scenario: PlanExecutionContext defaults project_resources to empty dict + When I create a PlanExecutionContext with a valid plan_id + Then the context project_resources should be an empty dict + + @context + Scenario: PlanExecutionContext defaults resource_bindings to empty dict + When I create a PlanExecutionContext with a valid plan_id + Then the context resource_bindings should be an empty dict + + # -- Changeset operations -- + + @changeset + Scenario: start_changeset creates a changeset with ULID + Given a PlanExecutionContext + When I call start_changeset + Then the returned changeset_id should be a valid ULID string + And the active_changeset_ids should contain the new ID + + @changeset + Scenario: record_change persists an entry into the active changeset + Given a PlanExecutionContext with an active changeset + When I record a ChangeEntry into the context + Then the changeset should contain the recorded entry + + @changeset + Scenario: record_change without active changeset raises PlanError + Given a PlanExecutionContext without an active changeset + When I attempt to record a ChangeEntry + Then a PlanError should be raised about no active changeset + + @changeset + Scenario: Multiple changes recorded into same changeset + Given a PlanExecutionContext with an active changeset + When I record three ChangeEntry instances + Then the changeset should contain exactly three entries + + @changeset + Scenario: get_changeset retrieves existing changeset + Given a PlanExecutionContext with an active changeset and one recorded entry + When I call get_changeset with the active changeset_id + Then the returned changeset should not be None + And the changeset plan_id should match + + @changeset + Scenario: get_changeset returns None for unknown ID + Given a PlanExecutionContext + When I call get_changeset with a nonexistent ID + Then the returned changeset should be None + + # -- Summarize -- + + @summarize + Scenario: summarize includes all expected fields + Given a PlanExecutionContext with an active changeset and one recorded entry + When I call summarize on the context + Then the summary should contain plan_id + And the summary should contain resource_binding_count of 0 + And the summary should contain decision_root_id + + @summarize + Scenario: summarize with no changesets shows zero resource counts + Given a PlanExecutionContext + When I call summarize on the context + Then the summary resource_binding_count should be 0 + + # -- RuntimeExecuteResult model -- + + @model + Scenario: RuntimeExecuteResult validates with valid data + When I create a RuntimeExecuteResult with valid fields + Then the runtime execute result should have a correct changeset_id + And the runtime execute result tool_call_count should be non-negative + + @model @error + Scenario: RuntimeExecuteResult rejects negative tool_call_count + When I attempt to create a RuntimeExecuteResult with negative tool_call_count + Then the runtime result should fail validation + + @model @error + Scenario: RuntimeExecuteResult rejects empty changeset_id + When I attempt to create a RuntimeExecuteResult with empty changeset_id + Then the runtime result should fail validation + + # -- RuntimeExecuteActor -- + + @actor + Scenario: RuntimeExecuteActor executes with stub decisions + Given a PlanExecutionContext + And a ToolRunner with an empty registry + And a RuntimeExecuteActor + When I execute with a list of two stub decisions + Then the runtime execute result should be a RuntimeExecuteResult + And the runtime execute result changeset_id should be a valid ULID + And the runtime execute result decision_ids_processed should have two entries + + @actor + Scenario: RuntimeExecuteActor with streaming callback + Given a PlanExecutionContext + And a ToolRunner with an empty registry + And a RuntimeExecuteActor + And a stream callback collector + When I execute with one stub decision and the stream callback + Then the callback should have received runtime_execute_started event + And the callback should have received runtime_execute_step event + And the callback should have received runtime_execute_complete event + + @actor + Scenario: RuntimeExecuteActor records sandbox_refs when sandbox_root is set + Given a PlanExecutionContext with sandbox_root "/tmp/sandbox" + And a ToolRunner with an empty registry + And a RuntimeExecuteActor + When I execute with one stub decision + Then the runtime execute result sandbox_refs should contain "/tmp/sandbox" + + @actor @error + Scenario: RuntimeExecuteActor rejects None tool_runner + Given a PlanExecutionContext + When I attempt to create a RuntimeExecuteActor with None tool_runner + Then the plan execution context should raise a ValidationError + + @actor @error + Scenario: RuntimeExecuteActor rejects None execution_context + Given a ToolRunner with an empty registry + When I attempt to create a RuntimeExecuteActor with None execution_context + Then the plan execution context should raise a ValidationError + + # -- PlanExecutor integration -- + + @executor + Scenario: PlanExecutor has_runtime is True with tool_calling_runtime + Given a mock lifecycle service + And a PlanExecutionContext + And a ToolRunner with an empty registry + When I create a PlanExecutor with a mock tool_calling_runtime + Then has_runtime should be True + + @executor + Scenario: PlanExecutor has_runtime is False without execution_context + Given a mock lifecycle service + When I create a PlanExecutor without execution_context + Then has_runtime should be False + + @executor + Scenario: PlanExecutor changeset_store returns store from context + Given a mock lifecycle service + And a PlanExecutionContext with a known changeset store + When I create a PlanExecutor with the execution_context + Then changeset_store should be the same object as the context store + + @executor + Scenario: PlanExecutor changeset_store returns None without context + Given a mock lifecycle service + When I create a PlanExecutor without execution_context + Then changeset_store should be None + + @executor + Scenario: PlanExecutor execution_context property returns context + Given a mock lifecycle service + And a PlanExecutionContext + When I create a PlanExecutor with the execution_context + Then execution_context property should return the context + + # -- Error handling -- + + @error @executor + Scenario: PlanExecutor rejects None lifecycle_service + When I attempt to create a PlanExecutor with None lifecycle_service + Then the plan execution context should raise a ValidationError containing "lifecycle_service" + + @error @executor + Scenario: PlanExecutor run_execute rejects empty plan_id + Given a mock lifecycle service + And a PlanExecutor without execution_context + When I attempt to run execute with an empty plan_id + Then the plan execution context should raise a ValidationError containing "plan_id" diff --git a/features/steps/plan_execute_runtime_steps.py b/features/steps/plan_execute_runtime_steps.py new file mode 100644 index 00000000..8ea7fa78 --- /dev/null +++ b/features/steps/plan_execute_runtime_steps.py @@ -0,0 +1,589 @@ +"""Step definitions for plan execute runtime integration tests.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when +from pydantic import ValidationError as PydanticValidationError +from ulid import ULID + +from cleveragents.application.services.plan_execution_context import ( + PlanExecutionContext, + RuntimeExecuteActor, + RuntimeExecuteResult, +) +from cleveragents.application.services.plan_executor import ( + PlanExecutor, + StrategyDecision, +) +from cleveragents.core.exceptions import PlanError, ValidationError +from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + InMemoryChangeSetStore, +) +from cleveragents.tool.registry import ToolRegistry +from cleveragents.tool.runner import ToolRunner + + +def _runner() -> ToolRunner: + return ToolRunner(registry=ToolRegistry()) + + +def _make_decision_root_id() -> str: + return str(ULID()) + + +def _mock_lifecycle() -> MagicMock: + return MagicMock() + + +def _entry( + plan_id: str, tool: str = "test/tool", path: str = "src/t.py" +) -> ChangeEntry: + return ChangeEntry( + plan_id=plan_id, + resource_id=str(ULID()), + tool_name=tool, + operation=ChangeOperation.MODIFY, + path=path, + ) + + +def _make_decisions(count: int) -> list[StrategyDecision]: + root_id = str(ULID()) + decisions = [] + for i in range(count): + decisions.append( + StrategyDecision( + decision_id=root_id if i == 0 else str(ULID()), + step_text=f"Step {i + 1}", + sequence=i, + parent_id=root_id if i > 0 else None, + ) + ) + return decisions + + +def _make_ctx(plan_id: str, **kwargs: Any) -> PlanExecutionContext: + """Create a PlanExecutionContext with required fields.""" + defaults: dict[str, Any] = {"plan_id": plan_id} + defaults.update(kwargs) + return PlanExecutionContext(**defaults) + + +# -- Background -- + + +@given("a fresh in-memory changeset store") +def step_bg_store(ctx: Any) -> None: + ctx.changeset_store = InMemoryChangeSetStore() + + +@given("a valid plan ID") +def step_bg_plan_id(ctx: Any) -> None: + ctx.plan_id = str(ULID()) + + +# -- Context creation -- + + +@when("I create a PlanExecutionContext with a valid plan_id") +def step_ctx_valid(ctx: Any) -> None: + ctx.exec_ctx = _make_ctx(ctx.plan_id) + + +@when("I create a PlanExecutionContext with all optional fields") +def step_ctx_all(ctx: Any) -> None: + ctx.decision_root_id = str(ULID()) + ctx.sandbox_root = "/tmp/test-sandbox" + ctx.automation_profile = "trusted" + ctx.exec_ctx = PlanExecutionContext( + plan_id=ctx.plan_id, + decision_root_id=ctx.decision_root_id, + sandbox_root=ctx.sandbox_root, + automation_profile=ctx.automation_profile, + project_resources={"repo": {"path": "/code"}}, + resource_bindings={}, + changeset_store=ctx.changeset_store, + ) + + +@when("I attempt to create a PlanExecutionContext with empty plan_id") +def step_ctx_empty(ctx: Any) -> None: + try: + PlanExecutionContext(plan_id="") + ctx.error = None + except ValidationError as e: + ctx.error = e + + +@when("I attempt to create a PlanExecutionContext with None plan_id") +def step_ctx_none(ctx: Any) -> None: + try: + PlanExecutionContext(plan_id=None) + ctx.error = None + except (ValidationError, TypeError) as e: + ctx.error = e + + +@when("I create a PlanExecutionContext without specifying a changeset store") +def step_ctx_no_store(ctx: Any) -> None: + ctx.exec_ctx = _make_ctx(ctx.plan_id) + + +@when("I create a PlanExecutionContext with a custom changeset store") +def step_ctx_custom_store(ctx: Any) -> None: + ctx.custom_store = InMemoryChangeSetStore() + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.custom_store) + + +# -- Context assertions -- + + +@then("the context plan_id should match the provided value") +def step_a_plan_id(ctx: Any) -> None: + assert ctx.exec_ctx.plan_id == ctx.plan_id + + +@then("the context should have the correct decision_root_id") +def step_a_drid(ctx: Any) -> None: + assert ctx.exec_ctx.decision_root_id == ctx.decision_root_id + + +@then("the context should have the correct sandbox_root") +def step_a_sandbox(ctx: Any) -> None: + assert ctx.exec_ctx.sandbox_root == ctx.sandbox_root + + +@then("the context should have the correct automation_profile") +def step_a_profile(ctx: Any) -> None: + assert ctx.exec_ctx.automation_profile == ctx.automation_profile + + +@then('the plan execution context should raise a ValidationError containing "{text}"') +def step_a_valerr(ctx: Any, text: str) -> None: + assert ctx.error is not None, "Expected error" + assert text in str(ctx.error), f"'{text}' not in '{ctx.error}'" + + +@then("the plan execution context should raise a ValidationError") +def step_a_valerr2(ctx: Any) -> None: + assert ctx.error is not None + + +@then("the context changeset_store should be an InMemoryChangeSetStore") +def step_a_default_store(ctx: Any) -> None: + assert isinstance(ctx.exec_ctx.changeset_store, InMemoryChangeSetStore) + + +@then("the context changeset_store should be the provided store") +def step_a_custom_store(ctx: Any) -> None: + assert ctx.exec_ctx.changeset_store is ctx.custom_store + + +@then("the context project_resources should be an empty dict") +def step_a_empty_res(ctx: Any) -> None: + assert ctx.exec_ctx.project_resources == {} + + +@then("the context resource_bindings should be an empty dict") +def step_a_empty_bind(ctx: Any) -> None: + assert ctx.exec_ctx.resource_bindings == {} + + +# -- Changeset operations -- + + +@given("a PlanExecutionContext") +def step_g_ctx(ctx: Any) -> None: + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.changeset_store) + + +@given('a PlanExecutionContext with sandbox_root "{path}"') +def step_g_ctx_sandbox(ctx: Any, path: str) -> None: + ctx.exec_ctx = _make_ctx( + ctx.plan_id, + sandbox_root=path, + changeset_store=ctx.changeset_store, + ) + + +@given("a PlanExecutionContext with an active changeset") +def step_g_ctx_cs(ctx: Any) -> None: + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.changeset_store) + ctx.active_cs_id = ctx.exec_ctx.start_changeset() + + +@given("a PlanExecutionContext without an active changeset") +def step_g_ctx_no_cs(ctx: Any) -> None: + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.changeset_store) + + +@given("a PlanExecutionContext with an active changeset and one recorded entry") +def step_g_ctx_entry(ctx: Any) -> None: + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.changeset_store) + ctx.active_cs_id = ctx.exec_ctx.start_changeset() + ctx.test_entry = _entry(ctx.plan_id) + ctx.exec_ctx.record_change(ctx.test_entry) + + +@when("I call start_changeset") +def step_w_start_cs(ctx: Any) -> None: + ctx.changeset_id_result = ctx.exec_ctx.start_changeset() + + +@when("I record a ChangeEntry into the context") +def step_w_record(ctx: Any) -> None: + ctx.test_entry = _entry(ctx.plan_id, path="src/new.py") + ctx.exec_ctx.record_change(ctx.test_entry) + + +@when("I attempt to record a ChangeEntry") +def step_w_record_fail(ctx: Any) -> None: + try: + ctx.exec_ctx.record_change(_entry(ctx.plan_id)) + ctx.error = None + except (PlanError, ValidationError, KeyError) as e: + ctx.error = e + + +@when("I record three ChangeEntry instances") +def step_w_record3(ctx: Any) -> None: + for i in range(3): + ctx.exec_ctx.record_change(_entry(ctx.plan_id, path=f"src/f{i}.py")) + + +@when("I call get_changeset with the active changeset_id") +def step_w_get_cs(ctx: Any) -> None: + ctx.retrieved_cs = ctx.exec_ctx.get_changeset(ctx.active_cs_id) + + +@when("I call get_changeset with a nonexistent ID") +def step_w_get_cs_miss(ctx: Any) -> None: + ctx.retrieved_cs = ctx.exec_ctx.get_changeset("nonexistent-id") + + +@then("the returned changeset_id should be a valid ULID string") +def step_a_ulid(ctx: Any) -> None: + cid = ctx.changeset_id_result + assert cid and len(cid) == 26 + + +@then("the active_changeset_ids should contain the new ID") +def step_a_active_ids(ctx: Any) -> None: + assert ctx.changeset_id_result in ctx.exec_ctx.active_changeset_ids + + +@then("the changeset should contain the recorded entry") +def step_a_entry_in(ctx: Any) -> None: + cs = ctx.exec_ctx.get_changeset(ctx.active_cs_id) + assert cs is not None + assert any(e.entry_id == ctx.test_entry.entry_id for e in cs.entries) + + +@then("a PlanError should be raised about no active changeset") +def step_a_plan_err(ctx: Any) -> None: + assert ctx.error is not None, "Expected error about no changeset" + + +@then("the changeset should contain exactly three entries") +def step_a_three(ctx: Any) -> None: + cs = ctx.exec_ctx.get_changeset(ctx.active_cs_id) + assert cs and len(cs.entries) == 3 + + +@then("the returned changeset should not be None") +def step_a_not_none(ctx: Any) -> None: + assert ctx.retrieved_cs is not None + + +@then("the changeset plan_id should match") +def step_a_cs_pid(ctx: Any) -> None: + assert ctx.retrieved_cs.plan_id == ctx.plan_id + + +@then("the returned changeset should be None") +def step_a_none(ctx: Any) -> None: + assert ctx.retrieved_cs is None + + +# -- Summarize -- + + +@when("I call summarize on the context") +def step_w_summarize(ctx: Any) -> None: + ctx.summary = ctx.exec_ctx.summarize() + + +@then("the summary should contain plan_id") +def step_a_sum_pid(ctx: Any) -> None: + assert ctx.summary["plan_id"] == ctx.plan_id + + +@then("the summary should contain resource_binding_count of {n:d}") +def step_a_sum_count(ctx: Any, n: int) -> None: + assert ctx.summary["resource_binding_count"] == n + + +@then("the summary should contain decision_root_id") +def step_a_sum_drid(ctx: Any) -> None: + assert "decision_root_id" in ctx.summary + + +@then("the summary resource_binding_count should be {n:d}") +def step_a_sum_count2(ctx: Any, n: int) -> None: + assert ctx.summary["resource_binding_count"] == n + + +# -- RuntimeExecuteResult model -- + + +@when("I create a RuntimeExecuteResult with valid fields") +def step_w_result_ok(ctx: Any) -> None: + ctx.result = RuntimeExecuteResult( + changeset_id=str(ULID()), + tool_call_count=5, + sandbox_refs=["/tmp/sb"], + decision_ids_processed=["d1", "d2"], + execution_duration_ms=42.5, + ) + + +@when("I attempt to create a RuntimeExecuteResult with negative tool_call_count") +def step_w_result_neg(ctx: Any) -> None: + try: + RuntimeExecuteResult( + changeset_id=str(ULID()), + tool_call_count=-1, + ) + ctx.error = None + except (PydanticValidationError, ValidationError, ValueError) as e: + ctx.error = e + + +@when("I attempt to create a RuntimeExecuteResult with empty changeset_id") +def step_w_result_empty_csid(ctx: Any) -> None: + try: + RuntimeExecuteResult( + changeset_id="", + tool_call_count=0, + ) + ctx.error = None + except (PydanticValidationError, ValidationError, ValueError) as e: + ctx.error = e + + +@then("the runtime execute result should have a correct changeset_id") +def step_a_res_csid(ctx: Any) -> None: + assert ctx.result.changeset_id + assert len(ctx.result.changeset_id) == 26 + + +@then("the runtime execute result tool_call_count should be non-negative") +def step_a_res_count(ctx: Any) -> None: + assert ctx.result.tool_call_count >= 0 + + +@then("the runtime result should fail validation") +def step_a_pydantic_err(ctx: Any) -> None: + assert ctx.error is not None + + +# -- RuntimeExecuteActor -- + + +@given("a ToolRunner with an empty registry") +def step_g_runner(ctx: Any) -> None: + ctx.tool_runner = _runner() + + +@given("a RuntimeExecuteActor") +def step_g_actor(ctx: Any) -> None: + ctx.runtime_actor = RuntimeExecuteActor( + tool_runner=ctx.tool_runner, + execution_context=ctx.exec_ctx, + ) + + +@given("a stream callback collector") +def step_g_collector(ctx: Any) -> None: + ctx.stream_events = [] + ctx.stream_callback = lambda t, d: ctx.stream_events.append((t, d)) + + +@when("I execute with a list of two stub decisions") +def step_w_exec2(ctx: Any) -> None: + ctx.result = ctx.runtime_actor.execute( + decisions=_make_decisions(2), + ) + + +@when("I execute with one stub decision and the stream callback") +def step_w_exec1_cb(ctx: Any) -> None: + ctx.result = ctx.runtime_actor.execute( + decisions=_make_decisions(1), + stream_callback=ctx.stream_callback, + ) + + +@when("I execute with one stub decision") +def step_w_exec1(ctx: Any) -> None: + ctx.result = ctx.runtime_actor.execute( + decisions=_make_decisions(1), + ) + + +@when("I attempt to create a RuntimeExecuteActor with None tool_runner") +def step_w_actor_no_run(ctx: Any) -> None: + try: + RuntimeExecuteActor( + tool_runner=None, + execution_context=ctx.exec_ctx, + ) + ctx.error = None + except (ValidationError, TypeError) as e: + ctx.error = e + + +@when("I attempt to create a RuntimeExecuteActor with None execution_context") +def step_w_actor_no_ctx(ctx: Any) -> None: + try: + RuntimeExecuteActor( + tool_runner=ctx.tool_runner, + execution_context=None, + ) + ctx.error = None + except (ValidationError, TypeError) as e: + ctx.error = e + + +@then("the runtime execute result should be a RuntimeExecuteResult") +def step_a_type(ctx: Any) -> None: + assert isinstance(ctx.result, RuntimeExecuteResult) + + +@then("the runtime execute result changeset_id should be a valid ULID") +def step_a_csid_ulid(ctx: Any) -> None: + assert len(ctx.result.changeset_id) == 26 + + +@then("the runtime execute result decision_ids_processed should have two entries") +def step_a_2ids(ctx: Any) -> None: + assert len(ctx.result.decision_ids_processed) == 2 + + +@then("the callback should have received runtime_execute_started event") +def step_a_evt_start(ctx: Any) -> None: + types = [e[0] for e in ctx.stream_events] + assert "runtime_execute_started" in types + + +@then("the callback should have received runtime_execute_step event") +def step_a_evt_step(ctx: Any) -> None: + types = [e[0] for e in ctx.stream_events] + assert "runtime_execute_step" in types + + +@then("the callback should have received runtime_execute_complete event") +def step_a_evt_done(ctx: Any) -> None: + types = [e[0] for e in ctx.stream_events] + assert "runtime_execute_complete" in types + + +@then('the runtime execute result sandbox_refs should contain "{path}"') +def step_a_srefs(ctx: Any, path: str) -> None: + assert ctx.result.sandbox_refs and path in ctx.result.sandbox_refs + + +# -- PlanExecutor integration -- + + +@given("a mock lifecycle service") +def step_g_mock_lc(ctx: Any) -> None: + ctx.lifecycle = _mock_lifecycle() + + +@given("a PlanExecutionContext with a known changeset store") +def step_g_ctx_known(ctx: Any) -> None: + ctx.known_store = InMemoryChangeSetStore() + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.known_store) + + +@when("I create a PlanExecutor with a mock tool_calling_runtime") +def step_w_pe_with_runtime(ctx: Any) -> None: + ctx.executor = PlanExecutor( + lifecycle_service=ctx.lifecycle, + tool_runner=ctx.tool_runner, + execution_context=ctx.exec_ctx, + ) + + +@when("I create a PlanExecutor with the execution_context") +def step_w_pe_ctx(ctx: Any) -> None: + if not hasattr(ctx, "exec_ctx"): + ctx.exec_ctx = _make_ctx(ctx.plan_id, changeset_store=ctx.changeset_store) + ctx.executor = PlanExecutor( + lifecycle_service=ctx.lifecycle, + execution_context=ctx.exec_ctx, + ) + + +@when("I create a PlanExecutor without execution_context") +def step_w_pe_no_ctx(ctx: Any) -> None: + ctx.executor = PlanExecutor(lifecycle_service=ctx.lifecycle) + + +@then("has_runtime should be True") +def step_a_hr_true(ctx: Any) -> None: + assert ctx.executor.has_runtime is True + + +@then("has_runtime should be False") +def step_a_hr_false(ctx: Any) -> None: + assert ctx.executor.has_runtime is False + + +@then("changeset_store should be the same object as the context store") +def step_a_cs_same(ctx: Any) -> None: + assert ctx.executor.changeset_store is ctx.known_store + + +@then("changeset_store should be None") +def step_a_cs_none(ctx: Any) -> None: + assert ctx.executor.changeset_store is None + + +@then("execution_context property should return the context") +def step_a_ec_prop(ctx: Any) -> None: + assert ctx.executor.execution_context is ctx.exec_ctx + + +# -- Error cases -- + + +@when("I attempt to create a PlanExecutor with None lifecycle_service") +def step_w_pe_none_lc(ctx: Any) -> None: + try: + PlanExecutor(lifecycle_service=None) + ctx.error = None + except ValidationError as e: + ctx.error = e + + +@given("a PlanExecutor without execution_context") +def step_g_pe_no_ctx(ctx: Any) -> None: + if not hasattr(ctx, "lifecycle"): + ctx.lifecycle = _mock_lifecycle() + ctx.executor = PlanExecutor(lifecycle_service=ctx.lifecycle) + + +@when("I attempt to run execute with an empty plan_id") +def step_w_exec_empty(ctx: Any) -> None: + try: + ctx.executor.run_execute("") + ctx.error = None + except (ValidationError, PlanError) as e: + ctx.error = e diff --git a/robot/helper_plan_execute_runtime.py b/robot/helper_plan_execute_runtime.py new file mode 100644 index 00000000..1073a0c0 --- /dev/null +++ b/robot/helper_plan_execute_runtime.py @@ -0,0 +1,208 @@ +"""Helper script for plan execute runtime Robot Framework tests. + +Dispatches to individual test functions via a COMMANDS dict. +""" + +from __future__ import annotations + +import sys + +from ulid import ULID + +from cleveragents.application.services.plan_execution_context import ( + PlanExecutionContext, + RuntimeExecuteActor, + RuntimeExecuteResult, +) +from cleveragents.application.services.plan_executor import ( + PlanExecutor, + StrategyDecision, +) +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + InMemoryChangeSetStore, +) +from cleveragents.tool.registry import ToolRegistry +from cleveragents.tool.runner import ToolRunner + + +def _make_runner() -> ToolRunner: + return ToolRunner(registry=ToolRegistry()) + + +def _make_decisions(count: int) -> list[StrategyDecision]: + root_id = str(ULID()) + return [ + StrategyDecision( + decision_id=root_id if i == 0 else str(ULID()), + step_text=f"Step {i + 1}", + sequence=i, + parent_id=root_id if i > 0 else None, + ) + for i in range(count) + ] + + +def _context_creation() -> None: + """Test PlanExecutionContext creation and defaults.""" + plan_id = str(ULID()) + ctx = PlanExecutionContext(plan_id=plan_id) + assert ctx.plan_id == plan_id + assert ctx.decision_root_id is None + assert ctx.sandbox_root is None + assert ctx.automation_profile is None + assert ctx.project_resources == {} + assert ctx.resource_bindings == {} + assert isinstance(ctx.changeset_store, InMemoryChangeSetStore) + + # With all fields + store = InMemoryChangeSetStore() + ctx2 = PlanExecutionContext( + plan_id=plan_id, + decision_root_id="root-123", + sandbox_root="/tmp/sb", + automation_profile="trusted", + project_resources={"r": "v"}, + changeset_store=store, + ) + assert ctx2.decision_root_id == "root-123" + assert ctx2.sandbox_root == "/tmp/sb" + assert ctx2.automation_profile == "trusted" + assert ctx2.changeset_store is store + + # Empty plan_id should fail + try: + PlanExecutionContext(plan_id="") + raise AssertionError("Expected ValidationError") + except ValidationError: + pass + + print("context-creation-ok") + + +def _changeset_ops() -> None: + """Test changeset start, record, get, summarize.""" + plan_id = str(ULID()) + store = InMemoryChangeSetStore() + ctx = PlanExecutionContext(plan_id=plan_id, changeset_store=store) + + cs_id = ctx.start_changeset() + assert len(cs_id) == 26 # ULID length + assert cs_id in ctx.active_changeset_ids + + entry = ChangeEntry( + plan_id=plan_id, + resource_id=str(ULID()), + tool_name="test/tool", + operation=ChangeOperation.MODIFY, + path="src/test.py", + ) + ctx.record_change(entry) + + cs = ctx.get_changeset(cs_id) + assert cs is not None + assert len(cs.entries) == 1 + assert cs.entries[0].entry_id == entry.entry_id + + # Unknown ID + assert ctx.get_changeset("nonexistent") is None + + # Summarize + summary = ctx.summarize() + assert summary["plan_id"] == plan_id + assert summary["active_changeset_count"] == 1 + assert len(summary["changeset_summaries"]) == 1 + + print("changeset-ops-ok") + + +def _runtime_actor() -> None: + """Test RuntimeExecuteActor execution.""" + plan_id = str(ULID()) + store = InMemoryChangeSetStore() + ctx = PlanExecutionContext(plan_id=plan_id, changeset_store=store) + runner = _make_runner() + actor = RuntimeExecuteActor(tool_runner=runner, execution_context=ctx) + + decisions = _make_decisions(2) + result = actor.execute(decisions=decisions) + + assert isinstance(result, RuntimeExecuteResult) + assert len(result.changeset_id) == 26 + assert len(result.decision_ids_processed) == 2 + assert result.execution_duration_ms >= 0.0 + + # With stream callback + events: list[tuple[str, dict]] = [] + actor2 = RuntimeExecuteActor(tool_runner=runner, execution_context=ctx) + actor2.execute( + decisions=_make_decisions(1), + stream_callback=lambda t, d: events.append((t, d)), + ) + types = [e[0] for e in events] + assert "runtime_execute_started" in types + assert "runtime_execute_complete" in types + + print("runtime-actor-ok") + + +def _executor_runtime() -> None: + """Test PlanExecutor with execution context (runtime mode).""" + plan_id = str(ULID()) + store = InMemoryChangeSetStore() + ctx = PlanExecutionContext(plan_id=plan_id, changeset_store=store) + + from unittest.mock import MagicMock + + lifecycle = MagicMock() + + executor = PlanExecutor( + lifecycle_service=lifecycle, + tool_runner=_make_runner(), + execution_context=ctx, + ) + assert executor.has_runtime is True + assert executor.changeset_store is store + assert executor.execution_context is ctx + + print("executor-runtime-ok") + + +def _executor_stub() -> None: + """Test PlanExecutor without execution context (stub mode).""" + from unittest.mock import MagicMock + + lifecycle = MagicMock() + + executor = PlanExecutor(lifecycle_service=lifecycle) + assert executor.has_runtime is False + assert executor.changeset_store is None + assert executor.execution_context is None + + print("executor-stub-ok") + + +COMMANDS: dict[str, object] = { + "context-creation": _context_creation, + "changeset-ops": _changeset_ops, + "runtime-actor": _runtime_actor, + "executor-runtime": _executor_runtime, + "executor-stub": _executor_stub, +} + + +def main() -> None: + if len(sys.argv) < 2: + raise SystemExit("Expected command argument") + command = sys.argv[1] + if command not in COMMANDS: + raise SystemExit(f"Unknown command: {command}") + func = COMMANDS[command] + if callable(func): + func() + + +if __name__ == "__main__": + main() diff --git a/robot/plan_execute_runtime.robot b/robot/plan_execute_runtime.robot new file mode 100644 index 00000000..07299dc5 --- /dev/null +++ b/robot/plan_execute_runtime.robot @@ -0,0 +1,46 @@ +*** Settings *** +Documentation Smoke tests for plan execute runtime integration. +... Covers PlanExecutionContext, changeset capture, +... RuntimeExecuteActor, and PlanExecutor runtime mode. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_plan_execute_runtime.py + +*** Test Cases *** +Plan Execution Context Creation + [Documentation] Verify PlanExecutionContext creates with valid plan_id and defaults + [Tags] plan execute runtime context + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} context-creation cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} context-creation-ok + +Changeset Start And Record + [Documentation] Verify changeset start_changeset and record_change flow + [Tags] plan execute runtime changeset + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} changeset-ops cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} changeset-ops-ok + +Runtime Execute Actor Dispatch + [Documentation] Verify RuntimeExecuteActor dispatches decisions and produces result + [Tags] plan execute runtime actor + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} runtime-actor cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} runtime-actor-ok + +PlanExecutor Runtime Mode + [Documentation] Verify PlanExecutor has_runtime and changeset_store when context provided + [Tags] plan execute runtime executor + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} executor-runtime cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} executor-runtime-ok + +PlanExecutor Stub Fallback + [Documentation] Verify PlanExecutor falls back to stub when no context + [Tags] plan execute runtime executor stub + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} executor-stub cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} executor-stub-ok diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 65cdd126..b61db4cc 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -3,6 +3,11 @@ Contains service classes that orchestrate business operations. """ +from cleveragents.application.services.plan_execution_context import ( + PlanExecutionContext, + RuntimeExecuteActor, + RuntimeExecuteResult, +) from cleveragents.application.services.session_service import ( PersistentSessionService, ) @@ -15,6 +20,9 @@ from cleveragents.application.services.tool_registry_service import ( __all__ = [ "PersistentSessionService", + "PlanExecutionContext", + "RuntimeExecuteActor", + "RuntimeExecuteResult", "SkillRegistryService", "ToolRegistryService", ] diff --git a/src/cleveragents/application/services/plan_execution_context.py b/src/cleveragents/application/services/plan_execution_context.py new file mode 100644 index 00000000..cbeb19e2 --- /dev/null +++ b/src/cleveragents/application/services/plan_execution_context.py @@ -0,0 +1,428 @@ +"""Plan execution context bridging plan metadata into the tool runtime. + +Carries plan_id, decision_root_id, sandbox_root, automation_profile, +project resources, and resource bindings. Integrates with ChangeSetStore +for capturing tool-call mutations during plan execution. + +## Key Classes + +- ``PlanExecutionContext`` -- carries all metadata needed by the execute + phase and delegates changeset operations to a ``ChangeSetStore``. +- ``RuntimeExecuteResult`` -- structured output from the runtime actor. +- ``RuntimeExecuteActor`` -- wraps ``ToolRunner`` to execute strategy + decisions with full changeset capture. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from typing import Any + +import structlog +from pydantic import BaseModel, ConfigDict, Field +from ulid import ULID + +from cleveragents.core.exceptions import PlanError, ValidationError +from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + ChangeSetStore, + InMemoryChangeSetStore, + SpecChangeSet, + ToolInvocation, +) +from cleveragents.tool.context import BoundResource +from cleveragents.tool.runner import ToolRunner + +logger = structlog.get_logger(__name__) + +# Type alias for streaming callbacks +StreamCallback = Callable[[str, dict[str, Any]], None] + + +# --------------------------------------------------------------------------- +# PlanExecutionContext +# --------------------------------------------------------------------------- + + +class PlanExecutionContext: + """Execution context for the plan execute phase. + + Bridges plan metadata (plan_id, decision root, sandbox, automation + profile, resources) into the tool runtime. All changeset operations + are delegated to the injected ``ChangeSetStore``. + + Parameters + ---------- + plan_id: + ULID of the plan being executed. Must be non-empty. + decision_root_id: + ULID of the root decision from the strategize phase. + sandbox_root: + Filesystem path for sandboxed execution. + automation_profile: + Name of the automation profile governing this execution. + project_resources: + Mapping of project resource names to their metadata. + resource_bindings: + Resolved resource bindings keyed by slot name. + changeset_store: + Store implementation for persisting changesets. Defaults + to ``InMemoryChangeSetStore`` when not provided. + """ + + def __init__( + self, + *, + plan_id: str, + decision_root_id: str | None = None, + sandbox_root: str | None = None, + automation_profile: str | None = None, + project_resources: dict[str, Any] | None = None, + resource_bindings: dict[str, BoundResource] | None = None, + changeset_store: ChangeSetStore | None = None, + ) -> None: + if not plan_id: + raise ValidationError("plan_id must not be empty") + + self._plan_id = plan_id + self._decision_root_id = decision_root_id + self._sandbox_root = sandbox_root + self._automation_profile = automation_profile + self._project_resources: dict[str, Any] = ( + project_resources if project_resources is not None else {} + ) + self._resource_bindings: dict[str, BoundResource] = ( + resource_bindings if resource_bindings is not None else {} + ) + self._changeset_store: ChangeSetStore = ( + changeset_store if changeset_store is not None else InMemoryChangeSetStore() + ) + self._active_changeset_ids: list[str] = [] + self._logger = logger.bind( + plan_id=plan_id, + component="plan_execution_context", + ) + self._logger.debug( + "PlanExecutionContext created", + decision_root_id=decision_root_id, + sandbox_root=sandbox_root, + automation_profile=automation_profile, + resource_count=len(self._project_resources), + binding_count=len(self._resource_bindings), + ) + + # -- Properties ---------------------------------------------------------- + + @property + def plan_id(self) -> str: + """The plan ULID.""" + return self._plan_id + + @property + def decision_root_id(self) -> str | None: + """Root decision node ULID from strategize.""" + return self._decision_root_id + + @decision_root_id.setter + def decision_root_id(self, value: str | None) -> None: + self._decision_root_id = value + + @property + def sandbox_root(self) -> str | None: + """Sandbox filesystem root.""" + return self._sandbox_root + + @property + def automation_profile(self) -> str | None: + """Automation profile name.""" + return self._automation_profile + + @property + def project_resources(self) -> dict[str, Any]: + """Project resources mapping.""" + return self._project_resources + + @property + def resource_bindings(self) -> dict[str, BoundResource]: + """Resolved resource bindings.""" + return self._resource_bindings + + @property + def changeset_store(self) -> ChangeSetStore: + """The underlying changeset store.""" + return self._changeset_store + + @property + def active_changeset_ids(self) -> list[str]: + """IDs of changesets started in this context.""" + return list(self._active_changeset_ids) + + # -- Changeset operations ------------------------------------------------ + + def start_changeset(self) -> str: + """Start a new changeset for this plan. + + Returns the changeset_id (ULID string). + """ + changeset_id = self._changeset_store.start(self._plan_id) + self._active_changeset_ids.append(changeset_id) + self._logger.info("Changeset started", changeset_id=changeset_id) + return changeset_id + + def record_change(self, entry: ChangeEntry) -> None: + """Record a change entry into the most recent active changeset. + + Raises ``PlanError`` if no changeset has been started. + """ + if not self._active_changeset_ids: + raise PlanError("No active changeset. Call start_changeset() first.") + changeset_id = self._active_changeset_ids[-1] + self._changeset_store.record(changeset_id, entry) + self._logger.debug( + "Change recorded", + changeset_id=changeset_id, + entry_id=entry.entry_id, + operation=entry.operation, + path=entry.path, + ) + + def get_changeset(self, changeset_id: str) -> SpecChangeSet | None: + """Retrieve a changeset by ID. + + Returns ``None`` if the changeset does not exist. + """ + return self._changeset_store.get(changeset_id) + + def summarize(self) -> dict[str, Any]: + """Return a summary of this execution context. + + Includes plan metadata, resource counts, and changeset summaries. + """ + changeset_summaries: list[dict[str, Any]] = [] + for cs_id in self._active_changeset_ids: + summary = self._changeset_store.summarize(cs_id) + if summary: + summary["changeset_id"] = cs_id + changeset_summaries.append(summary) + + return { + "plan_id": self._plan_id, + "decision_root_id": self._decision_root_id, + "sandbox_root": self._sandbox_root, + "automation_profile": self._automation_profile, + "project_resource_count": len(self._project_resources), + "resource_binding_count": len(self._resource_bindings), + "active_changeset_count": len(self._active_changeset_ids), + "active_changeset_ids": list(self._active_changeset_ids), + "changeset_summaries": changeset_summaries, + } + + +# --------------------------------------------------------------------------- +# RuntimeExecuteResult +# --------------------------------------------------------------------------- + + +class RuntimeExecuteResult(BaseModel): + """Structured output from the runtime execute actor. + + Extends the stub ``ExecuteResult`` concept with runtime-specific + metadata such as tool call count, sandbox references, processed + decision IDs, and wall-clock execution duration. + """ + + changeset_id: str = Field( + ..., + min_length=1, + description="ULID of the produced ChangeSet", + ) + tool_call_count: int = Field( + default=0, + ge=0, + description="Number of tool calls executed", + ) + sandbox_refs: list[str] = Field( + default_factory=list, + description="Sandbox reference paths used", + ) + decision_ids_processed: list[str] = Field( + default_factory=list, + description="Decision node IDs that were processed", + ) + execution_duration_ms: float = Field( + default=0.0, + ge=0.0, + description="Wall-clock execution time in milliseconds", + ) + + model_config = ConfigDict( + str_strip_whitespace=True, + validate_assignment=True, + ) + + +# --------------------------------------------------------------------------- +# RuntimeExecuteActor +# --------------------------------------------------------------------------- + + +class RuntimeExecuteActor: + """Execute actor that dispatches decisions through ``ToolRunner``. + + Wraps the tool-calling runtime to execute strategy decisions with + full changeset capture via ``PlanExecutionContext``. + + Parameters + ---------- + tool_runner: + The ``ToolRunner`` for discovering and executing tools. + execution_context: + The ``PlanExecutionContext`` carrying plan metadata and + changeset store. + """ + + def __init__( + self, + *, + tool_runner: ToolRunner, + execution_context: PlanExecutionContext, + ) -> None: + if tool_runner is None: + raise ValidationError("tool_runner must not be None") + if execution_context is None: + raise ValidationError("execution_context must not be None") + + self._tool_runner = tool_runner + self._execution_context = execution_context + self._logger = logger.bind( + plan_id=execution_context.plan_id, + component="runtime_execute_actor", + ) + + @property + def tool_runner(self) -> ToolRunner: + """The underlying tool runner.""" + return self._tool_runner + + @property + def execution_context(self) -> PlanExecutionContext: + """The execution context.""" + return self._execution_context + + def execute( + self, + decisions: list[Any], + stream_callback: StreamCallback | None = None, + ) -> RuntimeExecuteResult: + """Execute a list of strategy decisions through the tool runtime. + + For each decision, discovers available tools and records a stub + invocation into the changeset store. In later milestones, this + will dispatch real tool calls based on the decision content. + + Args: + decisions: Strategy decisions (StrategyDecision instances). + stream_callback: Optional callback for streaming status. + + Returns: + A ``RuntimeExecuteResult`` with changeset and execution data. + """ + start_time = time.monotonic() + plan_id = self._execution_context.plan_id + + if stream_callback is not None: + stream_callback( + "runtime_execute_started", + {"plan_id": plan_id, "decision_count": len(decisions)}, + ) + + # Start a changeset for this execution run + changeset_id = self._execution_context.start_changeset() + + tool_call_count = 0 + decision_ids_processed: list[str] = [] + sandbox_refs: list[str] = [] + + if self._execution_context.sandbox_root is not None: + sandbox_refs.append(self._execution_context.sandbox_root) + + for decision in decisions: + decision_id = getattr(decision, "decision_id", str(ULID())) + step_text = getattr(decision, "step_text", "unknown step") + sequence = getattr(decision, "sequence", 0) + + if stream_callback is not None: + stream_callback( + "runtime_execute_step", + { + "plan_id": plan_id, + "decision_id": decision_id, + "step": step_text, + "sequence": sequence, + }, + ) + + # Discover available tools + available_tools = self._tool_runner.discover() + tool_call_count += len(available_tools) + + # Record a stub invocation entry for each decision + invocation = ToolInvocation( + plan_id=plan_id, + tool_name="stub/execute-step", + arguments={"step_text": step_text, "sequence": sequence}, + result={"status": "stub_executed", "tools_found": len(available_tools)}, + success=True, + duration_ms=0.0, + sandbox_path=self._execution_context.sandbox_root, + ) + + # Record a change entry for traceability + entry = ChangeEntry( + plan_id=plan_id, + resource_id=decision_id, + tool_name="stub/execute-step", + operation=ChangeOperation.MODIFY, + path=f"decisions/{decision_id}", + ) + self._execution_context.record_change(entry) + decision_ids_processed.append(decision_id) + + self._logger.debug( + "Decision processed", + decision_id=decision_id, + step_text=step_text, + invocation_id=invocation.invocation_id, + ) + + elapsed_ms = (time.monotonic() - start_time) * 1000.0 + + if stream_callback is not None: + stream_callback( + "runtime_execute_complete", + { + "plan_id": plan_id, + "changeset_id": changeset_id, + "tool_call_count": tool_call_count, + "decision_count": len(decision_ids_processed), + "duration_ms": elapsed_ms, + }, + ) + + self._logger.info( + "Runtime execution completed", + changeset_id=changeset_id, + tool_call_count=tool_call_count, + decisions_processed=len(decision_ids_processed), + duration_ms=elapsed_ms, + ) + + return RuntimeExecuteResult( + changeset_id=changeset_id, + tool_call_count=tool_call_count, + sandbox_refs=sandbox_refs, + decision_ids_processed=decision_ids_processed, + execution_duration_ms=elapsed_ms, + ) diff --git a/src/cleveragents/application/services/plan_executor.py b/src/cleveragents/application/services/plan_executor.py index 665e70ea..bc7dfdbb 100644 --- a/src/cleveragents/application/services/plan_executor.py +++ b/src/cleveragents/application/services/plan_executor.py @@ -2,26 +2,8 @@ Provides local-only (no LLM) stub actors for M1 that integrate with the ``PlanLifecycleService`` to drive plans through the Strategize and Execute -phases. - -## Strategize Stub Actor - -Accepts plan context and produces a minimal decision tree derived from -the action's ``definition_of_done``. The actor is **read-only**: it -records decisions in plan metadata without modifying any resources. - -## Execute Stub Actor - -Accepts plan context and the decisions produced by the strategize phase. -Routes tool calls through ``ToolRunner`` with ``ChangeSetCapture`` to -record all mutations. Persists ``changeset_id`` and execution metadata -(tool call count, sandbox refs) into plan metadata. - -## Streaming Hooks - -Both actors accept an optional *stream_callback* that is invoked with -interim status messages during processing. This enables the ``--stream`` -CLI flag to emit real-time progress updates. +phases. When a ``PlanExecutionContext`` is provided, the execute phase +delegates to ``RuntimeExecuteActor`` for full changeset capture. """ from __future__ import annotations @@ -35,7 +17,13 @@ import structlog from pydantic import BaseModel, ConfigDict, Field from ulid import ULID +from cleveragents.application.services.plan_execution_context import ( + PlanExecutionContext, + RuntimeExecuteActor, + RuntimeExecuteResult, +) from cleveragents.core.exceptions import PlanError, ValidationError +from cleveragents.domain.models.core.change import ChangeSetStore from cleveragents.domain.models.core.plan import ( PlanInvariant, PlanPhase, @@ -46,7 +34,6 @@ from cleveragents.tool.runner import ToolRunner logger = structlog.get_logger(__name__) -# Type alias for streaming callbacks StreamCallback = Callable[[str, dict[str, Any]], None] @@ -61,30 +48,19 @@ class StrategyDecision(BaseModel): decision_id: str = Field(..., description="ULID for this decision node") step_text: str = Field(..., description="The step description text") sequence: int = Field(..., ge=0, description="Order in the decision tree") - parent_id: str | None = Field(default=None, description="Parent decision node ULID") + parent_id: str | None = Field(default=None, description="Parent decision ULID") - model_config = ConfigDict( - str_strip_whitespace=True, - validate_assignment=True, - ) + model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True) class StrategizeResult(BaseModel): """Output from the Strategize stub actor.""" - decision_root_id: str = Field(..., description="ULID of the root decision node") - decisions: list[StrategyDecision] = Field( - default_factory=list, description="Ordered list of strategy decisions" - ) - invariant_records: list[dict[str, Any]] = Field( - default_factory=list, - description="Records of invariant enforcement (stubbed for D2)", - ) + decision_root_id: str = Field(..., description="ULID of root decision node") + decisions: list[StrategyDecision] = Field(default_factory=list) + invariant_records: list[dict[str, Any]] = Field(default_factory=list) - model_config = ConfigDict( - str_strip_whitespace=True, - validate_assignment=True, - ) + model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True) class ExecuteResult(BaseModel): @@ -92,17 +68,10 @@ class ExecuteResult(BaseModel): changeset_id: str = Field(..., description="ID of the produced ChangeSet") changeset: ChangeSet = Field(..., description="The captured ChangeSet") - tool_calls_count: int = Field( - default=0, ge=0, description="Number of tool calls made" - ) - sandbox_refs: list[str] = Field( - default_factory=list, description="Sandbox reference IDs used" - ) + tool_calls_count: int = Field(default=0, ge=0) + sandbox_refs: list[str] = Field(default_factory=list) - model_config = ConfigDict( - str_strip_whitespace=True, - validate_assignment=True, - ) + model_config = ConfigDict(str_strip_whitespace=True, validate_assignment=True) # --------------------------------------------------------------------------- @@ -111,11 +80,7 @@ class ExecuteResult(BaseModel): class StrategizeStubActor: - """Local-only strategize actor for M1. - - Produces a minimal decision tree from the plan's definition of done. - This is a read-only operation: no resources are modified. - """ + """Local-only strategize actor producing a decision tree from definition_of_done.""" def execute( self, @@ -124,46 +89,26 @@ class StrategizeStubActor: invariants: list[PlanInvariant] | None = None, stream_callback: StreamCallback | None = None, ) -> StrategizeResult: - """Run the strategize stub. - - Args: - plan_id: The plan ULID. - definition_of_done: The action's completion criteria text. - invariants: Plan invariants for enforcement recording. - stream_callback: Optional callback for streaming status updates. - - Returns: - A ``StrategizeResult`` containing the decision tree. - - Raises: - ValidationError: If plan_id is empty. - """ if not plan_id: raise ValidationError("plan_id must not be empty") - if stream_callback is not None: stream_callback( - "strategize_started", - {"plan_id": plan_id, "phase": "strategize"}, + "strategize_started", {"plan_id": plan_id, "phase": "strategize"} ) - # Parse definition_of_done into steps steps = self._parse_steps(definition_of_done or "") - - # Build decision tree root_id = str(ULID()) decisions: list[StrategyDecision] = [] - for idx, step_text in enumerate(steps): - decision = StrategyDecision( - decision_id=str(ULID()) if idx > 0 else root_id, - step_text=step_text, - sequence=idx, - parent_id=root_id if idx > 0 else None, + decisions.append( + StrategyDecision( + decision_id=str(ULID()) if idx > 0 else root_id, + step_text=step_text, + sequence=idx, + parent_id=root_id if idx > 0 else None, + ) ) - decisions.append(decision) - # Record invariant enforcement (stubbed for D2 reconciliation) invariant_records: list[dict[str, Any]] = [] for inv in invariants or []: invariant_records.append( @@ -190,43 +135,28 @@ class StrategizeStubActor: decisions=decisions, invariant_records=invariant_records, ) - if stream_callback is not None: stream_callback( "strategize_complete", {"plan_id": plan_id, "decision_count": len(decisions)}, ) - return result @staticmethod def _parse_steps(definition_of_done: str) -> list[str]: - """Parse definition of done into discrete steps. - - Splits on newlines, strips whitespace, removes empty lines - and common list prefixes (``-``, ``*``, numbered). - - Args: - definition_of_done: Raw definition of done text. - - Returns: - List of step strings. - """ + """Parse definition of done into discrete steps.""" if not definition_of_done.strip(): return ["Complete the plan objectives"] - lines = definition_of_done.strip().splitlines() steps: list[str] = [] for line in lines: cleaned = line.strip() if not cleaned: continue - # Remove common list prefixes for prefix in ("-", "*", "•"): if cleaned.startswith(prefix): cleaned = cleaned[len(prefix) :].strip() break - # Remove numbered prefixes like "1.", "2)" if len(cleaned) > 2 and cleaned[0].isdigit(): rest = cleaned.lstrip("0123456789") if rest and rest[0] in (".", ")"): @@ -237,12 +167,7 @@ class StrategizeStubActor: class ExecuteStubActor: - """Local-only execute actor for M1. - - Accepts decisions from the strategize phase and executes them using - sandbox resources with tool calls routed through ``ToolRunner`` and - ``ChangeSetCapture``. - """ + """Local-only execute actor routing tool calls through ChangeSetCapture.""" def execute( self, @@ -252,46 +177,20 @@ class ExecuteStubActor: sandbox_root: str | None = None, stream_callback: StreamCallback | None = None, ) -> ExecuteResult: - """Run the execute stub. - - Args: - plan_id: The plan ULID. - decisions: Strategy decisions from the strategize phase. - tool_runner: Optional ToolRunner for executing tool calls. - sandbox_root: Optional sandbox root directory path. - stream_callback: Optional callback for streaming status updates. - - Returns: - An ``ExecuteResult`` with changeset and execution metadata. - - Raises: - ValidationError: If plan_id is empty. - """ if not plan_id: raise ValidationError("plan_id must not be empty") - if stream_callback is not None: - stream_callback( - "execute_started", - {"plan_id": plan_id, "phase": "execute"}, - ) + stream_callback("execute_started", {"plan_id": plan_id, "phase": "execute"}) - # Set up ChangeSet capture changeset_id = str(ULID()) capture = ChangeSetCapture( - plan_id=plan_id, - resource_id=changeset_id, - sandbox_root=sandbox_root, + plan_id=plan_id, resource_id=changeset_id, sandbox_root=sandbox_root ) - tool_calls_count = 0 sandbox_refs: list[str] = [] - if sandbox_root is not None: sandbox_refs.append(sandbox_root) - # In stub mode, we iterate decisions and record them as executed. - # Real LLM-driven execution will be added in later milestones. for decision in decisions: if stream_callback is not None: stream_callback( @@ -303,15 +202,10 @@ class ExecuteStubActor: "sequence": decision.sequence, }, ) - - # If a ToolRunner is provided, discover available tools - # (stub: no actual tool calls in M1, just counting) if tool_runner is not None: - available_tools = tool_runner.discover() - tool_calls_count += len(available_tools) + tool_calls_count += len(tool_runner.discover()) changeset = capture.get_changeset() - if stream_callback is not None: stream_callback( "execute_complete", @@ -321,7 +215,6 @@ class ExecuteStubActor: "tool_calls_count": tool_calls_count, }, ) - return ExecuteResult( changeset_id=changeset_id, changeset=changeset, @@ -338,8 +231,8 @@ class ExecuteStubActor: class PlanExecutor: """Orchestrates strategize and execute phases for a plan. - Bridges the ``PlanLifecycleService`` with the stub actors, handling - phase transitions, error capture, metadata persistence, and streaming. + When ``execution_context`` is provided, execute dispatches through + ``RuntimeExecuteActor``. Otherwise falls back to ``ExecuteStubActor``. """ def __init__( @@ -347,105 +240,84 @@ class PlanExecutor: lifecycle_service: Any, tool_runner: ToolRunner | None = None, sandbox_root: str | None = None, + execution_context: PlanExecutionContext | None = None, ) -> None: - """Initialize the plan executor. - - Args: - lifecycle_service: The ``PlanLifecycleService`` instance. - tool_runner: Optional ``ToolRunner`` for execute phase. - sandbox_root: Optional sandbox root directory. - """ if lifecycle_service is None: raise ValidationError("lifecycle_service must not be None") - self._lifecycle = lifecycle_service self._tool_runner = tool_runner self._sandbox_root = sandbox_root + self._execution_context = execution_context self._strategize_actor = StrategizeStubActor() self._execute_actor = ExecuteStubActor() self._logger = logger.bind(service="plan_executor") + @property + def has_runtime(self) -> bool: + """True if an execution context is configured for runtime mode.""" + return self._execution_context is not None + + @property + def changeset_store(self) -> ChangeSetStore | None: + """Return the changeset store from the execution context, if any.""" + if self._execution_context is not None: + return self._execution_context.changeset_store + return None + + @property + def execution_context(self) -> PlanExecutionContext | None: + """Return the execution context, if configured.""" + return self._execution_context + def run_strategize( self, plan_id: str, stream_callback: StreamCallback | None = None, ) -> StrategizeResult: - """Run the strategize phase for a plan. - - Transitions the plan through QUEUED -> PROCESSING -> COMPLETE, - invoking the strategize stub actor and persisting results. - - Args: - plan_id: The plan ULID. - stream_callback: Optional streaming callback. - - Returns: - The ``StrategizeResult`` from the stub actor. - - Raises: - PlanError: If the plan is not in the correct state. - ValidationError: If plan_id is empty. - """ + """Run the strategize phase for a plan.""" if not plan_id: raise ValidationError("plan_id must not be empty") plan = self._lifecycle.get_plan(plan_id) - - # Guard: must be in Strategize phase if plan.phase != PlanPhase.STRATEGIZE: raise PlanError( f"Plan {plan_id} is not in Strategize phase " f"(current: {plan.phase.value})" ) - # Start strategize self._lifecycle.start_strategize(plan_id) - try: - # Run the strategize stub actor result = self._strategize_actor.execute( plan_id=plan_id, definition_of_done=plan.definition_of_done, invariants=plan.invariants, stream_callback=stream_callback, ) - - # Persist decision tree into plan metadata plan = self._lifecycle.get_plan(plan_id) plan.decision_root_id = result.decision_root_id plan.timestamps.updated_at = datetime.now() - - # Store decisions and invariant records in error_details as - # structured metadata (Plan model uses error_details for - # arbitrary metadata storage until a dedicated field lands) plan.error_details = { "strategy_decisions": str(len(result.decisions)), "invariant_records": str(len(result.invariant_records)), } - self._lifecycle._commit_plan(plan) - - # Complete strategize + if self._execution_context is not None: + self._execution_context.decision_root_id = result.decision_root_id self._lifecycle.complete_strategize(plan_id) - self._logger.info( - "Strategize completed via executor", + "Strategize completed", plan_id=plan_id, decision_count=len(result.decisions), root_id=result.decision_root_id, ) - return result - except Exception as exc: - # Capture error details error_msg = f"{type(exc).__name__}: {exc}" - error_details = { + plan = self._lifecycle.get_plan(plan_id) + plan.error_details = { "exception_type": type(exc).__name__, "traceback": traceback.format_exc(), } - plan = self._lifecycle.get_plan(plan_id) - plan.error_details = error_details self._lifecycle._commit_plan(plan) self._lifecycle.fail_strategize(plan_id, error_msg) raise @@ -454,51 +326,18 @@ class PlanExecutor: self, plan_id: str, stream_callback: StreamCallback | None = None, - ) -> ExecuteResult: - """Run the execute phase for a plan. - - Guards that the plan is in Execute/QUEUED state, then transitions - through PROCESSING -> COMPLETE with full ChangeSet capture. - - Args: - plan_id: The plan ULID. - stream_callback: Optional streaming callback. - - Returns: - The ``ExecuteResult`` from the stub actor. - - Raises: - PlanError: If the plan is not in the correct state. - ValidationError: If plan_id is empty. - """ + ) -> ExecuteResult | RuntimeExecuteResult: + """Run execute phase — runtime mode if context set, else stub.""" if not plan_id: 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) - plan = self._lifecycle.get_plan(plan_id) - - # Guard: must be in Execute phase - if plan.phase != PlanPhase.EXECUTE: - raise PlanError( - f"Plan {plan_id} is not in Execute phase (current: {plan.phase.value})" - ) - - # Guard: must be in QUEUED state - if plan.state != ProcessingState.QUEUED: - raise PlanError( - f"Plan {plan_id} is not queued for execution (current: {plan.state})" - ) - - # Extract decisions from plan metadata (decision_root_id must exist) - if plan.decision_root_id is None: - raise PlanError( - f"Plan {plan_id} has no decision tree. " - "Strategize must complete before Execute." - ) - - # Build stub decisions from definition_of_done for execute - # (In full implementation, these come from the persisted decision tree) + def _build_decisions(self, plan: Any) -> list[StrategyDecision]: + """Build decisions from plan definition_of_done.""" steps = StrategizeStubActor._parse_steps(plan.definition_of_done or "") - decisions = [ + return [ StrategyDecision( decision_id=plan.decision_root_id if idx == 0 else str(ULID()), step_text=step, @@ -508,11 +347,88 @@ class PlanExecutor: for idx, step in enumerate(steps) ] - # Start execute - self._lifecycle.start_execute(plan_id) + def _guard_execute(self, plan_id: str) -> Any: + """Validate execute-phase guards and return the plan.""" + plan = self._lifecycle.get_plan(plan_id) + if plan.phase != PlanPhase.EXECUTE: + raise PlanError( + f"Plan {plan_id} is not in Execute phase (current: {plan.phase.value})" + ) + if plan.state != ProcessingState.QUEUED: + raise PlanError( + f"Plan {plan_id} is not queued for execution (current: {plan.state})" + ) + if plan.decision_root_id is None: + raise PlanError( + f"Plan {plan_id} has no decision tree. " + "Strategize must complete before Execute." + ) + return plan + def _run_execute_with_runtime( + self, + plan_id: str, + stream_callback: StreamCallback | None = None, + ) -> RuntimeExecuteResult: + """Execute using RuntimeExecuteActor with changeset capture.""" + plan = self._guard_execute(plan_id) + decisions = self._build_decisions(plan) + + if self._tool_runner is None: + raise PlanError("Runtime execute mode requires a ToolRunner.") + + assert self._execution_context is not None # guarded by caller + runtime_actor = RuntimeExecuteActor( + tool_runner=self._tool_runner, + execution_context=self._execution_context, + ) + self._lifecycle.start_execute(plan_id) + try: + result = runtime_actor.execute( + decisions=decisions, stream_callback=stream_callback + ) + 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", + } + plan.timestamps.updated_at = datetime.now() + self._lifecycle._commit_plan(plan) + self._lifecycle.complete_execute(plan_id) + self._logger.info( + "Execute completed (runtime)", + plan_id=plan_id, + changeset_id=result.changeset_id, + tool_calls=result.tool_call_count, + ) + return result + except Exception as exc: + 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", + } + self._lifecycle._commit_plan(plan) + self._lifecycle.fail_execute(plan_id, error_msg) + raise + + def _run_execute_with_stub( + self, + plan_id: str, + stream_callback: StreamCallback | None = None, + ) -> ExecuteResult: + """Execute using the legacy ExecuteStubActor.""" + plan = self._guard_execute(plan_id) + decisions = self._build_decisions(plan) + + self._lifecycle.start_execute(plan_id) try: - # Run the execute stub actor result = self._execute_actor.execute( plan_id=plan_id, decisions=decisions, @@ -520,38 +436,32 @@ class PlanExecutor: sandbox_root=self._sandbox_root, stream_callback=stream_callback, ) - - # Persist execution metadata into plan 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", } plan.timestamps.updated_at = datetime.now() self._lifecycle._commit_plan(plan) - - # Complete execute self._lifecycle.complete_execute(plan_id) - self._logger.info( - "Execute completed via executor", + "Execute completed (stub)", plan_id=plan_id, changeset_id=result.changeset_id, tool_calls=result.tool_calls_count, ) - return result - except Exception as exc: error_msg = f"{type(exc).__name__}: {exc}" - error_details = { + plan = self._lifecycle.get_plan(plan_id) + plan.error_details = { "exception_type": type(exc).__name__, "traceback": traceback.format_exc(), + "mode": "stub", } - plan = self._lifecycle.get_plan(plan_id) - plan.error_details = error_details self._lifecycle._commit_plan(plan) self._lifecycle.fail_execute(plan_id, error_msg) raise