forked from HAL9000/cleveragents-core
freemo
c9abb45adf
Added 246 new BDD scenarios across 9 feature files to improve unit test coverage for modules that were either entirely untested or had significant coverage gaps: - lock_service_coverage.feature (27 scenarios): validation branches, TTL boundaries, re-entrant acquisition, rollback on exceptions - plan_apply_service_coverage.feature (54 scenarios): operation labels, diff rendering (plain/rich/json), artifact building, validation gate, changeset resolution and cleanup - plan_executor_coverage.feature (51 scenarios): step parsing, execute actor integration, strategize/execute guards, stub retry/recovery, decision tree construction - skill_cli_coverage_r3.feature (22 scenarios): tools refresh, list/show JSON fallback, capability summary errors, remove confirmation - changeset_repository_coverage.feature (39 scenarios): entry/tool repos validation, database error wrapping, domain conversion, SQLite store CRUD operations - repositories_coverage.feature (20 scenarios): get_by_name/namespace errors, list_available filters, delete with ActionInUseError, plan update with invariants/processing_state/error_details - sandbox_copy_on_write_coverage.feature (12 scenarios): create OSError wrapping, get_path state transitions, commit edge cases, rollback errors, cleanup with missing paths - bridge_coverage.feature (8 scenarios): __del__ suppression, async task cancellation, execute_graph message type handling, stream config, state checkpointer - plan_cli_coverage.feature (13 scenarios): legacy apply/list/cd paths, use-action with estimation/invariant actors, lifecycle-apply guards, status errors, error recovery display All 246 scenarios (1105 steps) pass. Step definitions use unique prefixes to prevent ambiguous step conflicts with existing tests. ISSUES CLOSED: #467
1089 lines
41 KiB
Python
1089 lines
41 KiB
Python
"""Step definitions for plan_executor_coverage.feature.
|
|
|
|
Comprehensive coverage of plan_executor.py including _parse_steps
|
|
branches, ExecuteStubActor with tool_runner/sandbox/stream/read_only,
|
|
PlanExecutor runtime execution path, error recovery retry logic,
|
|
and result model edge cases.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.services.plan_executor import (
|
|
ExecuteResult,
|
|
ExecuteStubActor,
|
|
PlanExecutor,
|
|
StrategizeResult,
|
|
StrategizeStubActor,
|
|
StrategyDecision,
|
|
)
|
|
from cleveragents.core.exceptions import PlanError, ValidationError
|
|
from cleveragents.domain.models.core.plan import (
|
|
InvariantSource,
|
|
PlanInvariant,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Constants
|
|
# ----------------------------------------------------------------------
|
|
|
|
COV2_PLAN_ID = "01KCOV2PLANID000000000PLAN"
|
|
COV2_ROOT_ID = "01KCOV2ROOTID000000000ROOT"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Helpers
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def _cov2_make_plan(
|
|
*,
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
state: ProcessingState = ProcessingState.QUEUED,
|
|
definition_of_done: str | None = "Step one\nStep two",
|
|
decision_root_id: str | None = COV2_ROOT_ID,
|
|
invariants: list[PlanInvariant] | None = None,
|
|
read_only: bool = False,
|
|
) -> MagicMock:
|
|
"""Build a mock plan object with sensible defaults."""
|
|
plan = MagicMock()
|
|
plan.phase = phase
|
|
plan.state = state
|
|
plan.definition_of_done = definition_of_done
|
|
plan.decision_root_id = decision_root_id
|
|
plan.invariants = invariants or []
|
|
plan.timestamps = PlanTimestamps()
|
|
plan.changeset_id = None
|
|
plan.sandbox_refs = []
|
|
plan.error_details = None
|
|
plan.read_only = read_only
|
|
return plan
|
|
|
|
|
|
def _cov2_make_lifecycle(plan: Any | None = None) -> MagicMock:
|
|
"""Build a mock lifecycle service."""
|
|
lcs = MagicMock()
|
|
if plan is not None:
|
|
lcs.get_plan.return_value = plan
|
|
lcs.start_strategize = MagicMock()
|
|
lcs.complete_strategize = MagicMock()
|
|
lcs.fail_strategize = MagicMock()
|
|
lcs.start_execute = MagicMock()
|
|
lcs.complete_execute = MagicMock()
|
|
lcs.fail_execute = MagicMock()
|
|
lcs._commit_plan = MagicMock()
|
|
return lcs
|
|
|
|
|
|
def _cov2_make_decisions(count: int = 2) -> list[StrategyDecision]:
|
|
"""Build a list of StrategyDecision instances."""
|
|
root_id = "01KCOV2DEC000000000000DEC0"
|
|
decisions: list[StrategyDecision] = []
|
|
for i in range(count):
|
|
decisions.append(
|
|
StrategyDecision(
|
|
decision_id=root_id if i == 0 else f"01KCOV2DEC00000000000{i:05d}",
|
|
step_text=f"Step {i + 1}",
|
|
sequence=i,
|
|
parent_id=root_id if i > 0 else None,
|
|
)
|
|
)
|
|
return decisions
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# StrategizeStubActor._parse_steps
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh StrategizeStubActor for cov2")
|
|
def step_cov2_given_strategize_actor(context: Context) -> None:
|
|
"""Create a fresh StrategizeStubActor."""
|
|
context.cov2_actor = StrategizeStubActor()
|
|
|
|
|
|
@when('I cov2 parse steps from "{text}"')
|
|
def step_cov2_parse_steps(context: Context, text: str) -> None:
|
|
"""Parse steps from the given text."""
|
|
raw = text.encode("utf-8").decode("unicode_escape")
|
|
context.cov2_parsed = StrategizeStubActor._parse_steps(raw)
|
|
|
|
|
|
@when("I cov2 parse steps from an empty string")
|
|
def step_cov2_parse_steps_empty(context: Context) -> None:
|
|
"""Parse steps from an empty string."""
|
|
context.cov2_parsed = StrategizeStubActor._parse_steps("")
|
|
|
|
|
|
@then('the cov2 parsed steps count should be "{n}"')
|
|
def step_cov2_check_parsed_count(context: Context, n: str) -> None:
|
|
"""Verify parsed steps list has expected count."""
|
|
assert len(context.cov2_parsed) == int(n), (
|
|
f"Expected {n} steps, got {len(context.cov2_parsed)}: {context.cov2_parsed}"
|
|
)
|
|
|
|
|
|
@then('the cov2 parsed step at index "{idx}" should be "{expected}"')
|
|
def step_cov2_check_parsed_at_index(context: Context, idx: str, expected: str) -> None:
|
|
"""Verify a specific parsed step by index."""
|
|
actual = context.cov2_parsed[int(idx)]
|
|
assert actual == expected, f"Expected '{expected}' at index {idx}, got '{actual}'"
|
|
|
|
|
|
@then("the cov2 parsed steps should be the default objective")
|
|
def step_cov2_check_parsed_default(context: Context) -> None:
|
|
"""Verify parsed steps are the default fallback."""
|
|
assert context.cov2_parsed == ["Complete the plan objectives"], (
|
|
f"Expected default, got {context.cov2_parsed}"
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# StrategizeStubActor.execute
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when('I cov2 execute strategize with plan_id "{pid}" and definition "{defn}"')
|
|
def step_cov2_exec_strategize(context: Context, pid: str, defn: str) -> None:
|
|
"""Execute strategize with given plan_id and definition."""
|
|
raw_defn = defn.replace("\\n", "\n")
|
|
try:
|
|
context.cov2_strat_result = context.cov2_actor.execute(
|
|
plan_id=pid, definition_of_done=raw_defn
|
|
)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@when("I cov2 execute strategize with an empty plan_id")
|
|
def step_cov2_exec_strategize_empty_pid(context: Context) -> None:
|
|
"""Execute strategize with empty plan_id."""
|
|
try:
|
|
context.cov2_actor.execute(plan_id="", definition_of_done="x")
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@when('I cov2 execute strategize with plan_id "{pid}" and no definition')
|
|
def step_cov2_exec_strategize_no_defn(context: Context, pid: str) -> None:
|
|
"""Execute strategize with None definition_of_done."""
|
|
context.cov2_strat_result = context.cov2_actor.execute(
|
|
plan_id=pid, definition_of_done=None
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@when(
|
|
'I cov2 execute strategize with plan_id "{pid}" and definition "{defn}" without callback'
|
|
)
|
|
def step_cov2_exec_strategize_no_cb(context: Context, pid: str, defn: str) -> None:
|
|
"""Execute strategize without a stream callback."""
|
|
raw_defn = defn.replace("\\n", "\n")
|
|
context.cov2_strat_result = context.cov2_actor.execute(
|
|
plan_id=pid, definition_of_done=raw_defn, stream_callback=None
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@when("I cov2 execute strategize with invariants")
|
|
def step_cov2_exec_strategize_invariants(context: Context) -> None:
|
|
"""Execute strategize with invariant records."""
|
|
invariants = [
|
|
PlanInvariant(text="Must be safe", source=InvariantSource.PLAN),
|
|
PlanInvariant(text="Must be fast", source=InvariantSource.PROJECT),
|
|
]
|
|
context.cov2_strat_result = context.cov2_actor.execute(
|
|
plan_id=COV2_PLAN_ID, definition_of_done="Do it", invariants=invariants
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@given("a cov2 stream event collector")
|
|
def step_cov2_given_stream_collector(context: Context) -> None:
|
|
"""Create a stream event collector list."""
|
|
context.cov2_stream_events = []
|
|
|
|
|
|
@when("I cov2 execute strategize with stream callback")
|
|
def step_cov2_exec_strategize_with_cb(context: Context) -> None:
|
|
"""Execute strategize with a stream callback."""
|
|
|
|
def _cb(event: str, data: dict[str, Any]) -> None:
|
|
context.cov2_stream_events.append((event, data))
|
|
|
|
context.cov2_strat_result = context.cov2_actor.execute(
|
|
plan_id=COV2_PLAN_ID,
|
|
definition_of_done="A step",
|
|
stream_callback=_cb,
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@then('a cov2 ValidationError should be raised with message containing "{text}"')
|
|
def step_cov2_check_validation_error(context: Context, text: str) -> None:
|
|
"""Verify a ValidationError was raised containing the expected text."""
|
|
assert context.cov2_raised is not None, "Expected an exception but none was raised"
|
|
assert isinstance(context.cov2_raised, ValidationError), (
|
|
f"Expected ValidationError, got {type(context.cov2_raised).__name__}"
|
|
)
|
|
assert text in str(context.cov2_raised), (
|
|
f"Expected '{text}' in '{context.cov2_raised}'"
|
|
)
|
|
|
|
|
|
@then('the cov2 strategize result should have "{n}" decisions')
|
|
def step_cov2_check_decision_count(context: Context, n: str) -> None:
|
|
"""Verify the strategize result decision count."""
|
|
assert len(context.cov2_strat_result.decisions) == int(n), (
|
|
f"Expected {n} decisions, got {len(context.cov2_strat_result.decisions)}"
|
|
)
|
|
|
|
|
|
@then('the cov2 first decision text should be "{text}"')
|
|
def step_cov2_check_first_decision_text(context: Context, text: str) -> None:
|
|
"""Verify the first decision step_text."""
|
|
actual = context.cov2_strat_result.decisions[0].step_text
|
|
assert actual == text, f"Expected '{text}', got '{actual}'"
|
|
|
|
|
|
@then("the cov2 root decision should have no parent")
|
|
def step_cov2_check_root_no_parent(context: Context) -> None:
|
|
"""Verify the root decision has no parent_id."""
|
|
root = context.cov2_strat_result.decisions[0]
|
|
assert root.parent_id is None, f"Expected None parent, got {root.parent_id}"
|
|
|
|
|
|
@then("the cov2 non-root decisions should reference root as parent")
|
|
def step_cov2_check_children_parent(context: Context) -> None:
|
|
"""Verify non-root decisions reference the root decision id."""
|
|
root_id = context.cov2_strat_result.decision_root_id
|
|
for d in context.cov2_strat_result.decisions[1:]:
|
|
assert d.parent_id == root_id, (
|
|
f"Decision {d.decision_id} parent={d.parent_id}, expected {root_id}"
|
|
)
|
|
|
|
|
|
@then('the cov2 strategize result should have "{n}" invariant records')
|
|
def step_cov2_check_invariant_count(context: Context, n: str) -> None:
|
|
"""Verify the number of invariant records."""
|
|
assert len(context.cov2_strat_result.invariant_records) == int(n), (
|
|
f"Expected {n} invariant records, got {len(context.cov2_strat_result.invariant_records)}"
|
|
)
|
|
|
|
|
|
@then("each cov2 invariant record should have enforced set to true")
|
|
def step_cov2_check_invariant_enforced(context: Context) -> None:
|
|
"""Verify each invariant record is enforced."""
|
|
for rec in context.cov2_strat_result.invariant_records:
|
|
assert rec["enforced"] is True, f"Expected enforced=True, got {rec}"
|
|
|
|
|
|
@then('the cov2 stream events should include "{event_name}"')
|
|
def step_cov2_check_stream_event(context: Context, event_name: str) -> None:
|
|
"""Verify a specific event was emitted to the stream collector."""
|
|
names = [e[0] for e in context.cov2_stream_events]
|
|
assert event_name in names, f"Expected '{event_name}' in {names}"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# ExecuteStubActor
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh ExecuteStubActor for cov2")
|
|
def step_cov2_given_execute_actor(context: Context) -> None:
|
|
"""Create a fresh ExecuteStubActor."""
|
|
context.cov2_exec_actor = ExecuteStubActor()
|
|
|
|
|
|
@when("I cov2 execute execute-stub with empty plan_id")
|
|
def step_cov2_exec_execute_empty_pid(context: Context) -> None:
|
|
"""Execute the execute stub with an empty plan_id."""
|
|
try:
|
|
context.cov2_exec_actor.execute(plan_id="", decisions=[])
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@given("a cov2 mock tool runner that discovers 3 tools")
|
|
def step_cov2_given_tool_runner(context: Context) -> None:
|
|
"""Create a mock ToolRunner whose discover() returns 3 items."""
|
|
runner = MagicMock()
|
|
runner.discover.return_value = [MagicMock(), MagicMock(), MagicMock()]
|
|
context.cov2_tool_runner = runner
|
|
|
|
|
|
@when("I cov2 execute execute-stub with tool_runner and 2 decisions")
|
|
def step_cov2_exec_with_tool_runner(context: Context) -> None:
|
|
"""Execute the execute stub with a tool_runner and 2 decisions."""
|
|
decisions = _cov2_make_decisions(2)
|
|
context.cov2_exec_result = context.cov2_exec_actor.execute(
|
|
plan_id=COV2_PLAN_ID,
|
|
decisions=decisions,
|
|
tool_runner=context.cov2_tool_runner,
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@then('the cov2 execute result tool_calls_count should be "{n}"')
|
|
def step_cov2_check_tool_calls(context: Context, n: str) -> None:
|
|
"""Verify the tool_calls_count on the execute result."""
|
|
assert context.cov2_exec_result.tool_calls_count == int(n), (
|
|
f"Expected {n}, got {context.cov2_exec_result.tool_calls_count}"
|
|
)
|
|
|
|
|
|
@when('I cov2 execute execute-stub with sandbox_root "{path}"')
|
|
def step_cov2_exec_with_sandbox(context: Context, path: str) -> None:
|
|
"""Execute the execute stub with a sandbox_root."""
|
|
decisions = _cov2_make_decisions(1)
|
|
context.cov2_exec_result = context.cov2_exec_actor.execute(
|
|
plan_id=COV2_PLAN_ID, decisions=decisions, sandbox_root=path
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@then('the cov2 execute result sandbox_refs should contain "{path}"')
|
|
def step_cov2_check_sandbox_refs(context: Context, path: str) -> None:
|
|
"""Verify sandbox_refs contains the expected path."""
|
|
assert path in context.cov2_exec_result.sandbox_refs, (
|
|
f"Expected '{path}' in {context.cov2_exec_result.sandbox_refs}"
|
|
)
|
|
|
|
|
|
@when("I cov2 execute execute-stub without sandbox_root")
|
|
def step_cov2_exec_no_sandbox(context: Context) -> None:
|
|
"""Execute the execute stub without sandbox_root."""
|
|
decisions = _cov2_make_decisions(1)
|
|
context.cov2_exec_result = context.cov2_exec_actor.execute(
|
|
plan_id=COV2_PLAN_ID, decisions=decisions, sandbox_root=None
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@then("the cov2 execute result sandbox_refs should be empty")
|
|
def step_cov2_check_sandbox_empty(context: Context) -> None:
|
|
"""Verify sandbox_refs is empty."""
|
|
assert context.cov2_exec_result.sandbox_refs == [], (
|
|
f"Expected empty, got {context.cov2_exec_result.sandbox_refs}"
|
|
)
|
|
|
|
|
|
@when("I cov2 execute execute-stub with 2 decisions and stream callback")
|
|
def step_cov2_exec_with_stream(context: Context) -> None:
|
|
"""Execute the execute stub with stream callback and 2 decisions."""
|
|
decisions = _cov2_make_decisions(2)
|
|
|
|
def _cb(event: str, data: dict[str, Any]) -> None:
|
|
context.cov2_stream_events.append((event, data))
|
|
|
|
context.cov2_exec_result = context.cov2_exec_actor.execute(
|
|
plan_id=COV2_PLAN_ID, decisions=decisions, stream_callback=_cb
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@when("I cov2 execute execute-stub with read_only true")
|
|
def step_cov2_exec_read_only(context: Context) -> None:
|
|
"""Execute the execute stub with read_only=True."""
|
|
decisions = _cov2_make_decisions(1)
|
|
context.cov2_exec_result = context.cov2_exec_actor.execute(
|
|
plan_id=COV2_PLAN_ID, decisions=decisions, read_only=True
|
|
)
|
|
context.cov2_raised = None
|
|
|
|
|
|
@then("the cov2 execute result should have a valid changeset_id")
|
|
def step_cov2_check_exec_changeset_id(context: Context) -> None:
|
|
"""Verify the execute result has a non-empty changeset_id."""
|
|
assert context.cov2_exec_result.changeset_id, (
|
|
f"Expected non-empty changeset_id, got '{context.cov2_exec_result.changeset_id}'"
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor.__init__
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when("I cov2 construct PlanExecutor with None lifecycle")
|
|
def step_cov2_construct_none_lifecycle(context: Context) -> None:
|
|
"""Attempt to construct PlanExecutor with None lifecycle."""
|
|
try:
|
|
PlanExecutor(lifecycle_service=None)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@given("a cov2 mock lifecycle service")
|
|
def step_cov2_given_lifecycle(context: Context) -> None:
|
|
"""Create a mock lifecycle service."""
|
|
context.cov2_lifecycle = _cov2_make_lifecycle()
|
|
context.cov2_plan_id = COV2_PLAN_ID
|
|
|
|
|
|
@when("I cov2 construct PlanExecutor with that lifecycle")
|
|
def step_cov2_construct_with_lifecycle(context: Context) -> None:
|
|
"""Construct PlanExecutor with the mock lifecycle."""
|
|
try:
|
|
context.cov2_executor = PlanExecutor(lifecycle_service=context.cov2_lifecycle)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@then("the cov2 PlanExecutor should be initialised without error")
|
|
def step_cov2_check_init_ok(context: Context) -> None:
|
|
"""Verify no error on construction."""
|
|
assert context.cov2_raised is None, f"Unexpected error: {context.cov2_raised}"
|
|
assert context.cov2_executor is not None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor properties
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@given("a cov2 PlanExecutor without execution context")
|
|
def step_cov2_given_executor_no_ctx(context: Context) -> None:
|
|
"""Create a PlanExecutor without execution context."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle, execution_context=None
|
|
)
|
|
|
|
|
|
@given("a cov2 mock execution context")
|
|
def step_cov2_given_exec_ctx(context: Context) -> None:
|
|
"""Create a mock PlanExecutionContext."""
|
|
from cleveragents.application.services.plan_execution_context import (
|
|
PlanExecutionContext,
|
|
)
|
|
|
|
context.cov2_exec_ctx = PlanExecutionContext(plan_id=COV2_PLAN_ID)
|
|
|
|
|
|
@given("a cov2 PlanExecutor with execution context")
|
|
def step_cov2_given_executor_with_ctx(context: Context) -> None:
|
|
"""Create a PlanExecutor with execution context."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=context.cov2_exec_ctx,
|
|
)
|
|
|
|
|
|
@given("a cov2 mock execution context with changeset store")
|
|
def step_cov2_given_exec_ctx_store(context: Context) -> None:
|
|
"""Create a mock execution context with a changeset store."""
|
|
context.cov2_mock_store = MagicMock()
|
|
context.cov2_exec_ctx = MagicMock()
|
|
context.cov2_exec_ctx.changeset_store = context.cov2_mock_store
|
|
context.cov2_exec_ctx.decision_root_id = None
|
|
|
|
|
|
@given("a cov2 PlanExecutor with that execution context")
|
|
def step_cov2_given_executor_with_mock_ctx(context: Context) -> None:
|
|
"""Create a PlanExecutor with the mock execution context."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=context.cov2_exec_ctx,
|
|
)
|
|
|
|
|
|
@then('the cov2 executor has_runtime should be "{val}"')
|
|
def step_cov2_check_has_runtime(context: Context, val: str) -> None:
|
|
"""Verify has_runtime property."""
|
|
expected = val == "True"
|
|
assert context.cov2_executor.has_runtime is expected, (
|
|
f"Expected has_runtime={expected}, got {context.cov2_executor.has_runtime}"
|
|
)
|
|
|
|
|
|
@then("the cov2 executor changeset_store should be None")
|
|
def step_cov2_check_changeset_store_none(context: Context) -> None:
|
|
"""Verify changeset_store is None."""
|
|
assert context.cov2_executor.changeset_store is None
|
|
|
|
|
|
@then("the cov2 executor changeset_store should be the mock store")
|
|
def step_cov2_check_changeset_store_mock(context: Context) -> None:
|
|
"""Verify changeset_store is the mock store."""
|
|
assert context.cov2_executor.changeset_store is context.cov2_mock_store
|
|
|
|
|
|
@then("the cov2 executor execution_context should be None")
|
|
def step_cov2_check_exec_ctx_none(context: Context) -> None:
|
|
"""Verify execution_context is None."""
|
|
assert context.cov2_executor.execution_context is None
|
|
|
|
|
|
@then("the cov2 executor execution_context should not be None")
|
|
def step_cov2_check_exec_ctx_not_none(context: Context) -> None:
|
|
"""Verify execution_context is not None."""
|
|
assert context.cov2_executor.execution_context is not None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor.run_strategize - guards and happy path
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when("I cov2 call run_strategize with empty plan_id")
|
|
def step_cov2_run_strategize_empty(context: Context) -> None:
|
|
"""Call run_strategize with empty plan_id."""
|
|
try:
|
|
context.cov2_executor.run_strategize("")
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@given("a cov2 plan in Execute phase")
|
|
def step_cov2_given_plan_execute(context: Context) -> None:
|
|
"""Set up a plan in Execute phase."""
|
|
plan = _cov2_make_plan(phase=PlanPhase.EXECUTE)
|
|
context.cov2_lifecycle.get_plan.return_value = plan
|
|
context.cov2_mock_plan = plan
|
|
|
|
|
|
@given('a cov2 plan in Strategize phase with definition "{defn}"')
|
|
def step_cov2_given_plan_strategize(context: Context, defn: str) -> None:
|
|
"""Set up a plan in Strategize phase with given definition."""
|
|
raw = defn.replace("\\n", "\n")
|
|
plan = _cov2_make_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
definition_of_done=raw,
|
|
)
|
|
context.cov2_lifecycle.get_plan.return_value = plan
|
|
context.cov2_mock_plan = plan
|
|
|
|
|
|
@when("I cov2 call run_strategize")
|
|
def step_cov2_run_strategize(context: Context) -> None:
|
|
"""Call run_strategize."""
|
|
try:
|
|
context.cov2_strat_run_result = context.cov2_executor.run_strategize(
|
|
context.cov2_plan_id
|
|
)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@then('a cov2 PlanError should be raised containing "{text}"')
|
|
def step_cov2_check_plan_error(context: Context, text: str) -> None:
|
|
"""Verify a PlanError was raised containing the expected text."""
|
|
assert context.cov2_raised is not None, "Expected PlanError but none was raised"
|
|
assert isinstance(context.cov2_raised, PlanError), (
|
|
f"Expected PlanError, got {type(context.cov2_raised).__name__}: {context.cov2_raised}"
|
|
)
|
|
assert text in str(context.cov2_raised), (
|
|
f"Expected '{text}' in '{context.cov2_raised}'"
|
|
)
|
|
|
|
|
|
@then("the cov2 strategize run result should be a StrategizeResult")
|
|
def step_cov2_check_strat_result_type(context: Context) -> None:
|
|
"""Verify the result is a StrategizeResult."""
|
|
assert isinstance(context.cov2_strat_run_result, StrategizeResult), (
|
|
f"Expected StrategizeResult, got {type(context.cov2_strat_run_result).__name__}"
|
|
)
|
|
|
|
|
|
@then("the cov2 lifecycle should have called start_strategize")
|
|
def step_cov2_check_start_strategize(context: Context) -> None:
|
|
"""Verify start_strategize was called."""
|
|
context.cov2_lifecycle.start_strategize.assert_called_once_with(
|
|
context.cov2_plan_id
|
|
)
|
|
|
|
|
|
@then("the cov2 lifecycle should have called complete_strategize")
|
|
def step_cov2_check_complete_strategize(context: Context) -> None:
|
|
"""Verify complete_strategize was called."""
|
|
context.cov2_lifecycle.complete_strategize.assert_called_once_with(
|
|
context.cov2_plan_id
|
|
)
|
|
|
|
|
|
@then("the cov2 lifecycle should have called _commit_plan")
|
|
def step_cov2_check_commit_plan(context: Context) -> None:
|
|
"""Verify _commit_plan was called."""
|
|
assert context.cov2_lifecycle._commit_plan.called
|
|
|
|
|
|
@then("the cov2 execution context decision_root_id should be set")
|
|
def step_cov2_check_ctx_root_id(context: Context) -> None:
|
|
"""Verify the execution context decision_root_id was set."""
|
|
assert context.cov2_raised is None, f"Unexpected error: {context.cov2_raised}"
|
|
assert context.cov2_exec_ctx.decision_root_id is not None, (
|
|
"Expected decision_root_id to be set"
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor.run_strategize - exception path
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@given("a cov2 PlanExecutor with failing strategize actor")
|
|
def step_cov2_given_executor_failing_strat(context: Context) -> None:
|
|
"""Create a PlanExecutor whose strategize actor raises."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle, execution_context=None
|
|
)
|
|
context.cov2_executor._strategize_actor = MagicMock()
|
|
context.cov2_executor._strategize_actor.execute.side_effect = RuntimeError(
|
|
"cov2 boom in strategize"
|
|
)
|
|
|
|
|
|
@given("a cov2 mock error recovery service")
|
|
def step_cov2_given_error_recovery(context: Context) -> None:
|
|
"""Create a mock error recovery service."""
|
|
context.cov2_error_recovery = MagicMock()
|
|
context.cov2_error_recovery.max_retries = 1
|
|
context.cov2_error_recovery.record_error = MagicMock()
|
|
context.cov2_error_recovery.should_retry = MagicMock(return_value=False)
|
|
|
|
|
|
@given("a cov2 PlanExecutor with error recovery and failing strategize actor")
|
|
def step_cov2_given_executor_recovery_failing_strat(context: Context) -> None:
|
|
"""Create a PlanExecutor with error recovery and a failing strategize actor."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=None,
|
|
error_recovery_service=context.cov2_error_recovery,
|
|
)
|
|
context.cov2_executor._strategize_actor = MagicMock()
|
|
context.cov2_executor._strategize_actor.execute.side_effect = RuntimeError(
|
|
"cov2 boom in strategize"
|
|
)
|
|
|
|
|
|
@when("I cov2 call run_strategize expecting exception")
|
|
def step_cov2_run_strategize_expect_exc(context: Context) -> None:
|
|
"""Call run_strategize expecting an exception."""
|
|
try:
|
|
context.cov2_executor.run_strategize(context.cov2_plan_id)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@then("a cov2 RuntimeError should have been raised")
|
|
def step_cov2_check_runtime_error(context: Context) -> None:
|
|
"""Verify a RuntimeError was raised."""
|
|
assert context.cov2_raised is not None, "Expected an exception"
|
|
assert isinstance(context.cov2_raised, RuntimeError), (
|
|
f"Expected RuntimeError, got {type(context.cov2_raised).__name__}"
|
|
)
|
|
|
|
|
|
@then("the cov2 lifecycle should have called fail_strategize")
|
|
def step_cov2_check_fail_strategize(context: Context) -> None:
|
|
"""Verify fail_strategize was called."""
|
|
context.cov2_lifecycle.fail_strategize.assert_called_once()
|
|
args = context.cov2_lifecycle.fail_strategize.call_args[0]
|
|
assert context.cov2_plan_id in args
|
|
|
|
|
|
@then("the cov2 plan error_details should contain exception_type")
|
|
def step_cov2_check_error_details(context: Context) -> None:
|
|
"""Verify error_details was set with exception_type."""
|
|
plan = context.cov2_lifecycle.get_plan.return_value
|
|
assert plan.error_details is not None
|
|
if isinstance(plan.error_details, dict):
|
|
assert "exception_type" in plan.error_details
|
|
|
|
|
|
@then("the cov2 error recovery should have recorded a strategize error")
|
|
def step_cov2_check_recovery_strategize(context: Context) -> None:
|
|
"""Verify error_recovery.record_error was called for strategize."""
|
|
context.cov2_error_recovery.record_error.assert_called_once()
|
|
call_kwargs = context.cov2_error_recovery.record_error.call_args
|
|
assert call_kwargs[1].get(
|
|
"phase", call_kwargs[0][1] if len(call_kwargs[0]) > 1 else None
|
|
) == "strategize" or "strategize" in str(call_kwargs)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor._guard_execute
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when("I cov2 call guard_execute")
|
|
def step_cov2_call_guard_execute(context: Context) -> None:
|
|
"""Call _guard_execute."""
|
|
try:
|
|
context.cov2_executor._guard_execute(context.cov2_plan_id)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@given("a cov2 plan in Execute phase but Processing state")
|
|
def step_cov2_given_plan_exec_processing(context: Context) -> None:
|
|
"""Set up a plan in Execute phase but Processing state."""
|
|
plan = _cov2_make_plan(phase=PlanPhase.EXECUTE, state=ProcessingState.PROCESSING)
|
|
context.cov2_lifecycle.get_plan.return_value = plan
|
|
context.cov2_mock_plan = plan
|
|
|
|
|
|
@given("a cov2 plan in Execute-Queued state without decision root")
|
|
def step_cov2_given_plan_exec_no_root(context: Context) -> None:
|
|
"""Set up a plan in Execute-Queued state without decision_root_id."""
|
|
plan = _cov2_make_plan(
|
|
phase=PlanPhase.EXECUTE,
|
|
state=ProcessingState.QUEUED,
|
|
decision_root_id=None,
|
|
)
|
|
context.cov2_lifecycle.get_plan.return_value = plan
|
|
context.cov2_mock_plan = plan
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor.run_execute - routing and validation
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when("I cov2 call run_execute with empty plan_id")
|
|
def step_cov2_run_execute_empty(context: Context) -> None:
|
|
"""Call run_execute with empty plan_id."""
|
|
try:
|
|
context.cov2_executor.run_execute("")
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@given("a cov2 plan in Execute-Queued state with decision root and definition")
|
|
def step_cov2_given_plan_exec_full(context: Context) -> None:
|
|
"""Set up a fully valid plan for execute phase."""
|
|
plan = _cov2_make_plan(
|
|
phase=PlanPhase.EXECUTE,
|
|
state=ProcessingState.QUEUED,
|
|
definition_of_done="Implement feature\nWrite tests",
|
|
decision_root_id=COV2_ROOT_ID,
|
|
)
|
|
context.cov2_lifecycle.get_plan.return_value = plan
|
|
context.cov2_mock_plan = plan
|
|
|
|
|
|
@when("I cov2 call run_execute")
|
|
def step_cov2_run_execute(context: Context) -> None:
|
|
"""Call run_execute."""
|
|
try:
|
|
context.cov2_exec_run_result = context.cov2_executor.run_execute(
|
|
context.cov2_plan_id
|
|
)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@then("the cov2 execute run result should be an ExecuteResult")
|
|
def step_cov2_check_exec_result_type(context: Context) -> None:
|
|
"""Verify the result is an ExecuteResult."""
|
|
assert context.cov2_raised is None, f"Unexpected error: {context.cov2_raised}"
|
|
assert isinstance(context.cov2_exec_run_result, ExecuteResult), (
|
|
f"Expected ExecuteResult, got {type(context.cov2_exec_run_result).__name__}"
|
|
)
|
|
|
|
|
|
@then("the cov2 lifecycle should have called complete_execute")
|
|
def step_cov2_check_complete_execute(context: Context) -> None:
|
|
"""Verify complete_execute was called."""
|
|
context.cov2_lifecycle.complete_execute.assert_called_once_with(
|
|
context.cov2_plan_id
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor._run_execute_with_runtime
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@given("a cov2 PlanExecutor with execution context and tool runner")
|
|
def step_cov2_given_executor_runtime(context: Context) -> None:
|
|
"""Create a PlanExecutor with execution context and tool runner."""
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
runner = ToolRunner(registry=ToolRegistry())
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=context.cov2_exec_ctx,
|
|
tool_runner=runner,
|
|
)
|
|
|
|
|
|
@then("the cov2 execute run result should be a RuntimeExecuteResult")
|
|
def step_cov2_check_runtime_result_type(context: Context) -> None:
|
|
"""Verify the result is a RuntimeExecuteResult."""
|
|
from cleveragents.application.services.plan_execution_context import (
|
|
RuntimeExecuteResult,
|
|
)
|
|
|
|
assert context.cov2_raised is None, f"Unexpected error: {context.cov2_raised}"
|
|
assert isinstance(context.cov2_exec_run_result, RuntimeExecuteResult), (
|
|
f"Expected RuntimeExecuteResult, got {type(context.cov2_exec_run_result).__name__}"
|
|
)
|
|
|
|
|
|
@given("a cov2 PlanExecutor with execution context but no tool runner")
|
|
def step_cov2_given_executor_runtime_no_runner(context: Context) -> None:
|
|
"""Create a PlanExecutor with execution context but no tool_runner."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=context.cov2_exec_ctx,
|
|
tool_runner=None,
|
|
)
|
|
|
|
|
|
@when("I cov2 call run_execute expecting exception")
|
|
def step_cov2_run_execute_expect_exc(context: Context) -> None:
|
|
"""Call run_execute expecting an exception."""
|
|
try:
|
|
context.cov2_exec_run_result = context.cov2_executor.run_execute(
|
|
context.cov2_plan_id
|
|
)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@given("a cov2 PlanExecutor with execution context and tool runner and failing runtime")
|
|
def step_cov2_given_executor_runtime_failing(context: Context) -> None:
|
|
"""Create a PlanExecutor with runtime that will fail via patched RuntimeExecuteActor."""
|
|
from cleveragents.application.services.plan_execution_context import (
|
|
RuntimeExecuteActor,
|
|
)
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
runner = ToolRunner(registry=ToolRegistry())
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=context.cov2_exec_ctx,
|
|
tool_runner=runner,
|
|
)
|
|
# Patch RuntimeExecuteActor.execute so the real _run_execute_with_runtime
|
|
# try/except catches the error and calls fail_execute.
|
|
_orig_init = RuntimeExecuteActor.__init__
|
|
|
|
patcher = patch.object(
|
|
RuntimeExecuteActor,
|
|
"execute",
|
|
side_effect=RuntimeError("cov2 boom in runtime execute"),
|
|
)
|
|
patcher.start()
|
|
# Register cleanup
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(patcher.stop)
|
|
|
|
|
|
@then("the cov2 lifecycle should have called fail_execute")
|
|
def step_cov2_check_fail_execute(context: Context) -> None:
|
|
"""Verify fail_execute was called."""
|
|
context.cov2_lifecycle.fail_execute.assert_called()
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor._run_execute_with_stub - retry logic
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@given("a cov2 PlanExecutor with failing execute actor and no recovery")
|
|
def step_cov2_given_executor_failing_stub_no_recovery(context: Context) -> None:
|
|
"""Create a PlanExecutor with a failing execute actor and no recovery."""
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=None,
|
|
error_recovery_service=None,
|
|
)
|
|
context.cov2_executor._execute_actor = MagicMock()
|
|
context.cov2_executor._execute_actor.execute.side_effect = RuntimeError(
|
|
"cov2 boom in stub execute"
|
|
)
|
|
|
|
|
|
@given("a cov2 PlanExecutor with execute actor that fails once then succeeds")
|
|
def step_cov2_given_executor_retry_success(context: Context) -> None:
|
|
"""Create a PlanExecutor with error recovery where first attempt fails, second succeeds."""
|
|
error_recovery = MagicMock()
|
|
error_recovery.max_retries = 2
|
|
error_recovery.record_error = MagicMock()
|
|
error_recovery.should_retry = MagicMock(return_value=True)
|
|
context.cov2_error_recovery = error_recovery
|
|
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=None,
|
|
error_recovery_service=error_recovery,
|
|
)
|
|
|
|
# First call raises, second call succeeds
|
|
real_actor = ExecuteStubActor()
|
|
call_count = {"n": 0}
|
|
original_execute = real_actor.execute
|
|
|
|
def _flaky_execute(**kwargs: Any) -> Any:
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise RuntimeError("cov2 transient error")
|
|
return original_execute(**kwargs)
|
|
|
|
context.cov2_executor._execute_actor = MagicMock()
|
|
context.cov2_executor._execute_actor.execute.side_effect = _flaky_execute
|
|
|
|
|
|
@then("the cov2 error recovery should have recorded an execute error")
|
|
def step_cov2_check_recovery_execute(context: Context) -> None:
|
|
"""Verify error_recovery.record_error was called."""
|
|
context.cov2_error_recovery.record_error.assert_called()
|
|
|
|
|
|
@given("a cov2 PlanExecutor with always-failing execute actor and exhausted retries")
|
|
def step_cov2_given_executor_exhausted_retries(context: Context) -> None:
|
|
"""Create a PlanExecutor with error recovery that exhausts retries."""
|
|
error_recovery = MagicMock()
|
|
error_recovery.max_retries = 1
|
|
# First should_retry True, second False
|
|
error_recovery.should_retry = MagicMock(side_effect=[True, False])
|
|
error_recovery.record_error = MagicMock()
|
|
context.cov2_error_recovery = error_recovery
|
|
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=None,
|
|
error_recovery_service=error_recovery,
|
|
)
|
|
context.cov2_executor._execute_actor = MagicMock()
|
|
context.cov2_executor._execute_actor.execute.side_effect = RuntimeError(
|
|
"cov2 persistent failure"
|
|
)
|
|
|
|
|
|
@given("a cov2 PlanExecutor with failing execute actor and recovery denying retry")
|
|
def step_cov2_given_executor_no_retry(context: Context) -> None:
|
|
"""Create a PlanExecutor with error recovery that denies retry."""
|
|
error_recovery = MagicMock()
|
|
error_recovery.max_retries = 2
|
|
error_recovery.should_retry = MagicMock(return_value=False)
|
|
error_recovery.record_error = MagicMock()
|
|
context.cov2_error_recovery = error_recovery
|
|
|
|
context.cov2_executor = PlanExecutor(
|
|
lifecycle_service=context.cov2_lifecycle,
|
|
execution_context=None,
|
|
error_recovery_service=error_recovery,
|
|
)
|
|
context.cov2_executor._execute_actor = MagicMock()
|
|
context.cov2_executor._execute_actor.execute.side_effect = RuntimeError(
|
|
"cov2 no retry error"
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# PlanExecutor._build_decisions
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when("I cov2 call build_decisions")
|
|
def step_cov2_call_build_decisions(context: Context) -> None:
|
|
"""Call _build_decisions on the mock plan."""
|
|
plan = context.cov2_lifecycle.get_plan(context.cov2_plan_id)
|
|
context.cov2_built_decisions = context.cov2_executor._build_decisions(plan)
|
|
|
|
|
|
@then('the cov2 built decisions should have "{n}" entries')
|
|
def step_cov2_check_built_count(context: Context, n: str) -> None:
|
|
"""Verify the number of built decisions."""
|
|
assert len(context.cov2_built_decisions) == int(n), (
|
|
f"Expected {n} decisions, got {len(context.cov2_built_decisions)}"
|
|
)
|
|
|
|
|
|
@then("the cov2 first built decision should use the plan decision_root_id")
|
|
def step_cov2_check_first_built(context: Context) -> None:
|
|
"""Verify the first decision uses the plan's decision_root_id."""
|
|
first = context.cov2_built_decisions[0]
|
|
assert first.decision_id == COV2_ROOT_ID, (
|
|
f"Expected {COV2_ROOT_ID}, got {first.decision_id}"
|
|
)
|
|
assert first.parent_id is None
|
|
|
|
|
|
@then("the cov2 second built decision should have root as parent")
|
|
def step_cov2_check_second_built(context: Context) -> None:
|
|
"""Verify the second decision has root as parent."""
|
|
second = context.cov2_built_decisions[1]
|
|
assert second.parent_id == COV2_ROOT_ID, (
|
|
f"Expected parent {COV2_ROOT_ID}, got {second.parent_id}"
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Result model edge cases
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
@when("I cov2 create a StrategyDecision with negative sequence")
|
|
def step_cov2_create_bad_decision(context: Context) -> None:
|
|
"""Try to create a StrategyDecision with negative sequence."""
|
|
try:
|
|
StrategyDecision(
|
|
decision_id="01KTEST00000000000000TEST0",
|
|
step_text="Test step",
|
|
sequence=-1,
|
|
)
|
|
context.cov2_raised = None
|
|
except Exception as exc:
|
|
context.cov2_raised = exc
|
|
|
|
|
|
@then("a cov2 validation error should be raised for the model")
|
|
def step_cov2_check_model_validation(context: Context) -> None:
|
|
"""Verify a validation error was raised."""
|
|
assert context.cov2_raised is not None, "Expected a validation error"
|
|
|
|
|
|
@when("I cov2 create an ExecuteResult with minimal fields")
|
|
def step_cov2_create_minimal_exec_result(context: Context) -> None:
|
|
"""Create an ExecuteResult with only required fields."""
|
|
from cleveragents.tool.builtins.changeset import ChangeSet
|
|
|
|
context.cov2_minimal_result = ExecuteResult(
|
|
changeset_id="01KTEST00000000000000CSID0",
|
|
changeset=ChangeSet(plan_id=COV2_PLAN_ID),
|
|
)
|
|
|
|
|
|
@then('the cov2 ExecuteResult tool_calls_count should default to "{n}"')
|
|
def step_cov2_check_default_tool_calls(context: Context, n: str) -> None:
|
|
"""Verify tool_calls_count defaults correctly."""
|
|
assert context.cov2_minimal_result.tool_calls_count == int(n), (
|
|
f"Expected {n}, got {context.cov2_minimal_result.tool_calls_count}"
|
|
)
|
|
|
|
|
|
@then("the cov2 ExecuteResult sandbox_refs should default to empty list")
|
|
def step_cov2_check_default_sandbox(context: Context) -> None:
|
|
"""Verify sandbox_refs defaults to empty list."""
|
|
assert context.cov2_minimal_result.sandbox_refs == [], (
|
|
f"Expected [], got {context.cov2_minimal_result.sandbox_refs}"
|
|
)
|