freemo
8e6642e8c9
CI / lint (pull_request) Successful in 13s
CI / typecheck (pull_request) Successful in 31s
CI / security (pull_request) Successful in 32s
CI / quality (pull_request) Successful in 15s
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 1m17s
CI / integration_tests (pull_request) Successful in 2m50s
CI / coverage (pull_request) Successful in 4m16s
CI / unit_tests (pull_request) Successful in 6m28s
CI / docker (pull_request) Successful in 39s
CI / lint (push) Successful in 12s
CI / typecheck (push) Successful in 31s
CI / security (push) Successful in 29s
CI / quality (push) Successful in 15s
CI / build (push) Successful in 21s
CI / unit_tests (push) Successful in 2m11s
CI / benchmark-regression (push) Has been skipped
CI / docker (push) Successful in 40s
CI / integration_tests (push) Successful in 2m58s
CI / coverage (push) Successful in 6m20s
CI / benchmark-publish (push) Successful in 14m5s
CI / benchmark-regression (pull_request) Successful in 33m18s
Extended _compute_affected_subtree() to BFS over both the structural tree (parent-child plan relationships) and decision_dependencies edges (influence DAG). The algorithm performs a single O(V+E) BFS pass that unions neighbors from both edge sources, using a visited set for cycle detection to guard against data corruption. Decision creation now supports dependency_decision_ids parameter in record_decision() which populates the in-memory influence DAG store. get_influence_edges() returns the adjacency list format consumed by CorrectionService. All public CorrectionService methods (analyze_impact, execute_revert, execute_correction, generate_dry_run_report) accept an optional influence_edges parameter while remaining backward-compatible (defaults to None, preserving structural-only traversal when not provided). Key design decisions: - Single BFS pass over union of structural + influence edges rather than separate traversals, ensuring O(V+E) complexity and consistent visit order - Cycle detection via visited set with warning log (not an error) since cycles indicate data corruption, not a programming error - Influence edge logging: traversal emits count of influence edges processed for observability - Backward-compatible API: existing callers that only pass decision_tree continue to work identically ISSUES CLOSED: #542
274 lines
10 KiB
Python
274 lines
10 KiB
Python
"""Step definitions for influence_dag_traversal.feature.
|
|
|
|
Tests the extended BFS that traverses both the structural decision tree
|
|
and the influence DAG (``decision_dependencies`` edges) during correction
|
|
affected-subtree computation, as well as influence dependency recording
|
|
during decision creation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from behave import given, then, when
|
|
from ulid import ULID
|
|
|
|
from cleveragents.application.services.correction_service import CorrectionService
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
from cleveragents.domain.models.core.correction import CorrectionMode
|
|
|
|
# A valid ULID used as plan_id in decision recording scenarios.
|
|
_TEST_PLAN_ULID = str(ULID())
|
|
|
|
# -------------------------------------------------------------------
|
|
# Helpers
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
def _parse_csv(text: str) -> list[str]:
|
|
"""Split a comma-separated string into a stripped list."""
|
|
return [s.strip() for s in text.split(",") if s.strip()]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Given: service setup
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("an influence-aware correction service")
|
|
def step_create_influence_service(context: object) -> None:
|
|
context.service = CorrectionService() # type: ignore[attr-defined]
|
|
context.correction_id = None # type: ignore[attr-defined]
|
|
context.decision_tree: dict[str, list[str]] = {} # type: ignore[attr-defined]
|
|
context.influence_edges: dict[str, list[str]] = {} # type: ignore[attr-defined]
|
|
context.result = None # type: ignore[attr-defined]
|
|
context.impact = None # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a decision service for influence tracking")
|
|
def step_create_decision_service(context: object) -> None:
|
|
context.decision_service = DecisionService() # type: ignore[attr-defined]
|
|
context.recorded_decisions: dict[str, str] = {} # type: ignore[attr-defined]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Given: correction requests
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'a correction request targeting decision "{decision_id}" '
|
|
'in plan "{plan_id}" for revert'
|
|
)
|
|
def step_create_revert_request_influence(
|
|
context: object, decision_id: str, plan_id: str
|
|
) -> None:
|
|
svc: CorrectionService = context.service # type: ignore[attr-defined]
|
|
req = svc.request_correction(
|
|
plan_id=plan_id,
|
|
target_decision_id=decision_id,
|
|
mode=CorrectionMode.REVERT,
|
|
)
|
|
context.correction_id = req.correction_id # type: ignore[attr-defined]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Given: structural trees
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'a structural tree where "{root}" has children "{children_csv}" '
|
|
'and "{mid}" has children "{mid_children_csv}"'
|
|
)
|
|
def step_structural_tree_two_level(
|
|
context: object,
|
|
root: str,
|
|
children_csv: str,
|
|
mid: str,
|
|
mid_children_csv: str,
|
|
) -> None:
|
|
tree: dict[str, list[str]] = context.decision_tree # type: ignore[attr-defined]
|
|
tree[root] = _parse_csv(children_csv)
|
|
tree[mid] = _parse_csv(mid_children_csv)
|
|
|
|
|
|
@given('a structural tree where "{root}" has children "{children_csv}"')
|
|
def step_structural_tree_simple(context: object, root: str, children_csv: str) -> None:
|
|
tree: dict[str, list[str]] = context.decision_tree # type: ignore[attr-defined]
|
|
tree[root] = _parse_csv(children_csv)
|
|
|
|
|
|
@given('a structural tree with no children for "{node}"')
|
|
def step_structural_tree_leaf(context: object, node: str) -> None:
|
|
# Ensure the tree is set but the node has no children (empty).
|
|
# The node will still be found as the BFS root target.
|
|
_ = node # The tree simply has no entry for this node.
|
|
context.decision_tree = context.decision_tree or {} # type: ignore[attr-defined]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Given: influence edges
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("no influence edges")
|
|
def step_no_influence_edges(context: object) -> None:
|
|
context.influence_edges = {} # type: ignore[attr-defined]
|
|
|
|
|
|
@given(
|
|
'influence edges where "{src}" influences "{targets_csv}" '
|
|
'and "{src2}" influences "{targets2_csv}"'
|
|
)
|
|
def step_influence_edges_two_sources(
|
|
context: object,
|
|
src: str,
|
|
targets_csv: str,
|
|
src2: str,
|
|
targets2_csv: str,
|
|
) -> None:
|
|
edges: dict[str, list[str]] = context.influence_edges # type: ignore[attr-defined]
|
|
edges[src] = _parse_csv(targets_csv)
|
|
edges[src2] = _parse_csv(targets2_csv)
|
|
|
|
|
|
@given('influence edges where "{src}" influences "{targets_csv}"')
|
|
def step_influence_edges_single(context: object, src: str, targets_csv: str) -> None:
|
|
edges: dict[str, list[str]] = context.influence_edges # type: ignore[attr-defined]
|
|
edges[src] = _parse_csv(targets_csv)
|
|
|
|
|
|
@given('influence edges with a cycle "{a}" to "{b}" to "{c}" back to "{a_again}"')
|
|
def step_influence_cycle(context: object, a: str, b: str, c: str, a_again: str) -> None:
|
|
edges: dict[str, list[str]] = context.influence_edges # type: ignore[attr-defined]
|
|
edges[a] = [b]
|
|
edges[b] = [c]
|
|
edges[c] = [a_again]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# When: impact analysis & execution
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@when("I analyze the impact with influence support")
|
|
def step_analyze_impact_influence(context: object) -> None:
|
|
svc: CorrectionService = context.service # type: ignore[attr-defined]
|
|
context.impact = svc.analyze_impact( # type: ignore[attr-defined]
|
|
context.correction_id, # type: ignore[attr-defined]
|
|
context.decision_tree, # type: ignore[attr-defined]
|
|
context.influence_edges, # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@when("I execute the revert with influence support")
|
|
def step_execute_revert_influence(context: object) -> None:
|
|
svc: CorrectionService = context.service # type: ignore[attr-defined]
|
|
context.result = svc.execute_revert( # type: ignore[attr-defined]
|
|
context.correction_id, # type: ignore[attr-defined]
|
|
context.decision_tree, # type: ignore[attr-defined]
|
|
context.influence_edges, # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# When: decision recording
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@when('I record a root decision "{dtype}" for the tracked plan')
|
|
def step_record_root_decision(context: object, dtype: str) -> None:
|
|
svc: DecisionService = context.decision_service # type: ignore[attr-defined]
|
|
plan_id = _TEST_PLAN_ULID
|
|
decision = svc.record_decision(
|
|
plan_id=plan_id,
|
|
decision_type=dtype,
|
|
question="Root question?",
|
|
chosen_option="Option A",
|
|
)
|
|
recorded: dict[str, str] = context.recorded_decisions # type: ignore[attr-defined]
|
|
recorded["root"] = decision.decision_id
|
|
context.plan_id = plan_id # type: ignore[attr-defined]
|
|
|
|
|
|
@when(
|
|
'I record a child decision "implementation_choice" depending on the root decision'
|
|
)
|
|
def step_record_child_with_dependency(context: object) -> None:
|
|
svc: DecisionService = context.decision_service # type: ignore[attr-defined]
|
|
recorded: dict[str, str] = context.recorded_decisions # type: ignore[attr-defined]
|
|
root_id = recorded["root"]
|
|
plan_id: str = context.plan_id # type: ignore[attr-defined]
|
|
decision = svc.record_decision(
|
|
plan_id=plan_id,
|
|
decision_type="implementation_choice",
|
|
question="Impl question?",
|
|
chosen_option="Option B",
|
|
parent_decision_id=root_id,
|
|
dependency_decision_ids=[root_id],
|
|
)
|
|
recorded["child"] = decision.decision_id
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Then: affected decision assertions
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@then('the affected decision set should be "{expected_csv}"')
|
|
def step_assert_affected_decisions(context: object, expected_csv: str) -> None:
|
|
expected = set(_parse_csv(expected_csv))
|
|
impact = context.impact # type: ignore[attr-defined]
|
|
actual = set(impact.affected_decisions)
|
|
assert actual == expected, f"Expected affected {expected}, got {actual}"
|
|
|
|
|
|
@then("no infinite loop occurred")
|
|
def step_assert_no_infinite_loop(context: object) -> None:
|
|
# The fact that we reached this step means BFS terminated.
|
|
# Verify no duplicates in affected list.
|
|
impact = context.impact # type: ignore[attr-defined]
|
|
decisions = impact.affected_decisions
|
|
assert len(decisions) == len(set(decisions)), (
|
|
f"Duplicate decisions found: {decisions}"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Then: revert result assertions
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@then('the reverted decisions should include "{decision_id}"')
|
|
def step_assert_reverted_contains(context: object, decision_id: str) -> None:
|
|
result = context.result # type: ignore[attr-defined]
|
|
assert decision_id in result.reverted_decisions, (
|
|
f"Expected {decision_id} in reverted {result.reverted_decisions}"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Then: influence edge assertions
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@then("the influence edges for the tracked plan should link the root to the child")
|
|
def step_assert_influence_edges_present(context: object) -> None:
|
|
svc: DecisionService = context.decision_service # type: ignore[attr-defined]
|
|
recorded: dict[str, str] = context.recorded_decisions # type: ignore[attr-defined]
|
|
plan_id: str = context.plan_id # type: ignore[attr-defined]
|
|
edges = svc.get_influence_edges(plan_id)
|
|
root_id = recorded["root"]
|
|
child_id = recorded["child"]
|
|
assert root_id in edges, f"Expected root {root_id} in edges {edges}"
|
|
assert child_id in edges[root_id], (
|
|
f"Expected child {child_id} in edges[{root_id}]={edges[root_id]}"
|
|
)
|
|
|
|
|
|
@then("the influence edges for the tracked plan should be empty")
|
|
def step_assert_influence_edges_empty(context: object) -> None:
|
|
svc: DecisionService = context.decision_service # type: ignore[attr-defined]
|
|
plan_id: str = context.plan_id # type: ignore[attr-defined]
|
|
edges = svc.get_influence_edges(plan_id)
|
|
assert edges == {}, f"Expected empty edges, got {edges}"
|