"""Step definitions for decision_service_coverage_boost.feature. Targets uncovered lines in decision_service.py that were missed by existing test suites. All step names use the ``dscb-`` prefix. Targeted uncovered lines: 318-319 actor_reasoning validation 488 BFS visited-node continue in get_tree 500-501 Orphan subtree handling in get_tree 534 Cycle detection in get_path_to_root 578-579 mark_superseded persisted fallback for replacement 600 mark_superseded not-found in memory mode 627-628 delete_decision persisted fallback 741 SequenceConflictError in _next_sequence 757 _rehydrate_sequence early return 806 _record_dependencies blank upstream skip 832 get_influence_edges cross-plan filtering """ from __future__ import annotations from contextlib import contextmanager from typing import Any from unittest.mock import MagicMock from behave import given, then, when from behave.runner import Context from ulid import ULID from cleveragents.application.services.decision_service import ( _MAX_ACTOR_REASONING, DecisionNotFoundError, DecisionService, SequenceConflictError, ) from cleveragents.core.exceptions import ValidationError from cleveragents.domain.models.core.decision import ( Decision, DecisionType, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _ulid() -> str: """Generate a fresh ULID string.""" return str(ULID()) def _make_decision( decision_id: str | None = None, plan_id: str | None = None, parent_decision_id: str | None = None, sequence_number: int = 0, decision_type: DecisionType = DecisionType.STRATEGY_CHOICE, superseded_by: str | None = None, ) -> Decision: """Build a minimal valid Decision for testing.""" kwargs: dict[str, Any] = { "plan_id": plan_id or _ulid(), "sequence_number": sequence_number, "decision_type": decision_type, "question": "Test question", "chosen_option": "Test option", } if decision_id is not None: kwargs["decision_id"] = decision_id if parent_decision_id is not None: kwargs["parent_decision_id"] = parent_decision_id if superseded_by is not None: kwargs["superseded_by"] = superseded_by return Decision(**kwargs) def _build_mock_uow() -> tuple[MagicMock, MagicMock, MagicMock]: """Build a mock UnitOfWork with a mock transaction context manager.""" mock_decisions = MagicMock(name="decisions_repo") mock_ctx = MagicMock(name="uow_context") mock_ctx.decisions = mock_decisions mock_uow = MagicMock(name="unit_of_work") @contextmanager def _fake_transaction(): yield mock_ctx mock_uow.transaction = _fake_transaction return mock_uow, mock_ctx, mock_decisions # --------------------------------------------------------------------------- # Shared Given: in-memory DecisionService # --------------------------------------------------------------------------- @given("dscb- a DecisionService in memory mode") def step_dscb_inmemory_service(context: Context) -> None: context.dscb_service = DecisionService() context.dscb_plan_id = _ulid() context.dscb_error = None context.dscb_result = None # =================================================================== # Lines 318-319: actor_reasoning exceeds max length # =================================================================== @when("dscb- I record a decision with actor_reasoning exceeding the max length") def step_dscb_record_oversized_reasoning(context: Context) -> None: try: context.dscb_service.record_decision( plan_id=context.dscb_plan_id, decision_type=DecisionType.PROMPT_DEFINITION, question="Test question", chosen_option="Test option", actor_reasoning="x" * (_MAX_ACTOR_REASONING + 1), ) context.dscb_error = None except ValidationError as exc: context.dscb_error = exc @then('dscb- a ValidationError should have been raised mentioning "actor_reasoning"') def step_dscb_check_validation_error_reasoning(context: Context) -> None: assert context.dscb_error is not None, "Expected ValidationError but got None" assert isinstance(context.dscb_error, ValidationError), ( f"Expected ValidationError, got {type(context.dscb_error).__name__}" ) assert "actor_reasoning" in str(context.dscb_error), ( f"Error message should mention 'actor_reasoning': {context.dscb_error}" ) # =================================================================== # Line 488: BFS continue on already-visited node in get_tree # =================================================================== @given("dscb- a decision tree where the same child is reachable from two parents") def step_dscb_tree_with_shared_child(context: Context) -> None: """Create a tree where a child references one parent but is manually added to a second parent's adjacency, causing it to appear in the BFS queue twice. The simplest way to hit the `continue` on line 488 is to manually populate the in-memory cache with a structure where the BFS queue would encounter the same node twice. We create: root (no parent) child1 (parent=root) child2 (parent=root) Then we also set child2.parent_decision_id = root (already is), but we add child2's ID to `child1`'s adjacency as well. Since Decision is frozen, we do this by inserting child2 under child1 in the children_map — the simplest approach is to give child2 a parent_decision_id of child1, but *also* make child2 appear under root. Actually, a simpler approach: create two decisions that both claim to be children of root, and then also one of them claims to be a child of the other. But the Decision model is frozen. The cleanest approach: create a cycle-like structure: root (parent=None) childA (parent=root) childB (parent=childA) -- also we'll manually add childB as child of root by hacking _plan_decisions But that won't trigger line 488 since each node's parent_decision_id is unique. Line 488 fires when a decision_id is already in `visited`. The BFS builds children_map from parent_decision_id. For a node to appear twice in the queue, it would need to be a child of two parents. Since parent_decision_id is singular, we can't do that directly. But we can have *two Decision objects with the same decision_id* in the list returned by list_decisions — this is unusual but possible if we manually stuff the cache. Actually, looking more carefully: the BFS also appends children, and if a child's parent is one node but it also appears as the grandchild of another path, it won't be revisited unless it literally was already queued. Let me re-read the code: ``` while queue: node = queue.popleft() if node.decision_id in visited: continue # ← line 488 visited.add(node.decision_id) ``` The queue starts with `roots`. Then for each node, its children are appended. For a node to be visited twice, it would need to be in `roots` AND also be a child of another node. Or be a child of two different nodes. Simplest trigger: make the root also appear as a child of itself (parent=None makes it root, but we also add it to children_map under some other key). Or even simpler: make a decision that is a root (parent=None) and also appears in children_map under another node — by creating it with parent=None and then having another decision whose decision_id equals the root's parent... no. Let me just directly manipulate the service's internal state. """ svc = context.dscb_service plan_id = context.dscb_plan_id root_id = _ulid() child_id = _ulid() # Root decision root = _make_decision( decision_id=root_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, parent_decision_id=None, ) # Child that claims root as parent child = _make_decision( decision_id=child_id, plan_id=plan_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, parent_decision_id=root_id, ) # Insert ANOTHER copy of child with same decision_id — this means # when list_decisions returns, child appears once, but we'll add # child_id to _plan_decisions twice so it appears in list_decisions # twice (since list_decisions iterates _plan_decisions). svc._decisions[root_id] = root svc._decisions[child_id] = child svc._plan_decisions[plan_id] = [root_id, child_id, child_id] svc._plan_sequence[plan_id] = 2 context.dscb_expected_unique_count = 2 @when("dscb- I call get_tree for the plan") def step_dscb_call_get_tree(context: Context) -> None: context.dscb_result = context.dscb_service.get_tree(context.dscb_plan_id) @then("dscb- the tree result should contain each decision exactly once") def step_dscb_tree_unique_nodes(context: Context) -> None: ids = [d.decision_id for d in context.dscb_result] assert len(ids) == len(set(ids)), f"Tree contains duplicate decision IDs: {ids}" assert len(set(ids)) == context.dscb_expected_unique_count, ( f"Expected {context.dscb_expected_unique_count} unique decisions, " f"got {len(set(ids))}" ) # =================================================================== # Lines 500-501: Orphaned subtrees in get_tree # =================================================================== @given( "dscb- a plan with a root decision and an orphaned decision with a dangling parent" ) def step_dscb_tree_with_orphan(context: Context) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id root_id = _ulid() orphan_id = _ulid() nonexistent_parent_id = _ulid() # Not in the plan's decisions root = _make_decision( decision_id=root_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, parent_decision_id=None, ) # Orphan's parent is not in the plan → it's unreachable from root orphan = _make_decision( decision_id=orphan_id, plan_id=plan_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, parent_decision_id=nonexistent_parent_id, ) svc._decisions[root_id] = root svc._decisions[orphan_id] = orphan svc._plan_decisions[plan_id] = [root_id, orphan_id] svc._plan_sequence[plan_id] = 2 context.dscb_root_id = root_id context.dscb_orphan_id = orphan_id @then("dscb- the tree result should include both the root and the orphaned decision") def step_dscb_tree_includes_orphan(context: Context) -> None: ids = {d.decision_id for d in context.dscb_result} assert context.dscb_root_id in ids, "Root decision missing from tree result" assert context.dscb_orphan_id in ids, "Orphaned decision missing from tree result" @then("dscb- the root decision should appear before the orphan") def step_dscb_root_before_orphan(context: Context) -> None: ids = [d.decision_id for d in context.dscb_result] root_idx = ids.index(context.dscb_root_id) orphan_idx = ids.index(context.dscb_orphan_id) assert root_idx < orphan_idx, ( f"Root at index {root_idx} should appear before orphan at {orphan_idx}" ) # =================================================================== # Line 534: get_path_to_root breaks on cycle # =================================================================== @given("dscb- two decisions that form a parent cycle") def step_dscb_cyclic_parents(context: Context) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id id_a = _ulid() id_b = _ulid() # A's parent is B, B's parent is A → cycle decision_a = _make_decision( decision_id=id_a, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.STRATEGY_CHOICE, parent_decision_id=id_b, ) decision_b = _make_decision( decision_id=id_b, plan_id=plan_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, parent_decision_id=id_a, ) svc._decisions[id_a] = decision_a svc._decisions[id_b] = decision_b svc._plan_decisions[plan_id] = [id_a, id_b] svc._plan_sequence[plan_id] = 2 context.dscb_cycle_start_id = id_a @when("dscb- I call get_path_to_root from the first cycled decision") def step_dscb_path_from_cycle(context: Context) -> None: context.dscb_result = context.dscb_service.get_path_to_root( context.dscb_cycle_start_id ) @then("dscb- the path should terminate without infinite loop") def step_dscb_path_terminates(context: Context) -> None: # If we got here, the path terminated (no infinite loop) assert context.dscb_result is not None, "Path result should not be None" assert len(context.dscb_result) > 0, "Path should contain at least one decision" @then("dscb- the path should contain exactly {count:d} decisions") def step_dscb_path_count(context: Context, count: int) -> None: assert len(context.dscb_result) == count, ( f"Expected {count} decisions in path, got {len(context.dscb_result)}" ) # =================================================================== # Lines 578-579: mark_superseded DB fallback for replacement # =================================================================== @given("dscb- a DecisionService with a mock UoW for superseded fallback") def step_dscb_service_mock_uow_superseded(context: Context) -> None: mock_uow, mock_ctx, mock_decisions = _build_mock_uow() svc = DecisionService(unit_of_work=mock_uow) # Silence logger mock_logger = MagicMock(name="structlog_logger") mock_logger.bind = MagicMock(return_value=mock_logger) svc._logger = mock_logger context.dscb_service = svc context.dscb_mock_uow = mock_uow context.dscb_mock_ctx = mock_ctx context.dscb_mock_decisions = mock_decisions context.dscb_plan_id = _ulid() context.dscb_error = None context.dscb_result = None @given("dscb- the replacement decision exists only in the database not in cache") def step_dscb_replacement_in_db_only(context: Context) -> None: """Replacement is NOT in _decisions cache; the mock DB returns it.""" plan_id = context.dscb_plan_id context.dscb_replacement_id = _ulid() context.dscb_original_id = _ulid() replacement = _make_decision( decision_id=context.dscb_replacement_id, plan_id=plan_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, ) # DB lookup returns the replacement context.dscb_mock_decisions.get.return_value = replacement context.dscb_replacement = replacement @given("dscb- the original decision exists in the cache") def step_dscb_original_in_cache(context: Context) -> None: plan_id = context.dscb_plan_id svc = context.dscb_service original = _make_decision( decision_id=context.dscb_original_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, ) svc._decisions[context.dscb_original_id] = original # The mock update_superseded_by should return the updated decision updated = original.with_superseded_by(context.dscb_replacement_id) context.dscb_mock_decisions.update_superseded_by.return_value = updated context.dscb_expected_updated = updated @when("dscb- I call mark_superseded with the original and replacement IDs") def step_dscb_call_mark_superseded(context: Context) -> None: try: context.dscb_result = context.dscb_service.mark_superseded( context.dscb_original_id, context.dscb_replacement_id, ) context.dscb_error = None except Exception as exc: context.dscb_error = exc context.dscb_result = None @then("dscb- the original decision should be marked as superseded") def step_dscb_original_superseded(context: Context) -> None: assert context.dscb_error is None, f"Unexpected error: {context.dscb_error}" assert context.dscb_result is not None, "Expected a result from mark_superseded" assert context.dscb_result.superseded_by == context.dscb_replacement_id, ( f"Expected superseded_by={context.dscb_replacement_id}, " f"got {context.dscb_result.superseded_by}" ) # =================================================================== # Line 600: mark_superseded not-found in memory mode # =================================================================== @given("dscb- a replacement decision is recorded for the plan") def step_dscb_record_replacement(context: Context) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id context.dscb_replacement_id = _ulid() replacement = _make_decision( decision_id=context.dscb_replacement_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.STRATEGY_CHOICE, ) svc._decisions[context.dscb_replacement_id] = replacement @when("dscb- I call mark_superseded with a nonexistent original ID") def step_dscb_mark_superseded_nonexistent(context: Context) -> None: nonexistent_id = _ulid() try: context.dscb_service.mark_superseded( nonexistent_id, context.dscb_replacement_id ) context.dscb_error = None except DecisionNotFoundError as exc: context.dscb_error = exc @then("dscb- a DecisionNotFoundError should have been raised") def step_dscb_check_not_found_error(context: Context) -> None: assert context.dscb_error is not None, "Expected DecisionNotFoundError but got None" assert isinstance(context.dscb_error, DecisionNotFoundError), ( f"Expected DecisionNotFoundError, got {type(context.dscb_error).__name__}" ) # =================================================================== # Lines 627-628: delete_decision DB fallback # =================================================================== @given("dscb- a DecisionService with a mock UoW for delete fallback") def step_dscb_service_mock_uow_delete(context: Context) -> None: mock_uow, mock_ctx, mock_decisions = _build_mock_uow() svc = DecisionService(unit_of_work=mock_uow) mock_logger = MagicMock(name="structlog_logger") mock_logger.bind = MagicMock(return_value=mock_logger) svc._logger = mock_logger context.dscb_service = svc context.dscb_mock_uow = mock_uow context.dscb_mock_ctx = mock_ctx context.dscb_mock_decisions = mock_decisions context.dscb_plan_id = _ulid() context.dscb_error = None context.dscb_result = None @given("dscb- the decision exists in DB but not in cache for delete") def step_dscb_decision_in_db_not_cache(context: Context) -> None: plan_id = context.dscb_plan_id context.dscb_decision_id = _ulid() decision = _make_decision( decision_id=context.dscb_decision_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.STRATEGY_CHOICE, ) # Decision is NOT in svc._decisions (not in cache) # DB lookup returns it context.dscb_mock_decisions.get.return_value = decision @when("dscb- I call delete_decision with the decision ID") def step_dscb_call_delete(context: Context) -> None: try: context.dscb_result = context.dscb_service.delete_decision( context.dscb_decision_id ) context.dscb_error = None except Exception as exc: context.dscb_error = exc context.dscb_result = None @then("dscb- the decision should be deleted via the UoW") def step_dscb_delete_via_uow(context: Context) -> None: assert context.dscb_error is None, f"Unexpected error: {context.dscb_error}" assert context.dscb_result is True, "delete_decision should return True" context.dscb_mock_decisions.delete.assert_called_once_with(context.dscb_decision_id) # =================================================================== # Line 741: SequenceConflictError in _next_sequence # =================================================================== @given("dscb- a plan with a decision manually inserted at sequence 0") def step_dscb_manual_sequence_conflict(context: Context) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id decision_id = _ulid() decision = _make_decision( decision_id=decision_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, ) # Insert into cache manually (bypassing _store_decision which # would increment the counter) svc._decisions[decision_id] = decision svc._plan_decisions.setdefault(plan_id, []).append(decision_id) # Set the sequence counter to 0 so _next_sequence tries to assign 0 # again, which conflicts with the existing decision svc._plan_sequence[plan_id] = 0 svc._sequence_initialised.add(plan_id) @when("dscb- I try to generate the next sequence for that plan") def step_dscb_try_next_sequence(context: Context) -> None: try: context.dscb_service._next_sequence(context.dscb_plan_id) context.dscb_error = None except SequenceConflictError as exc: context.dscb_error = exc @then("dscb- a SequenceConflictError should have been raised") def step_dscb_check_sequence_conflict(context: Context) -> None: assert context.dscb_error is not None, "Expected SequenceConflictError but got None" assert isinstance(context.dscb_error, SequenceConflictError), ( f"Expected SequenceConflictError, got {type(context.dscb_error).__name__}" ) # =================================================================== # Line 757: _rehydrate_sequence early return # =================================================================== @given("dscb- the plan sequence counter is already set to {value:d}") def step_dscb_preset_sequence(context: Context, value: int) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id svc._plan_sequence[plan_id] = value @when("dscb- I call rehydrate_sequence for the plan") def step_dscb_call_rehydrate(context: Context) -> None: context.dscb_service._rehydrate_sequence(context.dscb_plan_id) @then("dscb- the plan sequence counter should still be {value:d}") def step_dscb_check_sequence_unchanged(context: Context, value: int) -> None: actual = context.dscb_service._plan_sequence.get(context.dscb_plan_id) assert actual == value, f"Expected sequence counter {value}, got {actual}" # =================================================================== # Line 806: _record_dependencies skips blank upstream IDs # =================================================================== @when("dscb- I record a decision with dependency IDs including blanks") def step_dscb_record_with_blank_deps(context: Context) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id # We need a valid upstream decision for the non-blank ID valid_upstream_id = _ulid() upstream = _make_decision( decision_id=valid_upstream_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, ) svc._decisions[valid_upstream_id] = upstream # Record with blanks + valid IDs context.dscb_result = svc.record_decision( plan_id=plan_id, decision_type=DecisionType.STRATEGY_CHOICE, question="Test with deps", chosen_option="Option A", dependency_decision_ids=["", " ", valid_upstream_id], ) context.dscb_valid_upstream_id = valid_upstream_id @then("dscb- only non-blank upstream IDs should be recorded in the dependency DAG") def step_dscb_check_deps_no_blanks(context: Context) -> None: svc = context.dscb_service # The valid upstream should have an edge to the new decision valid_id = context.dscb_valid_upstream_id new_decision_id = context.dscb_result.decision_id assert valid_id in svc._dependencies, ( f"Expected upstream {valid_id} in dependency DAG" ) assert new_decision_id in svc._dependencies[valid_id], ( f"Expected {new_decision_id} as target of {valid_id}" ) # Blank IDs should NOT appear as keys for key in svc._dependencies: assert key.strip(), f"Blank key found in dependency DAG: '{key}'" # =================================================================== # Line 832: get_influence_edges cross-plan filtering # =================================================================== @given("dscb- a plan with decisions and cross-plan dependency edges") def step_dscb_cross_plan_deps(context: Context) -> None: svc = context.dscb_service plan_id = context.dscb_plan_id # Decision in our plan in_plan_id = _ulid() in_plan_target_id = _ulid() in_plan = _make_decision( decision_id=in_plan_id, plan_id=plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, ) in_plan_target = _make_decision( decision_id=in_plan_target_id, plan_id=plan_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, parent_decision_id=in_plan_id, ) svc._decisions[in_plan_id] = in_plan svc._decisions[in_plan_target_id] = in_plan_target svc._plan_decisions[plan_id] = [in_plan_id, in_plan_target_id] # Decision from another plan that also has a dependency edge other_plan_id = _ulid() outside_id = _ulid() outside = _make_decision( decision_id=outside_id, plan_id=other_plan_id, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, ) svc._decisions[outside_id] = outside svc._plan_decisions[other_plan_id] = [outside_id] # Add dependency edges: # in_plan_id → in_plan_target_id (within plan) # outside_id → in_plan_target_id (cross-plan — should be excluded) svc._dependencies[in_plan_id] = [in_plan_target_id] svc._dependencies[outside_id] = [in_plan_target_id] context.dscb_in_plan_id = in_plan_id context.dscb_in_plan_target_id = in_plan_target_id context.dscb_outside_id = outside_id @when("dscb- I call get_influence_edges for the plan") def step_dscb_call_influence_edges(context: Context) -> None: context.dscb_result = context.dscb_service.get_influence_edges(context.dscb_plan_id) @then("dscb- only edges within the plan should be returned") def step_dscb_check_influence_edges(context: Context) -> None: edges = context.dscb_result # The in-plan edge should be present assert context.dscb_in_plan_id in edges, ( f"Expected in-plan source {context.dscb_in_plan_id} in edges" ) assert context.dscb_in_plan_target_id in edges[context.dscb_in_plan_id], ( "Expected in-plan target in edge targets" ) # The cross-plan edge should be excluded assert context.dscb_outside_id not in edges, ( f"Cross-plan source {context.dscb_outside_id} should NOT be in edges" )