fix(decision_service): persist decision dependency edges to database

Implement persistence of decision dependency edges to the decision_dependencies
table when DecisionService.record_decision() is called with dependency_decision_ids.

Changes:
- Add DecisionRepository.record_dependency() method to persist edges
- Add DecisionRepository.get_influence_edges() to retrieve DAG from database
- Update DecisionService._record_dependencies() to call repository when persisted
- Update DecisionService.get_influence_edges() to fetch from database in persisted mode
- Add comprehensive BDD tests for dependency persistence and influence DAG

This fixes the issue where decision dependency edges were only stored in-memory
and lost on service restart, breaking the CorrectionService._compute_affected_subtree()
functionality.

ISSUES CLOSED: #7926

features/decision_dependency_persistence.feature

@@ -0,0 +1,151 @@
+Feature: Decision dependency persistence and influence DAG
+  As a developer
+  I want decision dependencies to persist to the database
+  So that the influence DAG survives service restarts and enables correction workflows
+
+  Background:
+    Given a fresh database for decision dependencies
+    And a prerequisite action "local/decision-action" exists for dependencies
+    And a prerequisite plan exists for dependencies
+
+  # ------------------------------------------------------------------
+  # Basic dependency recording
+  # ------------------------------------------------------------------
+
+  Scenario: Record a single dependency edge
+    Given a persisted root decision for dependencies
+    And a persisted child decision for dependencies
+    When I record a dependency from the root to the child
+    Then the dependency should be persisted to the database
+    And the dependency should have relationship_type "influences"
+
+  Scenario: Record multiple dependencies for a single target
+    Given a persisted root decision for dependencies
+    And 2 persisted sibling decisions for dependencies
+    When I record dependencies from the root to both siblings
+    Then both dependencies should be persisted to the database
+
+  Scenario: Record dependencies with custom relationship type
+    Given a persisted root decision for dependencies
+    And a persisted child decision for dependencies
+    When I record a dependency with relationship_type "constrains"
+    Then the dependency should be persisted with relationship_type "constrains"
+
+  # ------------------------------------------------------------------
+  # Influence DAG retrieval
+  # ------------------------------------------------------------------
+
+  Scenario: Get influence edges returns empty dict for plan with no dependencies
+    Given a persisted root decision for dependencies
+    When I get influence edges for the plan
+    Then the influence edges should be empty
+
+  Scenario: Get influence edges returns single edge
+    Given a persisted root decision for dependencies
+    And a persisted child decision for dependencies
+    And a recorded dependency from root to child
+    When I get influence edges for the plan
+    Then the influence edges should contain 1 source
+    And the influence edges should map root to [child]
+
+  Scenario: Get influence edges returns multiple targets for single source
+    Given a persisted root decision for dependencies
+    And 2 persisted sibling decisions for dependencies
+    And recorded dependencies from root to both siblings
+    When I get influence edges for the plan
+    Then the influence edges should contain 1 source
+    And the influence edges should map root to [sibling1, sibling2]
+
+  Scenario: Get influence edges returns multiple sources
+    Given 2 persisted root decisions for dependencies
+    And a persisted child decision for dependencies
+    And recorded dependencies from both roots to the child
+    When I get influence edges for the plan
+    Then the influence edges should contain 2 sources
+
+  # ------------------------------------------------------------------
+  # Persistence across service restart
+  # ------------------------------------------------------------------
+
+  Scenario: Dependencies persist across DecisionService restart
+    Given a persisted root decision for dependencies
+    And a persisted child decision for dependencies
+    And a recorded dependency from root to child
+    When I create a new DecisionService instance with the same database
+    Then the new service should retrieve the same dependency edges
+
+  Scenario: Influence DAG is complete after service restart
+    Given a persisted decision tree with 3 levels for dependencies
+    And recorded dependencies forming a DAG
+    When I create a new DecisionService instance with the same database
+    Then the new service should retrieve all dependency edges
+
+  # ------------------------------------------------------------------
+  # Filtering by plan
+  # ------------------------------------------------------------------
+
+  Scenario: Get influence edges filters by plan ID
+    Given 2 persisted plans for dependencies
+    And a decision tree in plan 1 with dependencies
+    And a decision tree in plan 2 with dependencies
+    When I get influence edges for plan 1
+    Then the influence edges should only include decisions from plan 1
+
+  Scenario: Get influence edges excludes cross-plan dependencies
+    Given 2 persisted plans for dependencies
+    And a decision in plan 1 and a decision in plan 2
+    When I try to record a cross-plan dependency
+    Then the cross-plan dependency should not appear in influence edges
+
+  # ------------------------------------------------------------------
+  # Integration with DecisionService.record_decision
+  # ------------------------------------------------------------------
+
+  Scenario: record_decision with dependency_decision_ids persists edges
+    Given a persisted root decision for dependencies
+    When I record a new decision with dependency_decision_ids=[root]
+    Then the dependency should be persisted to the database
+    And get_influence_edges should return the dependency
+
+  Scenario: record_decision with multiple dependencies persists all edges
+    Given 2 persisted root decisions for dependencies
+    When I record a new decision with dependency_decision_ids=[root1, root2]
+    Then both dependencies should be persisted to the database
+    And get_influence_edges should return both dependencies
+
+  Scenario: record_decision with empty dependency_decision_ids creates no edges
+    Given a persisted root decision for dependencies
+    When I record a new decision with dependency_decision_ids=[]
+    Then no dependencies should be created
+    And get_influence_edges should be empty
+
+  # ------------------------------------------------------------------
+  # Constraint enforcement
+  # ------------------------------------------------------------------
+
+  Scenario: Cannot record self-loop dependency
+    Given a persisted root decision for dependencies
+    When I try to record a self-loop dependency
+    Then a database constraint error should be raised
+
+  Scenario: Duplicate dependency edges are handled gracefully
+    Given a persisted root decision for dependencies
+    And a persisted child decision for dependencies
+    And a recorded dependency from root to child
+    When I try to record the same dependency again
+    Then the operation should complete without error
+
+  # ------------------------------------------------------------------
+  # Correction workflow integration
+  # ------------------------------------------------------------------
+
+  Scenario: Influence edges support correction workflow
+    Given a persisted decision tree with dependencies
+    When I compute affected subtree using influence edges
+    Then the affected subtree should include all transitive dependents
+
+  Scenario: Correction decision inherits dependency information
+    Given a persisted root decision for dependencies
+    And a persisted child decision depending on the root
+    When I record a correction decision with the same dependencies
+    Then the correction decision should have the same influence edges

