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

@@ -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

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)

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"

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}"

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]]()

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

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: <id>.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

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,