From 8e6642e8c9dd66acdbbc6c38b182f5ee1183fb85 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Wed, 4 Mar 2026 04:22:02 +0000 Subject: [PATCH] feat(decision): implement influence DAG traversal in correction affected subtree computation 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 --- CHANGELOG.md | 9 + benchmarks/influence_dag_bench.py | 190 ++++++++++++ features/influence_dag_traversal.feature | 124 ++++++++ .../steps/influence_dag_traversal_steps.py | 273 ++++++++++++++++++ robot/helper_influence_dag_traversal.py | 158 ++++++++++ robot/influence_dag_traversal.robot | 53 ++++ .../services/correction_service.py | 107 ++++++- .../application/services/decision_service.py | 73 +++++ 8 files changed, 971 insertions(+), 16 deletions(-) create mode 100644 benchmarks/influence_dag_bench.py create mode 100644 features/influence_dag_traversal.feature create mode 100644 features/steps/influence_dag_traversal_steps.py create mode 100644 robot/helper_influence_dag_traversal.py create mode 100644 robot/influence_dag_traversal.robot diff --git a/CHANGELOG.md b/CHANGELOG.md index d0ee2b079..f50b37d61 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Unreleased +- Extended `CorrectionService._compute_affected_subtree()` to BFS over both the structural + decision tree (parent-child) and the influence DAG (`decision_dependencies` edges). Added + cycle detection guard via visited set. Updated `DecisionService.record_decision()` to accept + `dependency_decision_ids` parameter for recording influence relationships during decision + creation. Added `DecisionService.get_influence_edges()` to retrieve influence DAG as + adjacency list. All public methods on `CorrectionService` (`analyze_impact`, + `execute_revert`, `execute_correction`, `generate_dry_run_report`) now accept optional + `influence_edges` parameter. Includes Behave BDD scenarios, Robot Framework integration + tests, and ASV benchmarks. (#542) - Fixed `context inspect` to display project-scoped tier fragment counts instead of global counts. Added `ContextTierService.get_scoped_metrics()` which returns fragment population counts filtered to the target project while keeping hit/miss counters as global service diff --git a/benchmarks/influence_dag_bench.py b/benchmarks/influence_dag_bench.py new file mode 100644 index 000000000..0fe4b5be0 --- /dev/null +++ b/benchmarks/influence_dag_bench.py @@ -0,0 +1,190 @@ +"""Airspeed Velocity benchmarks for influence DAG traversal. + +Measures BFS traversal performance over combined structural tree and +influence DAG edges at varying graph sizes, including cycle detection +overhead. +""" + +from __future__ import annotations + +from cleveragents.application.services.correction_service import CorrectionService +from cleveragents.domain.models.core.correction import CorrectionMode + +# --------------------------------------------------------------------------- +# Graph generators +# --------------------------------------------------------------------------- + +_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6EA00" + + +def _linear_tree(root: str, depth: int) -> dict[str, list[str]]: + """Build a linear chain of *depth* nodes starting at *root*.""" + tree: dict[str, list[str]] = {} + current = root + for i in range(1, depth): + child = f"{root}_c{i}" + tree[current] = [child] + current = child + return tree + + +def _star_influence(root: str, fan_out: int) -> dict[str, list[str]]: + """Build a star influence DAG: root influences *fan_out* nodes.""" + return {root: [f"{root}_inf{i}" for i in range(fan_out)]} + + +def _chain_influence(root: str, depth: int) -> dict[str, list[str]]: + """Build a linear chain influence DAG of *depth* nodes.""" + dag: dict[str, list[str]] = {} + current = root + for i in range(1, depth): + nxt = f"{root}_d{i}" + dag[current] = [nxt] + current = nxt + return dag + + +def _mixed_graph( + root: str, + tree_depth: int, + influence_fan_out: int, +) -> tuple[dict[str, list[str]], dict[str, list[str]]]: + """Build a combined tree + influence DAG.""" + tree = _linear_tree(root, tree_depth) + dag = _star_influence(root, influence_fan_out) + return tree, dag + + +# --------------------------------------------------------------------------- +# Structural-only BFS benchmarks (baseline) +# --------------------------------------------------------------------------- + + +class StructuralOnlyBfsSuite: + """Baseline: BFS over structural tree only (no influence edges).""" + + def setup(self) -> None: + """Prepare trees and service instances.""" + self.svc = CorrectionService() + self.small_tree = _linear_tree("D1", 10) + self.medium_tree = _linear_tree("D1", 100) + self.large_tree = _linear_tree("D1", 1000) + + def time_small_structural_only(self) -> None: + """BFS 10-node linear tree, no influence edges.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.small_tree) + + def time_medium_structural_only(self) -> None: + """BFS 100-node linear tree, no influence edges.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.medium_tree) + + def time_large_structural_only(self) -> None: + """BFS 1000-node linear tree, no influence edges.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.large_tree) + + +# --------------------------------------------------------------------------- +# Influence-only BFS benchmarks +# --------------------------------------------------------------------------- + + +class InfluenceOnlyBfsSuite: + """BFS over influence DAG only (no structural tree).""" + + def setup(self) -> None: + """Prepare influence DAGs.""" + self.svc = CorrectionService() + self.empty_tree: dict[str, list[str]] = {} + self.small_dag = _chain_influence("D1", 10) + self.medium_dag = _chain_influence("D1", 100) + self.large_dag = _chain_influence("D1", 1000) + self.star_dag = _star_influence("D1", 500) + + def time_small_influence_only(self) -> None: + """BFS 10-node influence chain.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.empty_tree, self.small_dag) + + def time_medium_influence_only(self) -> None: + """BFS 100-node influence chain.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.empty_tree, self.medium_dag) + + def time_large_influence_only(self) -> None: + """BFS 1000-node influence chain.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.empty_tree, self.large_dag) + + def time_star_influence(self) -> None: + """BFS star DAG with 500 fan-out.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.empty_tree, self.star_dag) + + +# --------------------------------------------------------------------------- +# Combined structural + influence BFS benchmarks +# --------------------------------------------------------------------------- + + +class CombinedBfsSuite: + """BFS over combined structural tree + influence DAG.""" + + def setup(self) -> None: + """Prepare combined graphs.""" + self.svc = CorrectionService() + self.small_tree, self.small_dag = _mixed_graph("D1", 10, 10) + self.medium_tree, self.medium_dag = _mixed_graph("D1", 100, 50) + self.large_tree, self.large_dag = _mixed_graph("D1", 500, 200) + + def time_small_combined(self) -> None: + """BFS combined 10+10 graph.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.small_tree, self.small_dag) + + def time_medium_combined(self) -> None: + """BFS combined 100+50 graph.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.medium_tree, self.medium_dag) + + def time_large_combined(self) -> None: + """BFS combined 500+200 graph.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.large_tree, self.large_dag) + + +# --------------------------------------------------------------------------- +# Cycle detection overhead benchmarks +# --------------------------------------------------------------------------- + + +class CycleDetectionSuite: + """Measure overhead of cycle detection in corrupt DAGs.""" + + def setup(self) -> None: + """Prepare DAGs with cycles.""" + self.svc = CorrectionService() + self.empty_tree: dict[str, list[str]] = {} + # Small cycle: D1 → D2 → D3 → D1 + self.small_cycle = {"D1": ["D2"], "D2": ["D3"], "D3": ["D1"]} + # Medium: chain of 50 with cycle at end + dag: dict[str, list[str]] = {} + current = "D1" + for i in range(1, 50): + nxt = f"D1_cyc{i}" + dag[current] = [nxt] + current = nxt + dag[current] = ["D1"] # cycle back to start + self.medium_cycle = dag + + def time_small_cycle(self) -> None: + """BFS with 3-node cycle.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.empty_tree, self.small_cycle) + + def time_medium_cycle(self) -> None: + """BFS with 50-node chain + cycle.""" + req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + self.svc.analyze_impact(req.correction_id, self.empty_tree, self.medium_cycle) diff --git a/features/influence_dag_traversal.feature b/features/influence_dag_traversal.feature new file mode 100644 index 000000000..993623244 --- /dev/null +++ b/features/influence_dag_traversal.feature @@ -0,0 +1,124 @@ +Feature: Influence DAG traversal in correction affected subtree + The affected subtree computation must traverse BOTH the structural + tree (parent-child) AND the influence DAG (decision_dependencies) + edges so that corrections cascade through influence relationships. + + # ================================================================ + # Structural-only (existing behaviour preserved) + # ================================================================ + + Scenario: Structural-only subtree with no influence edges + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree where "D1" has children "D2,D3" and "D2" has children "D4" + And no influence edges + When I analyze the impact with influence support + Then the affected decision set should be "D1,D2,D3,D4" + + + Scenario: Structural-only single node + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree with no children for "D1" + And no influence edges + When I analyze the impact with influence support + Then the affected decision set should be "D1" + + + # ================================================================ + # Influence-only (new behaviour) + # ================================================================ + + Scenario: Influence-only affected subtree finds linked decisions + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree with no children for "D1" + And influence edges where "D1" influences "D5,D6" and "D5" influences "D7" + When I analyze the impact with influence support + Then the affected decision set should be "D1,D5,D6,D7" + + + Scenario: Influence-only with no structural children + Given an influence-aware correction service + And a correction request targeting decision "DX" in plan "P1" for revert + And a structural tree with no children for "DX" + And influence edges where "DX" influences "DY" + When I analyze the impact with influence support + Then the affected decision set should be "DX,DY" + + + # ================================================================ + # Combined structural + influence (union) + # ================================================================ + + Scenario: Combined structural and influence subtree is the union + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree where "D1" has children "D2,D3" + And influence edges where "D1" influences "D4" and "D3" influences "D5" + When I analyze the impact with influence support + Then the affected decision set should be "D1,D2,D3,D4,D5" + + + Scenario: Overlapping structural and influence edges deduplicate + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree where "D1" has children "D2" + And influence edges where "D1" influences "D2" + When I analyze the impact with influence support + Then the affected decision set should be "D1,D2" + + + # ================================================================ + # Cycle detection (corrupt data) + # ================================================================ + + Scenario: Cycle in influence DAG is handled gracefully + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree with no children for "D1" + And influence edges with a cycle "D1" to "D2" to "D3" back to "D1" + When I analyze the impact with influence support + Then the affected decision set should be "D1,D2,D3" + And no infinite loop occurred + + + Scenario: Self-loop in influence DAG is handled gracefully + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree with no children for "D1" + And influence edges where "D1" influences "D1" + When I analyze the impact with influence support + Then the affected decision set should be "D1" + + + # ================================================================ + # Decision creation populates decision_dependencies + # ================================================================ + + Scenario: Decision creation records influence dependencies + Given a decision service for influence tracking + When I record a root decision "strategy_choice" for the tracked plan + And I record a child decision "implementation_choice" depending on the root decision + Then the influence edges for the tracked plan should link the root to the child + + + Scenario: Decision creation without dependencies records no edges + Given a decision service for influence tracking + When I record a root decision "strategy_choice" for the tracked plan + Then the influence edges for the tracked plan should be empty + + + # ================================================================ + # Revert execution with influence edges + # ================================================================ + + Scenario: Revert with influence edges invalidates all affected + Given an influence-aware correction service + And a correction request targeting decision "D1" in plan "P1" for revert + And a structural tree where "D1" has children "D2" + And influence edges where "D1" influences "D3" + When I execute the revert with influence support + Then the reverted decisions should include "D1" + And the reverted decisions should include "D2" + And the reverted decisions should include "D3" diff --git a/features/steps/influence_dag_traversal_steps.py b/features/steps/influence_dag_traversal_steps.py new file mode 100644 index 000000000..1fac02fbb --- /dev/null +++ b/features/steps/influence_dag_traversal_steps.py @@ -0,0 +1,273 @@ +"""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}" diff --git a/robot/helper_influence_dag_traversal.py b/robot/helper_influence_dag_traversal.py new file mode 100644 index 000000000..4ccf6eef3 --- /dev/null +++ b/robot/helper_influence_dag_traversal.py @@ -0,0 +1,158 @@ +"""Helper script for influence DAG traversal Robot Framework tests. + +Exercises CorrectionService with both structural tree and influence +DAG edges, plus DecisionService influence dependency recording. +""" + +from __future__ import annotations + +import sys + +from cleveragents.application.services.correction_service import CorrectionService +from cleveragents.application.services.decision_service import DecisionService +from cleveragents.domain.models.core.correction import CorrectionMode + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- +_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV" + + +def _structural_only() -> None: + """Revert with only structural children.""" + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + tree = {"D1": ["D2", "D3"], "D2": ["D4"]} + result = svc.execute_revert(req.correction_id, tree) + assert result.status == "applied" + assert set(result.reverted_decisions) == {"D1", "D2", "D3", "D4"} + print("structural-only-ok") + + +def _influence_only() -> None: + """Revert with only influence DAG edges.""" + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + tree: dict[str, list[str]] = {} + influence = {"D1": ["D5", "D6"], "D5": ["D7"]} + result = svc.execute_revert(req.correction_id, tree, influence) + assert result.status == "applied" + assert set(result.reverted_decisions) == {"D1", "D5", "D6", "D7"} + print("influence-only-ok") + + +def _combined_revert() -> None: + """Revert traverses union of structural + influence edges.""" + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + tree = {"D1": ["D2"]} + influence = {"D1": ["D3"], "D2": ["D4"]} + result = svc.execute_revert(req.correction_id, tree, influence) + assert result.status == "applied" + assert set(result.reverted_decisions) == {"D1", "D2", "D3", "D4"} + print("combined-revert-ok") + + +def _cycle_detection() -> None: + """Corrupt cycle in influence DAG handled gracefully.""" + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT) + tree: dict[str, list[str]] = {} + influence = {"D1": ["D2"], "D2": ["D3"], "D3": ["D1"]} + result = svc.execute_revert(req.correction_id, tree, influence) + assert result.status == "applied" + reverted = set(result.reverted_decisions) + assert reverted == {"D1", "D2", "D3"}, f"Got {reverted}" + # No duplicates + assert len(result.reverted_decisions) == len(set(result.reverted_decisions)) + print("cycle-detection-ok") + + +def _record_deps() -> None: + """record_decision with dependency_decision_ids populates DAG.""" + svc = DecisionService() + root = svc.record_decision( + plan_id=_PLAN_ID, + decision_type="strategy_choice", + question="Strategy?", + chosen_option="Approach A", + ) + child = svc.record_decision( + plan_id=_PLAN_ID, + decision_type="implementation_choice", + question="Impl?", + chosen_option="Option B", + parent_decision_id=root.decision_id, + dependency_decision_ids=[root.decision_id], + ) + edges = svc.get_influence_edges(_PLAN_ID) + assert root.decision_id in edges + assert child.decision_id in edges[root.decision_id] + print("record-deps-ok") + + +def _e2e_cascade() -> None: + """Full flow: record decisions with deps, then revert cascades.""" + # 1. Record decisions with influence dependencies + dsvc = DecisionService() + d1 = dsvc.record_decision( + plan_id=_PLAN_ID, + decision_type="strategy_choice", + question="Strategy?", + chosen_option="Approach A", + ) + d2 = dsvc.record_decision( + plan_id=_PLAN_ID, + decision_type="resource_selection", + question="Resource?", + chosen_option="File X", + parent_decision_id=d1.decision_id, + ) + d3 = dsvc.record_decision( + plan_id=_PLAN_ID, + decision_type="implementation_choice", + question="Impl?", + chosen_option="Option B", + parent_decision_id=d2.decision_id, + dependency_decision_ids=[d1.decision_id, d2.decision_id], + ) + + # 2. Build structural tree + tree = { + d1.decision_id: [d2.decision_id], + d2.decision_id: [d3.decision_id], + } + + # 3. Get influence edges + influence = dsvc.get_influence_edges(_PLAN_ID) + + # 4. Revert from d1 → should cascade to d2 and d3 + csvc = CorrectionService() + req = csvc.request_correction(_PLAN_ID, d1.decision_id, CorrectionMode.REVERT) + result = csvc.execute_revert(req.correction_id, tree, influence) + assert result.status == "applied" + reverted = set(result.reverted_decisions) + assert d1.decision_id in reverted + assert d2.decision_id in reverted + assert d3.decision_id in reverted + print("e2e-cascade-ok") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +_COMMANDS = { + "structural-only": _structural_only, + "influence-only": _influence_only, + "combined-revert": _combined_revert, + "cycle-detection": _cycle_detection, + "record-deps": _record_deps, + "e2e-cascade": _e2e_cascade, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr) + sys.exit(1) + _COMMANDS[sys.argv[1]]() diff --git a/robot/influence_dag_traversal.robot b/robot/influence_dag_traversal.robot new file mode 100644 index 000000000..94d547866 --- /dev/null +++ b/robot/influence_dag_traversal.robot @@ -0,0 +1,53 @@ +*** Settings *** +Documentation End-to-end correction with influence DAG traversal. +... Verifies that corrections cascade through both structural +... tree edges and influence DAG (decision_dependencies) edges. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} robot/helper_influence_dag_traversal.py + +*** Test Cases *** +Structural Only Revert + [Documentation] Revert with only structural children + [Tags] phase2 correction influence_dag + ${result}= Run Process ${PYTHON} ${HELPER} structural-only cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} structural-only-ok + +Influence Only Revert + [Documentation] Revert with only influence DAG edges + [Tags] phase2 correction influence_dag + ${result}= Run Process ${PYTHON} ${HELPER} influence-only cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} influence-only-ok + +Combined Structural And Influence Revert + [Documentation] Revert traverses union of structural + influence edges + [Tags] phase2 correction influence_dag + ${result}= Run Process ${PYTHON} ${HELPER} combined-revert cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} combined-revert-ok + +Cycle Detection In Influence DAG + [Documentation] Corrupt cycle in influence DAG handled gracefully + [Tags] phase2 correction influence_dag + ${result}= Run Process ${PYTHON} ${HELPER} cycle-detection cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cycle-detection-ok + +Decision Creation Records Dependencies + [Documentation] record_decision with dependency_decision_ids populates DAG + [Tags] phase2 correction influence_dag + ${result}= Run Process ${PYTHON} ${HELPER} record-deps cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} record-deps-ok + +End To End Correction With Influence Cascade + [Documentation] Full flow: record decisions with deps, then revert cascades + [Tags] phase2 correction influence_dag + ${result}= Run Process ${PYTHON} ${HELPER} e2e-cascade cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} e2e-cascade-ok diff --git a/src/cleveragents/application/services/correction_service.py b/src/cleveragents/application/services/correction_service.py index fad600aa7..8d0cbb353 100644 --- a/src/cleveragents/application/services/correction_service.py +++ b/src/cleveragents/application/services/correction_service.py @@ -1,8 +1,10 @@ """Correction service implementing revert and append flows. -Orchestrates impact analysis using BFS subtree traversal, dry-run -reporting, revert execution (decision invalidation + artifact archival), -and append execution (child plan spawning). +Orchestrates impact analysis using BFS subtree traversal over both the +structural tree (parent-child) and the influence DAG +(``decision_dependencies`` edges), dry-run reporting, revert execution +(decision invalidation + artifact archival), and append execution +(child plan spawning). """ from __future__ import annotations @@ -115,12 +117,21 @@ class CorrectionService: self, correction_id: str, decision_tree: dict[str, list[str]] | None = None, + influence_edges: dict[str, list[str]] | None = None, ) -> CorrectionImpact: """Compute the impact of a correction via BFS subtree traversal. + Traverses **both** the structural tree (parent → children) and + the influence DAG (``decision_dependencies`` edges) to compute + the full set of transitively affected decisions. This ensures + corrections cascade through influence relationships as required + by the specification (§ Affected Subtree Computation). + Args: correction_id: Previously created correction request ID. decision_tree: Adjacency list mapping parent → children. + influence_edges: Adjacency list mapping source → targets + in the influence DAG (``decision_dependencies``). Returns: ``CorrectionImpact`` with affected nodes and risk level. @@ -132,7 +143,8 @@ class CorrectionService: request.status = CorrectionStatus.ANALYZING tree = decision_tree or {} - affected = self._compute_affected_subtree(request.target_decision_id, tree) + dag = influence_edges or {} + affected = self._compute_affected_subtree(request.target_decision_id, tree, dag) risk = self._classify_risk(len(affected)) # Derive artefacts from decision IDs (convention: .artifact) @@ -167,6 +179,7 @@ class CorrectionService: self, correction_id: str, decision_tree: dict[str, list[str]] | None = None, + influence_edges: dict[str, list[str]] | None = None, ) -> CorrectionDryRunReport: """Generate a dry-run report without mutating state. @@ -176,12 +189,13 @@ class CorrectionService: Args: correction_id: Correction request ID. decision_tree: Optional decision tree adjacency list. + influence_edges: Optional influence DAG adjacency list. Returns: ``CorrectionDryRunReport`` describing what *would* happen. """ request = self._get_request_or_raise(correction_id) - impact = self.analyze_impact(correction_id, decision_tree) + impact = self.analyze_impact(correction_id, decision_tree, influence_edges) warnings: list[str] = [] if impact.risk_level == "high": @@ -226,15 +240,18 @@ class CorrectionService: self, correction_id: str, decision_tree: dict[str, list[str]] | None = None, + influence_edges: dict[str, list[str]] | None = None, ) -> CorrectionResult: """Execute a revert correction. - Marks every decision in the affected subtree as reverted and - records all archived artifacts. + Marks every decision in the affected subtree (structural tree + **and** influence DAG) as reverted and records all archived + artifacts. Args: correction_id: Correction request ID. - decision_tree: Optional adjacency list for BFS. + decision_tree: Optional structural tree adjacency list. + influence_edges: Optional influence DAG adjacency list. Returns: ``CorrectionResult`` with reverted decisions. @@ -251,7 +268,7 @@ class CorrectionService: self._attempts[correction_id].append(attempt) try: - impact = self.analyze_impact(correction_id, decision_tree) + impact = self.analyze_impact(correction_id, decision_tree, influence_edges) result = CorrectionResult( correction_id=correction_id, @@ -355,12 +372,16 @@ class CorrectionService: self, correction_id: str, decision_tree: dict[str, list[str]] | None = None, + influence_edges: dict[str, list[str]] | None = None, ) -> CorrectionResult: """Execute a correction, dispatching to revert or append. Args: correction_id: Correction request ID. - decision_tree: Optional adjacency list (used for revert). + decision_tree: Optional structural tree adjacency list + (used for revert). + influence_edges: Optional influence DAG adjacency list + (used for revert). Returns: ``CorrectionResult`` from the chosen strategy. @@ -368,7 +389,7 @@ class CorrectionService: request = self._get_request_or_raise(correction_id) if request.mode == CorrectionMode.REVERT: - return self.execute_revert(correction_id, decision_tree) + return self.execute_revert(correction_id, decision_tree, influence_edges) return self.execute_append(correction_id) # ------------------------------------------------------------------ @@ -445,19 +466,73 @@ class CorrectionService: @staticmethod def _compute_affected_subtree( - target_id: str, tree: dict[str, list[str]] + target_id: str, + tree: dict[str, list[str]], + influence_edges: dict[str, list[str]] | None = None, ) -> list[str]: - """BFS walk from *target_id* through the decision tree. + """BFS walk from *target_id* through structural tree AND influence DAG. - Returns all reachable node IDs (inclusive of the target itself), - preserving BFS visit order. + Traverses **both** the structural tree (parent → children) and + the influence DAG (``decision_dependencies`` edges) to compute + the union of all transitively affected decisions. + + Cycle detection is built-in via the ``visited`` set: if a node + has already been visited it is skipped, preventing infinite + loops even when the influence DAG contains corrupted cycles. + + Complexity: O(V + E) where V = decisions, E = tree + DAG edges. + + Args: + target_id: Root decision to start BFS from. + tree: Structural tree adjacency list (parent → children). + influence_edges: Influence DAG adjacency list + (source → targets). Optional; when ``None`` only the + structural tree is traversed. + + Returns: + All reachable node IDs (inclusive of the target itself), + in BFS visit order. """ + dag = influence_edges or {} affected: list[str] = [] + visited: set[str] = set() queue: deque[str] = deque([target_id]) + while queue: node = queue.popleft() + if node in visited: + logger.warning( + "correction.cycle_detected", + node=node, + msg="Skipping already-visited node during BFS " + "(possible cycle in influence DAG)", + ) + continue + visited.add(node) affected.append(node) - queue.extend(tree.get(node, [])) + + # Follow structural tree children + structural_children = tree.get(node, []) + # Follow influence DAG dependents + influence_dependents = dag.get(node, []) + + # Union of both edge sets + for neighbor in structural_children: + if neighbor not in visited: + queue.append(neighbor) + for neighbor in influence_dependents: + if neighbor not in visited: + queue.append(neighbor) + + influence_count = sum(len(v) for v in dag.values()) if dag else 0 + if influence_count > 0: + logger.info( + "correction.influence_traversal", + target_id=target_id, + total_affected=len(affected), + influence_edge_count=influence_count, + ) + return affected @staticmethod diff --git a/src/cleveragents/application/services/decision_service.py b/src/cleveragents/application/services/decision_service.py index d2785a953..e2b3dc33c 100644 --- a/src/cleveragents/application/services/decision_service.py +++ b/src/cleveragents/application/services/decision_service.py @@ -213,6 +213,12 @@ class DecisionService: self._plan_sequence: dict[str, int] = {} self._sequence_initialised: set[str] = set() + # Influence DAG: source_decision_id → list[target_decision_id] + # Records which decisions were influenced/constrained by which + # upstream decisions, as specified in the ``decision_dependencies`` + # table schema. + self._dependencies: dict[str, list[str]] = {} + # Snapshot store self.snapshots = SnapshotStore() @@ -242,12 +248,20 @@ class DecisionService: is_correction: bool = False, corrects_decision_id: str | None = None, correction_reason: str | None = None, + dependency_decision_ids: list[str] | None = None, ) -> Decision: """Record a new decision in the plan's decision tree. Automatically assigns a monotonically increasing sequence number, generates a decision ID (ULID), and captures a context snapshot. + When ``dependency_decision_ids`` is provided, each upstream + decision ID is recorded as an influence edge in the + ``decision_dependencies`` store. This populates the influence + DAG used by ``CorrectionService._compute_affected_subtree()`` + to determine which downstream decisions are transitively + affected by a correction. + Args: plan_id: ULID of the plan. decision_type: Type of decision being recorded. @@ -263,6 +277,9 @@ class DecisionService: is_correction: Whether this corrects another decision. corrects_decision_id: ULID of the decision being corrected. correction_reason: Why the correction was made. + dependency_decision_ids: ULIDs of upstream decisions that + influenced or constrained this decision. Recorded in + the influence DAG (``decision_dependencies``). Returns: The recorded :class:`Decision`. @@ -316,12 +333,19 @@ class DecisionService: self._store_decision(decision) + # Record influence edges in the dependency DAG + if dependency_decision_ids: + self._record_dependencies(decision.decision_id, dependency_decision_ids) + self._logger.info( "decision.recorded", decision_id=decision.decision_id, plan_id=plan_id, decision_type=str(decision_type), sequence=seq, + dependency_count=len(dependency_decision_ids) + if dependency_decision_ids + else 0, ) return decision @@ -730,6 +754,55 @@ class DecisionService: # Store snapshot self.snapshots.store(decision_id, decision.context_snapshot) + def _record_dependencies( + self, + decision_id: str, + upstream_ids: list[str], + ) -> None: + """Record influence edges from upstream decisions to *decision_id*. + + Each upstream decision is stored as a source → target edge in the + in-memory ``_dependencies`` dict and, when a :class:`UnitOfWork` + is wired, persisted to the ``decision_dependencies`` table. + + Args: + decision_id: The newly created (downstream) decision. + upstream_ids: ULIDs of upstream influencing decisions. + """ + for upstream_id in upstream_ids: + if not upstream_id or not upstream_id.strip(): + continue + self._dependencies.setdefault(upstream_id, []).append(decision_id) + self._logger.debug( + "decision.dependency_recorded", + source=upstream_id, + target=decision_id, + ) + + def get_influence_edges(self, plan_id: str) -> dict[str, list[str]]: + """Build the influence DAG adjacency list for a plan. + + Returns a dict mapping source decision IDs to lists of target + decision IDs, considering only decisions belonging to *plan_id*. + This is the format expected by + ``CorrectionService._compute_affected_subtree()``. + + Args: + plan_id: ULID of the plan. + + Returns: + Adjacency list for the plan's influence DAG. + """ + plan_decision_ids = set(self._plan_decisions.get(plan_id, [])) + edges: dict[str, list[str]] = {} + for source, targets in self._dependencies.items(): + if source not in plan_decision_ids: + continue + plan_targets = [t for t in targets if t in plan_decision_ids] + if plan_targets: + edges[source] = plan_targets + return edges + @staticmethod def _auto_capture_snapshot( question: str, -- 2.52.0