features/steps/decision_dependency_persistence_steps.py

@@ -0,0 +1,672 @@
+"""Step definitions for decision_dependency_persistence.feature.
+
+Tests decision dependency persistence, influence DAG retrieval,
+and integration with DecisionService.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from ulid import ULID
+
+from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.core.exceptions import DatabaseError
+from cleveragents.domain.models.core.action import Action, ActionState
+from cleveragents.domain.models.core.decision import Decision, DecisionType
+from cleveragents.domain.models.core.plan import (
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    DecisionRepository,
+    LifecyclePlanRepository,
+)
+from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_PLAN_ID_1 = "01HV000000000000000000DP01"
+_PLAN_ID_2 = "01HV000000000000000000DP02"
+
+
+def _setup_db(context: Context) -> None:
+    """Create an in-memory SQLite DB and attach repos."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    context._ddp_engine = engine
+    context._ddp_session = session
+    context._ddp_factory = lambda: session
+    context._ddp_repo = DecisionRepository(session_factory=context._ddp_factory)
+    context._ddp_action_repo = ActionRepository(session_factory=context._ddp_factory)
+    context._ddp_plan_repo = LifecyclePlanRepository(
+        session_factory=context._ddp_factory,
+    )
+    context._ddp_uow = UnitOfWork(
+        engine=engine,
+        session_factory=context._ddp_factory,
+    )
+
+
+def _make_decision(
+    plan_id: str = _PLAN_ID_1,
+    parent_decision_id: str | None = None,
+    sequence_number: int = 0,
+    decision_type: DecisionType = DecisionType.PROMPT_DEFINITION,
+    question: str = "What approach should we take?",
+    chosen_option: str = "Build a REST API",
+) -> Decision:
+    """Create a Decision domain model."""
+    return Decision(
+        plan_id=plan_id,
+        sequence_number=sequence_number,
+        decision_type=decision_type,
+        question=question,
+        chosen_option=chosen_option,
+        parent_decision_id=parent_decision_id,
+    )
+
+
+def _create_prerequisite_action(context: Context, action_name: str) -> None:
+    """Create a prerequisite action for testing."""
+    ns = NamespacedName.parse(action_name)
+    action = Action(
+        namespaced_name=ns,
+        description="Prerequisite action for dependency tests",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        state=ActionState.AVAILABLE,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+    )
+    context._ddp_action_repo.create(action)
+    context._ddp_session.commit()
+
+
+def _create_prerequisite_plan(context: Context, plan_id: str) -> None:
+    """Create a prerequisite plan for testing."""
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    plan = Plan(
+        identity=PlanIdentity(plan_id=plan_id, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="decision-plan"),
+        action_name="local/decision-action",
+        description="Decision test plan",
+        definition_of_done="All assertions pass",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.PROCESSING,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="test-dependency",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+    context._ddp_plan_repo.create(plan)
+    context._ddp_session.commit()
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a fresh database for decision dependencies")
+def step_fresh_db(context: Context) -> None:
+    """Set up a fresh in-memory database."""
+    _setup_db(context)
+
+
+@given('a prerequisite action "{action_name}" exists for dependencies')
+def step_prereq_action(context: Context, action_name: str) -> None:
+    """Create a prerequisite action."""
+    _create_prerequisite_action(context, action_name)
+
+
+@given("a prerequisite plan exists for dependencies")
+def step_prereq_plan(context: Context) -> None:
+    """Create a prerequisite plan."""
+    _create_prerequisite_plan(context, _PLAN_ID_1)
+
+
+# ---------------------------------------------------------------------------
+# Decision creation
+# ---------------------------------------------------------------------------
+
+
+@given("a persisted root decision for dependencies")
+def step_persisted_root_dep(context: Context) -> None:
+    """Create and persist a root decision."""
+    d = _make_decision(plan_id=_PLAN_ID_1, sequence_number=0)
+    context._ddp_repo.create(d)
+    context._ddp_session.commit()
+    context._ddp_root_id = d.decision_id
+    context._ddp_root_decision = d
+
+
+@given("a persisted child decision for dependencies")
+def step_persisted_child_dep(context: Context) -> None:
+    """Create and persist a child decision."""
+    d = _make_decision(
+        plan_id=_PLAN_ID_1,
+        parent_decision_id=context._ddp_root_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context._ddp_repo.create(d)
+    context._ddp_session.commit()
+    context._ddp_child_id = d.decision_id
+    context._ddp_child_decision = d
+
+
+@given("{count:d} persisted sibling decisions for dependencies")
+def step_persisted_siblings_dep(context: Context, count: int) -> None:
+    """Create and persist multiple sibling decisions."""
+    context._ddp_sibling_ids = []
+    for i in range(count):
+        d = _make_decision(
+            plan_id=_PLAN_ID_1,
+            parent_decision_id=context._ddp_root_id,
+            sequence_number=i + 1,
+            decision_type=DecisionType.STRATEGY_CHOICE,
+        )
+        context._ddp_repo.create(d)
+        context._ddp_sibling_ids.append(d.decision_id)
+    context._ddp_session.commit()
+
+
+@given("{count:d} persisted root decisions for dependencies")
+def step_persisted_roots_dep(context: Context, count: int) -> None:
+    """Create and persist multiple root decisions."""
+    context._ddp_root_ids = []
+    for i in range(count):
+        d = _make_decision(
+            plan_id=_PLAN_ID_1,
+            sequence_number=i,
+        )
+        context._ddp_repo.create(d)
+        context._ddp_root_ids.append(d.decision_id)
+    context._ddp_session.commit()
+
+
+@given("{count:d} persisted plans for dependencies")
+def step_persisted_plans_dep(context: Context, count: int) -> None:
+    """Create and persist multiple plans."""
+    context._ddp_plan_ids = [_PLAN_ID_1]
+    if count > 1:
+        _create_prerequisite_plan(context, _PLAN_ID_2)
+        context._ddp_plan_ids.append(_PLAN_ID_2)
+
+
+@given("a decision tree in plan 1 with dependencies")
+def step_decision_tree_plan1_dep(context: Context) -> None:
+    """Create a decision tree in plan 1 with dependencies."""
+    # Create root
+    root = _make_decision(plan_id=_PLAN_ID_1, sequence_number=0)
+    context._ddp_repo.create(root)
+    context._ddp_session.commit()
+    root_id = root.decision_id
+
+    # Create child
+    child = _make_decision(
+        plan_id=_PLAN_ID_1,
+        parent_decision_id=root_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context._ddp_repo.create(child)
+    context._ddp_session.commit()
+
+    # Record dependency
+    context._ddp_repo.record_dependency(root_id, child.decision_id)
+    context._ddp_session.commit()
+
+    context._ddp_plan1_root_id = root_id
+    context._ddp_plan1_child_id = child.decision_id
+
+
+@given("a decision tree in plan 2 with dependencies")
+def step_decision_tree_plan2_dep(context: Context) -> None:
+    """Create a decision tree in plan 2 with dependencies."""
+    # Create root
+    root = _make_decision(plan_id=_PLAN_ID_2, sequence_number=0)
+    context._ddp_repo.create(root)
+    context._ddp_session.commit()
+    root_id = root.decision_id
+
+    # Create child
+    child = _make_decision(
+        plan_id=_PLAN_ID_2,
+        parent_decision_id=root_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context._ddp_repo.create(child)
+    context._ddp_session.commit()
+
+    # Record dependency
+    context._ddp_repo.record_dependency(root_id, child.decision_id)
+    context._ddp_session.commit()
+
+    context._ddp_plan2_root_id = root_id
+    context._ddp_plan2_child_id = child.decision_id
+
+
+@given("a decision in plan 1 and a decision in plan 2")
+def step_cross_plan_decisions_dep(context: Context) -> None:
+    """Create decisions in different plans."""
+    d1 = _make_decision(plan_id=_PLAN_ID_1, sequence_number=0)
+    context._ddp_repo.create(d1)
+
+    d2 = _make_decision(plan_id=_PLAN_ID_2, sequence_number=0)
+    context._ddp_repo.create(d2)
+    context._ddp_session.commit()
+
+    context._ddp_cross_plan_id1 = d1.decision_id
+    context._ddp_cross_plan_id2 = d2.decision_id
+
+
+@given("a persisted decision tree with 3 levels for dependencies")
+def step_decision_tree_3levels_dep(context: Context) -> None:
+    """Create a 3-level decision tree."""
+    # Level 0: root
+    root = _make_decision(plan_id=_PLAN_ID_1, sequence_number=0)
+    context._ddp_repo.create(root)
+    context._ddp_session.commit()
+    root_id = root.decision_id
+
+    # Level 1: child
+    child = _make_decision(
+        plan_id=_PLAN_ID_1,
+        parent_decision_id=root_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context._ddp_repo.create(child)
+    context._ddp_session.commit()
+    child_id = child.decision_id
+
+    # Level 2: grandchild
+    grandchild = _make_decision(
+        plan_id=_PLAN_ID_1,
+        parent_decision_id=child_id,
+        sequence_number=2,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+    )
+    context._ddp_repo.create(grandchild)
+    context._ddp_session.commit()
+    grandchild_id = grandchild.decision_id
+
+    context._ddp_tree_root_id = root_id
+    context._ddp_tree_child_id = child_id
+    context._ddp_tree_grandchild_id = grandchild_id
+
+
+@given("recorded dependencies forming a DAG")
+def step_recorded_dag_dep(context: Context) -> None:
+    """Record dependencies forming a DAG."""
+    context._ddp_repo.record_dependency(
+        context._ddp_tree_root_id,
+        context._ddp_tree_child_id,
+    )
+    context._ddp_repo.record_dependency(
+        context._ddp_tree_child_id,
+        context._ddp_tree_grandchild_id,
+    )
+    context._ddp_session.commit()
+
+
+# ---------------------------------------------------------------------------
+# Dependency recording
+# ---------------------------------------------------------------------------
+
+
+@when("I record a dependency from the root to the child")
+def step_record_dependency_root_child(context: Context) -> None:
+    """Record a dependency from root to child."""
+    context._ddp_repo.record_dependency(
+        context._ddp_root_id,
+        context._ddp_child_id,
+    )
+    context._ddp_session.commit()
+
+
+@when("I record dependencies from the root to both siblings")
+def step_record_dependencies_root_siblings(context: Context) -> None:
+    """Record dependencies from root to all siblings."""
+    for sibling_id in context._ddp_sibling_ids:
+        context._ddp_repo.record_dependency(
+            context._ddp_root_id,
+            sibling_id,
+        )
+    context._ddp_session.commit()
+
+
+@when("I record a dependency with relationship_type {rel_type}")
+def step_record_dependency_custom_type(context: Context, rel_type: str) -> None:
+    """Record a dependency with custom relationship type."""
+    context._ddp_repo.record_dependency(
+        context._ddp_root_id,
+        context._ddp_child_id,
+        relationship_type=rel_type,
+    )
+    context._ddp_session.commit()
+    context._ddp_custom_rel_type = rel_type
+
+
+@when("I try to record a self-loop dependency")
+def step_try_self_loop_dependency(context: Context) -> None:
+    """Try to record a self-loop dependency."""
+    try:
+        context._ddp_repo.record_dependency(
+            context._ddp_root_id,
+            context._ddp_root_id,
+        )
+        context._ddp_session.commit()
+        context._ddp_self_loop_error = None
+    except DatabaseError as e:
+        context._ddp_self_loop_error = e
+
+
+@when("I try to record the same dependency again")
+def step_try_duplicate_dependency(context: Context) -> None:
+    """Try to record the same dependency again."""
+    try:
+        context._ddp_repo.record_dependency(
+            context._ddp_root_id,
+            context._ddp_child_id,
+        )
+        context._ddp_session.commit()
+        context._ddp_duplicate_error = None
+    except DatabaseError as e:
+        context._ddp_duplicate_error = e
+
+
+@when("I try to record a cross-plan dependency")
+def step_try_cross_plan_dependency(context: Context) -> None:
+    """Try to record a cross-plan dependency."""
+    try:
+        context._ddp_repo.record_dependency(
+            context._ddp_cross_plan_id1,
+            context._ddp_cross_plan_id2,
+        )
+        context._ddp_session.commit()
+        context._ddp_cross_plan_error = None
+    except DatabaseError as e:
+        context._ddp_cross_plan_error = e
+
+
+# ---------------------------------------------------------------------------
+# Influence edges retrieval
+# ---------------------------------------------------------------------------
+
+
+@when("I get influence edges for the plan")
+def step_get_influence_edges(context: Context) -> None:
+    """Get influence edges for the plan."""
+    context._ddp_influence_edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+
+
+@when("I get influence edges for plan 1")
+def step_get_influence_edges_plan1(context: Context) -> None:
+    """Get influence edges for plan 1."""
+    context._ddp_influence_edges_plan1 = context._ddp_repo.get_influence_edges(
+        _PLAN_ID_1
+    )
+
+
+@when("I get influence edges for plan 2")
+def step_get_influence_edges_plan2(context: Context) -> None:
+    """Get influence edges for plan 2."""
+    context._ddp_influence_edges_plan2 = context._ddp_repo.get_influence_edges(
+        _PLAN_ID_2
+    )
+
+
+# ---------------------------------------------------------------------------
+# DecisionService integration
+# ---------------------------------------------------------------------------
+
+
+@when("I create a new DecisionService instance with the same database")
+def step_new_decision_service(context: Context) -> None:
+    """Create a new DecisionService with the same database."""
+    context._ddp_new_service = DecisionService(
+        unit_of_work=context._ddp_uow,
+    )
+
+
+@when("I record a new decision with dependency_decision_ids={deps}")
+def step_record_decision_with_deps(context: Context, deps: str) -> None:
+    """Record a new decision with dependencies."""
+    # Parse the deps string (e.g., "[root]" or "[root1, root2]")
+    service = DecisionService(unit_of_work=context._ddp_uow)
+
+    if deps == "[]":
+        dep_ids = []
+    elif deps == "[root]":
+        dep_ids = [context._ddp_root_id]
+    elif deps == "[root1, root2]":
+        dep_ids = context._ddp_root_ids
+    else:
+        dep_ids = []
+
+    decision = service.record_decision(
+        plan_id=_PLAN_ID_1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        question="What strategy?",
+        chosen_option="Strategy A",
+        dependency_decision_ids=dep_ids,
+    )
+    context._ddp_new_decision_id = decision.decision_id
+    context._ddp_new_decision = decision
+
+
+# ---------------------------------------------------------------------------
+# Assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the dependency should be persisted to the database")
+def step_assert_dependency_persisted(context: Context) -> None:
+    """Assert that the dependency was persisted."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    assert context._ddp_root_id in edges, "Root should be in edges"
+    assert context._ddp_child_id in edges[context._ddp_root_id], (
+        "Child should be in root's targets"
+    )
+
+
+@then("the dependency should have relationship_type {rel_type}")
+def step_assert_dependency_rel_type(context: Context, rel_type: str) -> None:
+    """Assert that the dependency has the correct relationship type."""
+    # Note: The current implementation doesn't return relationship_type,
+    # but we can verify it was stored by checking the database directly
+    from cleveragents.infrastructure.database.models import DecisionDependencyModel
+
+    session = context._ddp_factory()
+    dep = (
+        session.query(DecisionDependencyModel)
+        .filter_by(
+            source_decision_id=context._ddp_root_id,
+            target_decision_id=context._ddp_child_id,
+        )
+        .first()
+    )
+    assert dep is not None, "Dependency should exist"
+    assert dep.relationship_type == rel_type, (
+        f"Relationship type should be {rel_type}, got {dep.relationship_type}"
+    )
+
+
+@then("the dependency should be persisted with relationship_type {rel_type}")
+def step_assert_dependency_persisted_with_type(context: Context, rel_type: str) -> None:
+    """Assert that the dependency was persisted with the correct type."""
+    from cleveragents.infrastructure.database.models import DecisionDependencyModel
+
+    session = context._ddp_factory()
+    dep = (
+        session.query(DecisionDependencyModel)
+        .filter_by(
+            source_decision_id=context._ddp_root_id,
+            target_decision_id=context._ddp_child_id,
+        )
+        .first()
+    )
+    assert dep is not None, "Dependency should exist"
+    assert dep.relationship_type == rel_type, (
+        f"Relationship type should be {rel_type}, got {dep.relationship_type}"
+    )
+
+
+@then("both dependencies should be persisted to the database")
+def step_assert_both_dependencies_persisted(context: Context) -> None:
+    """Assert that both dependencies were persisted."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    assert context._ddp_root_id in edges, "Root should be in edges"
+    assert len(edges[context._ddp_root_id]) == 2, "Root should have 2 targets"
+    for sibling_id in context._ddp_sibling_ids:
+        assert sibling_id in edges[context._ddp_root_id], (
+            f"Sibling {sibling_id} should be in root's targets"
+        )
+
+
+@then("the influence edges should be empty")
+def step_assert_influence_edges_empty(context: Context) -> None:
+    """Assert that influence edges are empty."""
+    assert context._ddp_influence_edges == {}, "Influence edges should be empty"
+
+
+@then("the influence edges should contain {count:d} source")
+def step_assert_influence_edges_source_count(context: Context, count: int) -> None:
+    """Assert the number of sources in influence edges."""
+    assert len(context._ddp_influence_edges) == count, (
+        f"Should have {count} source(s), got {len(context._ddp_influence_edges)}"
+    )
+
+
+@then("the influence edges should map root to {targets}")
+def step_assert_influence_edges_mapping(context: Context, targets: str) -> None:
+    """Assert that influence edges map root to the specified targets."""
+    # Parse targets string (e.g., "[child]" or "[sibling1, sibling2]")
+    if targets == "[child]":
+        expected = [context._ddp_child_id]
+    elif targets == "[sibling1, sibling2]":
+        expected = context._ddp_sibling_ids
+    else:
+        expected = []
+
+    assert context._ddp_root_id in context._ddp_influence_edges, (
+        "Root should be in edges"
+    )
+    actual = context._ddp_influence_edges[context._ddp_root_id]
+    assert set(actual) == set(expected), f"Expected {expected}, got {actual}"
+
+
+@then("the influence edges should only include decisions from plan 1")
+def step_assert_influence_edges_plan1_only(context: Context) -> None:
+    """Assert that influence edges only include plan 1 decisions."""
+    edges = context._ddp_influence_edges_plan1
+    plan1_ids = {
+        context._ddp_plan1_root_id,
+        context._ddp_plan1_child_id,
+    }
+    for source, targets in edges.items():
+        assert source in plan1_ids, f"Source {source} should be in plan 1"
+        for target in targets:
+            assert target in plan1_ids, f"Target {target} should be in plan 1"
+
+
+@then("the cross-plan dependency should not appear in influence edges")
+def step_assert_no_cross_plan_dependency(context: Context) -> None:
+    """Assert that cross-plan dependencies don't appear in edges."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    # The cross-plan dependency should not appear because the target is in a different plan
+    if context._ddp_cross_plan_id1 in edges:
+        assert context._ddp_cross_plan_id2 not in edges[context._ddp_cross_plan_id1], (
+            "Cross-plan dependency should not appear"
+        )
+
+
+@then("the new service should retrieve the same dependency edges")
+def step_assert_new_service_same_edges(context: Context) -> None:
+    """Assert that a new service retrieves the same edges."""
+    new_edges = context._ddp_new_service.get_influence_edges(_PLAN_ID_1)
+    assert context._ddp_root_id in new_edges, "Root should be in new edges"
+    assert context._ddp_child_id in new_edges[context._ddp_root_id], (
+        "Child should be in new root's targets"
+    )
+
+
+@then("the new service should retrieve all dependency edges")
+def step_assert_new_service_all_edges(context: Context) -> None:
+    """Assert that a new service retrieves all edges."""
+    new_edges = context._ddp_new_service.get_influence_edges(_PLAN_ID_1)
+    assert len(new_edges) == 2, f"Should have 2 sources, got {len(new_edges)}"
+    assert context._ddp_tree_root_id in new_edges, "Root should be in edges"
+    assert context._ddp_tree_child_id in new_edges, "Child should be in edges"
+
+
+@then("a database constraint error should be raised")
+def step_assert_constraint_error(context: Context) -> None:
+    """Assert that a constraint error was raised."""
+    assert context._ddp_self_loop_error is not None, (
+        "Should have raised a constraint error"
+    )
+
+
+@then("the operation should complete without error")
+def step_assert_no_error(context: Context) -> None:
+    """Assert that the operation completed without error."""
+    assert context._ddp_duplicate_error is None, "Should not have raised an error"
+
+
+@then("no dependencies should be created")
+def step_assert_no_dependencies_created(context: Context) -> None:
+    """Assert that no dependencies were created."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    assert edges == {}, "No dependencies should be created"
+
+
+@then("get_influence_edges should return the dependency")
+def step_assert_get_influence_edges_returns_dep(context: Context) -> None:
+    """Assert that get_influence_edges returns the dependency."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    assert context._ddp_root_id in edges, "Root should be in edges"
+    assert context._ddp_new_decision_id in edges[context._ddp_root_id], (
+        "New decision should be in root's targets"
+    )
+
+
+@then("get_influence_edges should return both dependencies")
+def step_assert_get_influence_edges_returns_both(context: Context) -> None:
+    """Assert that get_influence_edges returns both dependencies."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    for root_id in context._ddp_root_ids:
+        assert root_id in edges, f"Root {root_id} should be in edges"
+        assert context._ddp_new_decision_id in edges[root_id], (
+            f"New decision should be in {root_id}'s targets"
+        )
+
+
+@then("get_influence_edges should be empty")
+def step_assert_get_influence_edges_empty(context: Context) -> None:
+    """Assert that get_influence_edges returns empty dict."""
+    edges = context._ddp_repo.get_influence_edges(_PLAN_ID_1)
+    assert edges == {}, "Influence edges should be empty"

src/cleveragents/application/services/decision_service.py

@@ -907,54 +907,69 @@ 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*.
+     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.
+         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,
-            )
+         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)
 
-    def get_influence_edges(self, plan_id: str) -> dict[str, list[str]]:
-        """Build the influence DAG adjacency list for a plan.
+             # Persist to database if available
+             if self._persisted and self.unit_of_work is not None:
+                 with self.unit_of_work.transaction() as ctx:
+                     ctx.decisions.record_dependency(upstream_id, decision_id)
 
-        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()``.
+             self._logger.debug(
+                 "decision.dependency_recorded",
+                 source=upstream_id,
+                 target=decision_id,
+             )
 
-        Args:
-            plan_id: ULID of the plan.
+     def get_influence_edges(self, plan_id: str) -> dict[str, list[str]]:
+         """Build the influence DAG adjacency list for a 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
+         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()``.
+
+         In persisted mode, retrieves edges from the database. In in-memory
+         mode, uses the in-memory ``_dependencies`` cache.
+
+         Args:
+             plan_id: ULID of the plan.
+
+         Returns:
+             Adjacency list for the plan's influence DAG.
+         """
+         # In persisted mode, fetch from database
+         if self._persisted and self.unit_of_work is not None:
+             with self.unit_of_work.transaction() as ctx:
+                 return ctx.decisions.get_influence_edges(plan_id)
+
+         # In-memory mode: use the in-memory cache
+         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(

src/cleveragents/infrastructure/database/repositories.py

@@ -5556,35 +5556,140 @@ class DecisionRepository(DecisionRepositoryProtocol):
                 f"Failed to get max sequence number for plan {plan_id}: {exc}",
             ) from exc
 
-    # --- COUNT -------------------------------------------------------------
+     # --- COUNT -------------------------------------------------------------
 
-    @database_retry
-    def count(self, plan_id: str) -> int:
-        """Return the number of decisions for *plan_id*.
+     @database_retry
+     def count(self, plan_id: str) -> int:
+         """Return the number of decisions for *plan_id*.
 
-        Uses ``COUNT(*)`` for efficiency instead of loading all rows.
+         Uses ``COUNT(*)`` for efficiency instead of loading all rows.
 
-        Args:
-            plan_id: ULID of the plan.
+         Args:
+             plan_id: ULID of the plan.
 
-        Returns:
-            The decision count (``0`` when no decisions exist).
+         Returns:
+             The decision count (``0`` when no decisions exist).
 
-        Raises:
-            DatabaseError: On transient or unexpected DB errors.
-        """
-        session = self._session()
-        try:
-            result: int = (
-                session.query(sa_func.count(DecisionModel.decision_id))
-                .filter(DecisionModel.plan_id == plan_id)
-                .scalar()
-            ) or 0
-            return result
-        except (OperationalError, SQLAlchemyDatabaseError) as exc:
-            raise DatabaseError(
-                f"Failed to count decisions for plan {plan_id}: {exc}",
-            ) from exc
+         Raises:
+             DatabaseError: On transient or unexpected DB errors.
+         """
+         session = self._session()
+         try:
+             result: int = (
+                 session.query(sa_func.count(DecisionModel.decision_id))
+                 .filter(DecisionModel.plan_id == plan_id)
+                 .scalar()
+             ) or 0
+             return result
+         except (OperationalError, SQLAlchemyDatabaseError) as exc:
+             raise DatabaseError(
+                 f"Failed to count decisions for plan {plan_id}: {exc}",
+             ) from exc
+
+     # --- DEPENDENCIES ------------------------------------------------------
+
+     @database_retry
+     def record_dependency(
+         self,
+         source_decision_id: str,
+         target_decision_id: str,
+         relationship_type: str = "influences",
+     ) -> None:
+         """Record an influence edge from source to target decision.
+
+         Persists a decision dependency edge to the ``decision_dependencies``
+         table. This records that the source decision influenced or
+         constrained the target decision.
+
+         Args:
+             source_decision_id: ULID of the upstream (influencing) decision.
+             target_decision_id: ULID of the downstream (influenced) decision.
+             relationship_type: Type of relationship (default: "influences").
+
+         Raises:
+             DatabaseError: On constraint violation or transient DB errors.
+         """
+         from cleveragents.infrastructure.database.models import (
+             DecisionDependencyModel,
+         )
+
+         session = self._session()
+         try:
+             now_iso = datetime.now(UTC).isoformat()
+             dep_model = DecisionDependencyModel(
+                 source_decision_id=source_decision_id,
+                 target_decision_id=target_decision_id,
+                 relationship_type=relationship_type,
+                 created_at=now_iso,
+             )
+             session.add(dep_model)
+             session.flush()
+         except IntegrityError as exc:
+             session.rollback()
+             raise DatabaseError(
+                 f"Failed to record dependency {source_decision_id} -> "
+                 f"{target_decision_id}: {exc}",
+             ) from exc
+         except (OperationalError, SQLAlchemyDatabaseError) as exc:
+             session.rollback()
+             raise DatabaseError(
+                 f"Failed to record dependency {source_decision_id} -> "
+                 f"{target_decision_id}: {exc}",
+             ) from exc
+
+     @database_retry
+     def get_influence_edges(self, plan_id: str) -> dict[str, list[str]]:
+         """Retrieve 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*.
+
+         Args:
+             plan_id: ULID of the plan.
+
+         Returns:
+             Adjacency list mapping source → [targets] for the plan.
+
+         Raises:
+             DatabaseError: On transient or unexpected DB errors.
+         """
+         from cleveragents.infrastructure.database.models import (
+             DecisionDependencyModel,
+             DecisionModel,
+         )
+
+         session = self._session()
+         try:
+             # Get all decision IDs for the plan
+             plan_decision_ids = (
+                 session.query(DecisionModel.decision_id)
+                 .filter(DecisionModel.plan_id == plan_id)
+                 .all()
+             )
+             plan_ids_set = {row[0] for row in plan_decision_ids}
+
+             # Get all dependencies where both source and target are in the plan
+             deps = (
+                 session.query(
+                     DecisionDependencyModel.source_decision_id,
+                     DecisionDependencyModel.target_decision_id,
+                 )
+                 .filter(
+                     DecisionDependencyModel.source_decision_id.in_(plan_ids_set),
+                     DecisionDependencyModel.target_decision_id.in_(plan_ids_set),
+                 )
+                 .all()
+             )
+
+             # Build adjacency list
+             edges: dict[str, list[str]] = {}
+             for source, target in deps:
+                 edges.setdefault(source, []).append(target)
+             return edges
+         except (OperationalError, SQLAlchemyDatabaseError) as exc:
+             raise DatabaseError(
+                 f"Failed to get influence edges for plan {plan_id}: {exc}",
+             ) from exc
 
 
 # ---------------------------------------------------------------------------