fix(plan): resolve AmbiguousStep errors in plan correct BDD steps
CI / lint (pull_request) Failing after 0s
CI / typecheck (pull_request) Failing after 0s
CI / quality (pull_request) Failing after 0s
CI / unit_tests (pull_request) Failing after 0s
CI / security (pull_request) Failing after 1s
CI / integration_tests (pull_request) Failing after 0s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / build (pull_request) Failing after 0s
CI / helm (pull_request) Failing after 0s
CI / e2e_tests (pull_request) Failing after 0s
CI / push-validation (pull_request) Failing after 0s
CI / status-check (pull_request) Failing after 0s

Consolidate the four extended @when variants (with guidance, without
--yes, with --yes, with --dry-run) into a single @when step that reads
option flags from context variables set by @given steps. Behave's
registration-time conflict detection uses re.search without end anchors,
so the base mode "{mode}" pattern falsely matched all four longer
variants as prefixes.

Also:
- Add decision ID validation to the @when step so the "decision not
  found" scenario actually raises an error instead of silently passing
- Rename "affected decisions" @then step to avoid pattern collision with
  the identical step already defined in correction_flows_steps.py
- Fix ruff format violations (wrapped long decorator and assertion lines)

ISSUES CLOSED: #9286
This commit is contained in:
2026-06-11 19:14:48 -04:00
parent 33d239fdd2
commit dbdbd4915c
2 changed files with 64 additions and 108 deletions
@@ -17,7 +17,8 @@ Feature: agents plan correct with revert and append correction modes
And the plan status should transition to "correcting" then "active"
Scenario: Append mode adds guidance without re-executing
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "append" guidance "Use FastAPI instead"
Given I have guidance "Use FastAPI instead"
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "append"
Then the correction should succeed
And the guidance should be appended to decision "D2" context
And no decisions should be pruned
@@ -25,13 +26,15 @@ Feature: agents plan correct with revert and append correction modes
And the plan status should remain "active"
Scenario: Revert mode shows confirmation prompt
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "revert" without --yes flag
Given the --yes flag is not set
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "revert"
Then a confirmation prompt should be displayed
And the prompt should warn about pruning decisions "D3,D4"
And the prompt should ask for user confirmation
Scenario: Revert mode skips confirmation with --yes flag
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "revert" with --yes flag
Given the --yes flag is set
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "revert"
Then no confirmation prompt should be displayed
And the correction should proceed immediately
@@ -55,8 +58,9 @@ Feature: agents plan correct with revert and append correction modes
And agents plan tree should show the corrected tree
Scenario: Dry-run mode analyzes impact without executing
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "revert" with --dry-run flag
Given the --dry-run flag is set
When I invoke plan correct with plan_id "plan-001" decision_id "D2" mode "revert"
Then the impact analysis should be displayed
And the affected decisions should be "D2,D3,D4"
And the correction should report affected decisions "D2,D3,D4"
And the risk level should be calculated
And no changes should be made to the plan
@@ -1,4 +1,5 @@
"""Steps for plan_correct_revert_append_modes.feature"""
from behave import given, then, when
from src.cleveragents.application.services.correction_service import CorrectionService
from src.cleveragents.domain.models.core.correction import (
@@ -9,15 +10,16 @@ from src.cleveragents.domain.models.core.correction import (
@given("a plan correction service")
def step_plan_correction_service(context):
"""Initialize a correction service for testing."""
context.correction_service = CorrectionService()
context.plans = {}
context.decision_trees = {}
context.guidance = ""
context.yes_flag = True
context.dry_run = False
@given('a plan with ID "{plan_id}" and root decision "{decision_id}"')
def step_plan_with_id(context, plan_id, decision_id):
"""Create a mock plan with a root decision."""
context.current_plan_id = plan_id
context.current_root_decision = decision_id
context.plans[plan_id] = {
@@ -29,10 +31,7 @@ def step_plan_with_id(context, plan_id, decision_id):
@given('a decision tree with decisions "{tree_spec}"')
def step_decision_tree(context, tree_spec):
"""Create a decision tree from a specification string.
Format: "D1->D2,D3;D2->D4" means D1 has children D2 and D3, D2 has child D4
"""
"""Parse "D1->D2,D3;D2->D4": D1 has children D2, D3; D2 has child D4."""
tree = {}
for edge in tree_spec.split(";"):
parent, children = edge.split("->")
@@ -40,234 +39,187 @@ def step_decision_tree(context, tree_spec):
context.decision_trees[context.current_plan_id] = tree
@when('I invoke plan correct with plan_id "{plan_id}" decision_id "{decision_id}" mode "{mode}"')
def step_invoke_plan_correct_basic(context, plan_id, decision_id, mode):
"""Invoke plan correct with basic parameters."""
context.correction_request = CorrectionRequest(
plan_id=plan_id,
target_decision_id=decision_id,
mode=CorrectionMode(mode),
)
context.last_error = None
try:
# This would normally call the CLI command
# For now, we'll just validate the request
if plan_id not in context.plans:
raise ValueError(f"Plan {plan_id} not found")
context.correction_succeeded = True
except Exception as e:
context.last_error = str(e)
context.correction_succeeded = False
@given('I have guidance "{guidance}"')
def step_set_guidance(context, guidance):
context.guidance = guidance
@when('I invoke plan correct with plan_id "{plan_id}" decision_id "{decision_id}" mode "{mode}" guidance "{guidance}"')
def step_invoke_plan_correct_with_guidance(context, plan_id, decision_id, mode, guidance):
"""Invoke plan correct with guidance."""
@given("the --yes flag is set")
def step_yes_flag_set(context):
context.yes_flag = True
@given("the --yes flag is not set")
def step_yes_flag_not_set(context):
context.yes_flag = False
@given("the --dry-run flag is set")
def step_dry_run_flag_set(context):
context.dry_run = True
@when(
'I invoke plan correct with plan_id "{plan_id}" decision_id "{decision_id}" mode "{mode}"'
)
def step_invoke_plan_correct(context, plan_id, decision_id, mode):
guidance = getattr(context, "guidance", "")
yes_flag = getattr(context, "yes_flag", True)
dry_run = getattr(context, "dry_run", False)
context.correction_request = CorrectionRequest(
plan_id=plan_id,
target_decision_id=decision_id,
mode=CorrectionMode(mode),
guidance=guidance,
dry_run=dry_run,
)
context.show_confirmation_prompt = not yes_flag
context.dry_run_executed = dry_run
context.last_error = None
try:
if plan_id not in context.plans:
raise ValueError(f"Plan {plan_id} not found")
plan_status = context.plans[plan_id].get("status", "active")
if plan_status not in ("active", "pending"):
raise ValueError(
f"Plan {plan_id} is not correctable (status: {plan_status})"
)
plan_tree = context.decision_trees.get(plan_id, {})
if plan_tree:
all_decisions: set[str] = set(plan_tree.keys())
for children_list in plan_tree.values():
all_decisions.update(children_list)
if decision_id not in all_decisions:
raise ValueError(f"Decision {decision_id} not found")
if not dry_run and not yes_flag:
context.correction_succeeded = False
return
context.correction_succeeded = True
except Exception as e:
context.last_error = str(e)
context.correction_succeeded = False
@when('I invoke plan correct with plan_id "{plan_id}" decision_id "{decision_id}" mode "{mode}" without --yes flag')
def step_invoke_plan_correct_no_yes(context, plan_id, decision_id, mode):
"""Invoke plan correct without --yes flag (should show prompt)."""
context.correction_request = CorrectionRequest(
plan_id=plan_id,
target_decision_id=decision_id,
mode=CorrectionMode(mode),
)
context.show_confirmation_prompt = True
context.last_error = None
@when('I invoke plan correct with plan_id "{plan_id}" decision_id "{decision_id}" mode "{mode}" with --yes flag')
def step_invoke_plan_correct_with_yes(context, plan_id, decision_id, mode):
"""Invoke plan correct with --yes flag (should skip prompt)."""
context.correction_request = CorrectionRequest(
plan_id=plan_id,
target_decision_id=decision_id,
mode=CorrectionMode(mode),
)
context.show_confirmation_prompt = False
context.last_error = None
try:
if plan_id not in context.plans:
raise ValueError(f"Plan {plan_id} not found")
context.correction_succeeded = True
except Exception as e:
context.last_error = str(e)
context.correction_succeeded = False
@when('I invoke plan correct with plan_id "{plan_id}" decision_id "{decision_id}" mode "{mode}" with --dry-run flag')
def step_invoke_plan_correct_dry_run(context, plan_id, decision_id, mode):
"""Invoke plan correct with --dry-run flag."""
context.correction_request = CorrectionRequest(
plan_id=plan_id,
target_decision_id=decision_id,
mode=CorrectionMode(mode),
dry_run=True,
)
context.last_error = None
try:
if plan_id not in context.plans:
raise ValueError(f"Plan {plan_id} not found")
context.correction_succeeded = True
context.dry_run_executed = True
except Exception as e:
context.last_error = str(e)
context.correction_succeeded = False
@given('a plan with status "{status}"')
def step_plan_with_status(context, status):
"""Set the status of the current plan."""
if context.current_plan_id in context.plans:
context.plans[context.current_plan_id]["status"] = status
@then("the correction should succeed")
def step_correction_should_succeed(context):
"""Verify the correction succeeded."""
assert context.correction_succeeded, f"Correction failed: {context.last_error}"
@then('decisions "{decisions}" should be pruned from the tree')
def step_decisions_pruned(context, decisions):
"""Verify specific decisions were pruned."""
context.pruned_decisions = decisions.split(",")
@then('the LLM should be re-executed from decision "{decision_id}"')
def step_llm_reexecuted(context, decision_id):
"""Verify LLM was re-executed from target decision."""
context.reexecution_target = decision_id
@then('the plan status should transition to "{status1}" then "{status2}"')
def step_plan_status_transition(context, status1, status2):
"""Verify plan status transitions."""
context.expected_status_transitions = [status1, status2]
@then('the guidance should be appended to decision "{decision_id}" context')
def step_guidance_appended(context, decision_id):
"""Verify guidance was appended."""
assert hasattr(context.correction_request, "guidance")
assert context.correction_request.guidance is not None
@then("no decisions should be pruned")
def step_no_decisions_pruned(context):
"""Verify no decisions were pruned."""
context.pruned_decisions = []
@then("the LLM should not be re-executed")
def step_llm_not_reexecuted(context):
"""Verify LLM was not re-executed."""
context.reexecution_target = None
@then('the plan status should remain "{status}"')
def step_plan_status_remains(context, status):
"""Verify plan status remains unchanged."""
context.expected_status = status
@then("a confirmation prompt should be displayed")
def step_confirmation_prompt_displayed(context):
"""Verify confirmation prompt was shown."""
assert context.show_confirmation_prompt
@then('the prompt should warn about pruning decisions "{decisions}"')
def step_prompt_warns_about_pruning(context, decisions):
"""Verify prompt warns about pruning."""
context.warned_decisions = decisions.split(",")
@then("the prompt should ask for user confirmation")
def step_prompt_asks_confirmation(context):
"""Verify prompt asks for confirmation."""
assert context.show_confirmation_prompt
@then("no confirmation prompt should be displayed")
def step_no_confirmation_prompt(context):
"""Verify no confirmation prompt was shown."""
assert not context.show_confirmation_prompt
@then("the correction should proceed immediately")
def step_correction_proceeds_immediately(context):
"""Verify correction proceeded without prompt."""
assert context.correction_succeeded
@then("an error should be raised indicating plan not found")
def step_error_plan_not_found(context):
"""Verify error for missing plan."""
assert context.last_error is not None
assert "not found" in context.last_error.lower()
@then("an error should be raised indicating decision not found")
def step_error_decision_not_found(context):
"""Verify error for missing decision."""
assert context.last_error is not None
assert "not found" in context.last_error.lower()
@then("an error should be raised indicating plan is not correctable")
def step_error_plan_not_correctable(context):
"""Verify error for non-correctable plan."""
assert context.last_error is not None
assert "not correctable" in context.last_error.lower() or "cannot" in context.last_error.lower()
assert (
"not correctable" in context.last_error.lower()
or "cannot" in context.last_error.lower()
)
@then("the updated decision tree should be persisted to the database")
def step_tree_persisted(context):
"""Verify decision tree was persisted."""
context.tree_persisted = True
@then("agents plan tree should show the corrected tree")
def step_plan_tree_shows_corrected(context):
"""Verify plan tree shows corrected version."""
context.tree_displayed_correctly = True
@then("the impact analysis should be displayed")
def step_impact_analysis_displayed(context):
"""Verify impact analysis was shown."""
assert context.dry_run_executed
@then('the affected decisions should be "{decisions}"')
@then('the correction should report affected decisions "{decisions}"')
def step_affected_decisions(context, decisions):
"""Verify affected decisions list."""
context.affected_decisions = decisions.split(",")
@then("the risk level should be calculated")
def step_risk_level_calculated(context):
"""Verify risk level was calculated."""
context.risk_level_calculated = True
@then("no changes should be made to the plan")
def step_no_changes_made(context):
"""Verify no changes were made in dry-run."""
assert context.correction_request.dry_run