diff --git a/CHANGELOG.md b/CHANGELOG.md index 71833a437..51f786bd9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1122,6 +1122,8 @@ Documentation Report (Cycle N)` issues every 10 cycles (~3.3 hours). The manager behavior for empty validation summaries and no-attachment runs. ### Fixed +- **Decision dependency persistence** (#7926): `DecisionService._record_dependencies` now persists all dependency edges in a single atomic ``UnitOfWork`` transaction via ``DecisionRepository.record_dependency()``. ``get_influence_edges()`` reads from the database in persisted mode and merges with in-memory edges for completeness after restarts, so the influence DAG survives service restarts and ``CorrectionService._compute_affected_subtree()`` can traverse persisted edges. Added ``DecisionRepositoryProtocol.record_dependency`` and ``get_influence_edges`` declarations. Duplicate edge inserts are silently ignored. + - **Plan Concurrency Race Condition** (#7989): Fixed critical race condition in `execute_plan()` and `apply_plan()` where concurrent CLI/worker sessions could simultaneously modify the same plan, diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index ff96cca40..bc4e48150 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -27,6 +27,7 @@ Below are some of the specific details of various contributions. +* HAL 9000 has implemented persistent decision dependency edge tracking, including `DecisionRepository.record_dependency()`/`.get_influence_edges()` methods and updated `DecisionService` to persist influence DAG edges to the database (closes #7926). * Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner. * Jeffrey Phillips Freeman has contributed the invariant merge precedence fix (#9126): restored the missing ACTION scope in ``merge_invariants()`` and ``InvariantSet.merge()``, corrected all module docstrings from ``plan > project > global`` to the spec-compliant ``plan > action > project > global``, and added comprehensive BDD test coverage for four-tier merge precedence. diff --git a/features/decision_dependency_persistence.feature b/features/decision_dependency_persistence.feature new file mode 100644 index 000000000..5cf0e30e6 --- /dev/null +++ b/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 diff --git a/features/decision_phase_gating.feature b/features/decision_phase_gating.feature index 46daaa283..06c90a72d 100644 --- a/features/decision_phase_gating.feature +++ b/features/decision_phase_gating.feature @@ -113,8 +113,8 @@ Feature: Decision type phase-gating at recording time When I try to record a "strategy_choice" decision without explicit phase Then a phase violation error should be raised - Scenario: Phase-gating skipped when plan not found in database - Given a phase-gated decision service with an empty database + Scenario: Phase-gating skipped for ungated persisted phase + Given a phase-gated decision service with a persisted apply plan When I record a "tool_invocation" decision without explicit phase Then the phase-gated decision should be recorded successfully diff --git a/features/steps/decision_dependency_basic_steps.py b/features/steps/decision_dependency_basic_steps.py new file mode 100644 index 000000000..4740c26cf --- /dev/null +++ b/features/steps/decision_dependency_basic_steps.py @@ -0,0 +1,350 @@ +"""Basic steps for decision dependency persistence feature.""" + +from __future__ import annotations + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.core.exceptions import DatabaseError +from cleveragents.infrastructure.database.models import DecisionDependencyModel + +from features.steps.helpers.decision_dependency_helpers import ( + PLAN_ID_1, + PLAN_ID_2, + create_child, + create_prerequisite_action, + create_prerequisite_plan, + create_root, + get_edges, + record_dependency, + setup_db, +) + + +@given("a fresh database for decision dependencies") +def step_fresh_db(context: Context) -> None: + """Set up a fresh 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) + + +@given("a persisted root decision for dependencies") +def step_persisted_root_dep(context: Context) -> None: + """Create and persist a root decision.""" + root = create_root(context) + context._ddp_root_id = root.decision_id + context._ddp_root_decision = root + + +@given("a persisted child decision for dependencies") +def step_persisted_child_dep(context: Context) -> None: + """Create and persist a child decision.""" + root_id = getattr(context, "_ddp_root_id", None) + if root_id is None: + root_id = context._ddp_root_ids[0] + child = create_child(context, root_id) + context._ddp_child_id = child.decision_id + context._ddp_child_decision = child + + +@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 index in range(count): + sibling = create_child( + context, + context._ddp_root_id, + sequence_number=index + 1, + ) + context._ddp_sibling_ids.append(sibling.decision_id) + + +@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 index in range(count): + root = create_root(context, index=index) + context._ddp_root_ids.append(root.decision_id) + context._ddp_root_id = context._ddp_root_ids[0] + + +@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.""" + root = create_root(context, PLAN_ID_1) + child = create_child(context, root.decision_id, PLAN_ID_1) + record_dependency(context, root.decision_id, child.decision_id) + context._ddp_plan1_root_id = root.decision_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.""" + root = create_root(context, PLAN_ID_2) + child = create_child(context, root.decision_id, PLAN_ID_2) + record_dependency(context, root.decision_id, child.decision_id) + context._ddp_plan2_root_id = root.decision_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 = create_root(context, PLAN_ID_1) + d2 = create_root(context, PLAN_ID_2) + context._ddp_cross_plan_id1 = d1.decision_id + context._ddp_cross_plan_id2 = d2.decision_id + + +@given("a recorded dependency from root to child") +def step_given_recorded_dependency_root_child(context: Context) -> None: + """Record a dependency from root to child during scenario setup.""" + record_dependency(context, context._ddp_root_id, context._ddp_child_id) + + +@given("recorded dependencies from root to both siblings") +def step_given_recorded_dependencies_root_siblings(context: Context) -> None: + """Record dependencies from root to all siblings during setup.""" + for sibling_id in context._ddp_sibling_ids: + record_dependency(context, context._ddp_root_id, sibling_id) + + +@given("recorded dependencies from both roots to the child") +def step_given_recorded_dependencies_both_roots(context: Context) -> None: + """Record dependencies from each root to the child during setup.""" + for root_id in context._ddp_root_ids: + record_dependency(context, root_id, context._ddp_child_id) + + +@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.""" + record_dependency(context, context._ddp_root_id, context._ddp_child_id) + + +@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: + record_dependency(context, context._ddp_root_id, sibling_id) + + +@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.""" + record_dependency( + context, + context._ddp_root_id, + context._ddp_child_id, + relationship_type=rel_type, + ) + 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: + record_dependency(context, context._ddp_root_id, context._ddp_root_id) + context._ddp_self_loop_error = None + except DatabaseError as exc: + context._ddp_self_loop_error = exc + + +@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: + record_dependency(context, context._ddp_root_id, context._ddp_child_id) + context._ddp_duplicate_error = None + except DatabaseError as exc: + context._ddp_duplicate_error = exc + + +@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: + record_dependency( + context, + context._ddp_cross_plan_id1, + context._ddp_cross_plan_id2, + ) + context._ddp_cross_plan_error = None + except DatabaseError as exc: + context._ddp_cross_plan_error = exc + + +@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 = get_edges(context, 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 = get_edges(context, 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 = get_edges(context, PLAN_ID_2) + + +@then("the dependency should be persisted to the database") +def step_assert_dependency_persisted(context: Context) -> None: + """Assert that the dependency was persisted.""" + edges = get_edges(context) + target_id = getattr(context, "_ddp_new_decision_id", None) + if target_id is None: + target_id = context._ddp_child_id + assert context._ddp_root_id in edges, "Root should be in edges" + assert target_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.""" + _assert_dependency_relationship_type(context, rel_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.""" + _assert_dependency_relationship_type(context, rel_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 = get_edges(context) + if hasattr(context, "_ddp_new_decision_id"): + 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" + ) + return + + 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 contain {count:d} sources") +def step_assert_influence_edges_sources_count(context: Context, count: int) -> None: + """Assert the number of sources in influence edges.""" + step_assert_influence_edges_source_count(context, count) + + +@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.""" + 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 = get_edges(context) + 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("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" + + +def _assert_dependency_relationship_type(context: Context, rel_type: str) -> None: + 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}" + ) diff --git a/features/steps/decision_dependency_integration_steps.py b/features/steps/decision_dependency_integration_steps.py new file mode 100644 index 000000000..3704287ca --- /dev/null +++ b/features/steps/decision_dependency_integration_steps.py @@ -0,0 +1,184 @@ +"""DecisionService and correction workflow steps for dependency persistence.""" + +from __future__ import annotations + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.decision_service import DecisionService +from cleveragents.domain.models.core.decision import DecisionType + +from features.steps.helpers.decision_dependency_helpers import ( + PLAN_ID_1, + affected_subtree, + create_child, + create_three_level_tree, + get_edges, + record_dependency, + record_three_level_edges, +) + + +@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.""" + create_three_level_tree(context) + + +@given("recorded dependencies forming a DAG") +def step_recorded_dag_dep(context: Context) -> None: + """Record dependencies forming a DAG.""" + record_three_level_edges(context) + + +@given("a persisted decision tree with dependencies") +def step_persisted_decision_tree_with_dependencies(context: Context) -> None: + """Create a persisted decision tree and matching dependency edges.""" + create_three_level_tree(context) + record_three_level_edges(context) + context._ddp_root_id = context._ddp_tree_root_id + context._ddp_child_id = context._ddp_tree_child_id + + +@given("a persisted child decision depending on the root") +def step_persisted_child_decision_depending_on_root(context: Context) -> None: + """Create a child decision and persist a root -> child dependency.""" + child = create_child(context, context._ddp_root_id) + context._ddp_child_id = child.decision_id + context._ddp_child_decision = child + record_dependency(context, context._ddp_root_id, context._ddp_child_id) + + +@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.""" + 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: + raise AssertionError(f"Unsupported dependency list literal: {deps}") + + 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 + + +@when("I compute affected subtree using influence edges") +def step_compute_affected_subtree(context: Context) -> None: + """Compute all downstream decisions from the persisted influence DAG.""" + context._ddp_influence_edges = get_edges(context) + context._ddp_affected_subtree = affected_subtree( + context._ddp_tree_root_id, + context._ddp_influence_edges, + ) + + +@when("I record a correction decision with the same dependencies") +def step_record_correction_decision_same_dependencies(context: Context) -> None: + """Record a correction decision that depends on the same root.""" + service = DecisionService(unit_of_work=context._ddp_uow) + correction = service.record_decision( + plan_id=PLAN_ID_1, + decision_type=DecisionType.USER_INTERVENTION, + question="How should the previous decision be corrected?", + chosen_option="Apply corrected approach", + parent_decision_id=context._ddp_child_id, + dependency_decision_ids=[context._ddp_root_id], + ) + context._ddp_correction_decision_id = correction.decision_id + + +@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("no dependencies should be created") +def step_assert_no_dependencies_created(context: Context) -> None: + """Assert that no dependencies were created.""" + edges = get_edges(context) + 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 = get_edges(context) + 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 = get_edges(context) + 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 = get_edges(context) + assert edges == {}, "Influence edges should be empty" + + +@then("the affected subtree should include all transitive dependents") +def step_assert_affected_subtree_transitive(context: Context) -> None: + """Assert transitive dependents are included in the affected subtree.""" + expected = { + context._ddp_tree_child_id, + context._ddp_tree_grandchild_id, + } + assert expected <= context._ddp_affected_subtree, ( + f"Expected {expected}, got {context._ddp_affected_subtree}" + ) + + +@then("the correction decision should have the same influence edges") +def step_assert_correction_decision_has_same_edges(context: Context) -> None: + """Assert the correction decision is linked to the same upstream root.""" + edges = get_edges(context) + assert context._ddp_root_id in edges, "Root should be in influence edges" + assert context._ddp_child_id in edges[context._ddp_root_id], ( + "Original child dependency should remain" + ) + assert context._ddp_correction_decision_id in edges[context._ddp_root_id], ( + "Correction decision should depend on the same root" + ) diff --git a/features/steps/decision_phase_gating_steps.py b/features/steps/decision_phase_gating_steps.py index 221c8dd5a..df131504a 100644 --- a/features/steps/decision_phase_gating_steps.py +++ b/features/steps/decision_phase_gating_steps.py @@ -23,6 +23,7 @@ from cleveragents.application.services.phase_gating import ( validate_phase_gating, ) from cleveragents.core.exceptions import DecisionPhaseViolationError, ValidationError +from cleveragents.domain.models.core.action import Action from cleveragents.domain.models.core.decision import ( EXECUTE_TYPES, STRATEGIZE_TYPES, @@ -57,6 +58,7 @@ def step_phase_gated_service(context: Context) -> None: context.pg_error = None context._pg_plan_id = str(ULID()) context._pg_seq = 0 + context._pg_root_decision_id = None # --------------------------------------------------------------------------- @@ -76,22 +78,10 @@ def step_persisted_execute_plan(context: Context) -> None: _setup_persisted_plan(context, PlanPhase.EXECUTE) -@given("a phase-gated decision service with an empty database") -def step_persisted_empty_db(context: Context) -> None: - """Create a DecisionService with a wired UoW but no plans in the DB.""" - fd, db_path = tempfile.mkstemp(suffix=".db", prefix="pg_empty_") - os.close(fd) - context._pg_db_path = db_path - uow = UnitOfWork(f"sqlite:///{db_path}") - uow.init_database() - - context._pg_plan_id = str(ULID()) - context.pg_service = DecisionService(unit_of_work=uow) - context.pg_result = None - context.pg_error = None - context._pg_seq = 0 - - _register_db_cleanup(context, db_path, uow) +@given("a phase-gated decision service with a persisted apply plan") +def step_persisted_apply_plan(context: Context) -> None: + """Create a DecisionService backed by a real database with an ungated plan.""" + _setup_persisted_plan(context, PlanPhase.APPLY) def _setup_persisted_plan(context: Context, phase: PlanPhase) -> None: @@ -104,7 +94,9 @@ def _setup_persisted_plan(context: Context, phase: PlanPhase) -> None: plan_id = str(ULID()) context._pg_plan_id = plan_id + context._pg_root_decision_id = None + action = _phase_test_action() plan = Plan( identity=PlanIdentity(plan_id=plan_id), namespaced_name=NamespacedName(namespace="local", name="pg-test"), @@ -114,6 +106,7 @@ def _setup_persisted_plan(context: Context, phase: PlanPhase) -> None: processing_state=ProcessingState.PROCESSING, ) with uow.transaction() as txn: + txn.actions.create(action) txn.lifecycle_plans.create(plan) context.pg_service = DecisionService(unit_of_work=uow) @@ -138,6 +131,40 @@ def _register_db_cleanup(context: Context, db_path: str, uow: UnitOfWork) -> Non context._cleanup_handlers.append(_cleanup_db) +def _phase_test_action() -> Action: + """Return the minimal action row required by persisted phase-gating plans.""" + return Action( + namespaced_name=NamespacedName(namespace="local", name="pg-test-action"), + description="Phase gating test action", + definition_of_done="Phase gating test done", + strategy_actor="test/strategy", + execution_actor="test/execution", + created_by="phase-gating-test", + ) + + +def _ensure_persisted_parent_decision( + context: Context, + plan_id: str, + decision_type: DecisionType, +) -> str | None: + """Create a persisted root decision before child decisions when FKs apply.""" + if decision_type == DecisionType.PROMPT_DEFINITION: + return None + if not getattr(context.pg_service, "_persisted", False): + return str(ULID()) + if getattr(context, "_pg_root_decision_id", None) is None: + root = context.pg_service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Phase-gated root question", + chosen_option="Phase-gated root choice", + plan_phase=PlanPhase.STRATEGIZE, + ) + context._pg_root_decision_id = root.decision_id + return context._pg_root_decision_id + + @given("a phase-gated decision service without persistence") def step_no_persistence_service(context: Context) -> None: context.pg_service = DecisionService() @@ -167,11 +194,8 @@ def step_record_with_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) - # Determine parent_decision_id: prompt_definition must be root - parent_id = None dt = DecisionType(dtype) - if dt != DecisionType.PROMPT_DEFINITION: - parent_id = str(ULID()) + parent_id = _ensure_persisted_parent_decision(context, plan_id, dt) d = svc.record_decision( plan_id=plan_id, @@ -190,7 +214,11 @@ def step_try_record_with_phase(context: Context, dtype: str, phase: str) -> None svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) - parent_id = str(ULID()) + parent_id = _ensure_persisted_parent_decision( + context, + plan_id, + DecisionType(dtype), + ) try: svc.record_decision( @@ -211,10 +239,8 @@ def step_record_without_phase(context: Context, dtype: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) - parent_id = str(ULID()) dt = DecisionType(dtype) - if dt == DecisionType.PROMPT_DEFINITION: - parent_id = None + parent_id = _ensure_persisted_parent_decision(context, plan_id, dt) try: d = svc.record_decision( @@ -253,10 +279,8 @@ def step_record_with_string_phase(context: Context, dtype: str, phase: str) -> N svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) - parent_id = None dt = DecisionType(dtype) - if dt != DecisionType.PROMPT_DEFINITION: - parent_id = str(ULID()) + parent_id = _ensure_persisted_parent_decision(context, plan_id, dt) d = svc.record_decision( plan_id=plan_id, @@ -275,10 +299,8 @@ def step_record_with_enum_phase(context: Context, dtype: str, phase: str) -> Non svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) - parent_id = None dt = DecisionType(dtype) - if dt != DecisionType.PROMPT_DEFINITION: - parent_id = str(ULID()) + parent_id = _ensure_persisted_parent_decision(context, plan_id, dt) d = svc.record_decision( plan_id=plan_id, @@ -367,7 +389,11 @@ def step_phase_map_execute(context: Context, phase: str) -> None: def step_try_record_invalid_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) - parent_id = str(ULID()) + parent_id = _ensure_persisted_parent_decision( + context, + plan_id, + DecisionType(dtype), + ) try: svc.record_decision( diff --git a/features/steps/decision_recording_steps.py b/features/steps/decision_recording_steps.py index 74f476145..c58045e3f 100644 --- a/features/steps/decision_recording_steps.py +++ b/features/steps/decision_recording_steps.py @@ -24,12 +24,20 @@ from cleveragents.application.services.decision_service import ( SnapshotStore, ) from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.action import Action from cleveragents.domain.models.core.decision import ( ArtifactRef, ContextSnapshot, Decision, DecisionType, ) +from cleveragents.domain.models.core.plan import ( + NamespacedName, + Plan, + PlanIdentity, + PlanPhase, + ProcessingState, +) def _resolve_plan_id(context: Context, symbolic_id: str) -> str: @@ -43,9 +51,45 @@ def _resolve_plan_id(context: Context, symbolic_id: str) -> str: if symbolic_id not in registry: registry[symbolic_id] = str(ULID()) context._plan_id_registry = registry + _ensure_persisted_plan(context, registry[symbolic_id]) return registry[symbolic_id] +def _dsvc_test_action() -> Action: + """Return the minimal action row required by persisted decision tests.""" + return Action( + namespaced_name=NamespacedName(namespace="local", name="dsvc-test-action"), + description="Decision service persistence test action", + definition_of_done="Decision service persistence test done", + strategy_actor="test/strategy", + execution_actor="test/execution", + created_by="decision-recording-test", + ) + + +def _ensure_persisted_plan(context: Context, plan_id: str) -> None: + """Seed FK prerequisites for persisted decision-service scenarios.""" + uow = getattr(context, "_dsvc_uow", None) + if uow is None: + return + + with uow.transaction() as txn: + action_name = "local/dsvc-test-action" + if txn.actions.get_by_id(action_name) is None: + txn.actions.create(_dsvc_test_action()) + if txn.lifecycle_plans.get(plan_id) is None: + txn.lifecycle_plans.create( + Plan( + identity=PlanIdentity(plan_id=plan_id), + namespaced_name=NamespacedName(namespace="local", name="dsvc-test"), + description="Decision service persistence test plan", + action_name=action_name, + phase=PlanPhase.APPLY, + processing_state=ProcessingState.PROCESSING, + ) + ) + + # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @@ -67,6 +111,7 @@ def step_decision_service(context: Context) -> None: context.delete_result = None context.explicit_snapshot = None context._plan_id_registry = {} + context._dsvc_uow = None # --------------------------------------------------------------------------- @@ -1042,6 +1087,7 @@ def step_persisted_decision_service(context: Context) -> None: context._dsvc_db_path = db_path uow = UnitOfWork(f"sqlite:///{db_path}") uow.init_database() + context._dsvc_uow = uow context.decision_service = DecisionService(unit_of_work=uow) context.recorded_decisions = [] @@ -1081,6 +1127,7 @@ def step_recreate_service_same_db(context: Context) -> None: db_path = context._dsvc_db_path uow = UnitOfWork(f"sqlite:///{db_path}") uow.init_database() + context._dsvc_uow = uow context.decision_service = DecisionService(unit_of_work=uow) # Preserve _plan_id_registry so symbolic IDs ("P1") still resolve diff --git a/features/steps/helpers/__init__.py b/features/steps/helpers/__init__.py new file mode 100644 index 000000000..103aa2187 --- /dev/null +++ b/features/steps/helpers/__init__.py @@ -0,0 +1 @@ +"""Shared Behave step helpers.""" diff --git a/features/steps/helpers/decision_dependency_helpers.py b/features/steps/helpers/decision_dependency_helpers.py new file mode 100644 index 000000000..f5ec2336d --- /dev/null +++ b/features/steps/helpers/decision_dependency_helpers.py @@ -0,0 +1,224 @@ +"""Shared helpers for decision dependency persistence Behave steps.""" + +from __future__ import annotations + +import tempfile +from collections import deque +from datetime import UTC, datetime +from pathlib import Path + +from behave.runner import Context +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +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 + +PLAN_ID_1 = "01HV000000000000000000DP01" +PLAN_ID_2 = "01HV000000000000000000DP02" + + +def setup_db(context: Context) -> None: + """Create a per-scenario SQLite database and attach repositories.""" + tmp_dir = Path(tempfile.mkdtemp(prefix="ddp-", dir="/tmp")) + db_path = tmp_dir / "decision-dependencies.sqlite" + engine = create_engine(f"sqlite:///{db_path}", echo=False) + Base.metadata.create_all(engine) + session_factory = sessionmaker( + bind=engine, + expire_on_commit=False, + autoflush=False, + autocommit=False, + ) + session = session_factory() + + context._ddp_tmp_dir = tmp_dir + context._ddp_db_url = f"sqlite:///{db_path}" + context._ddp_engine = engine + context._ddp_session = session + context._ddp_uow_session_factory = session_factory + 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( + context._ddp_db_url, + require_confirmation=False, + ) + context._ddp_uow._engine = engine + context._ddp_uow._session_factory = session_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() + + +def persist_decision(context: Context, decision: Decision) -> Decision: + """Persist one decision and commit it.""" + context._ddp_repo.create(decision) + context._ddp_session.commit() + return decision + + +def create_root(context: Context, plan_id: str = PLAN_ID_1, index: int = 0) -> Decision: + """Create a persisted root decision.""" + return persist_decision( + context, + make_decision(plan_id=plan_id, sequence_number=index), + ) + + +def create_child( + context: Context, + root_id: str, + plan_id: str = PLAN_ID_1, + sequence_number: int = 1, +) -> Decision: + """Create a persisted child decision.""" + return persist_decision( + context, + make_decision( + plan_id=plan_id, + parent_decision_id=root_id, + sequence_number=sequence_number, + decision_type=DecisionType.STRATEGY_CHOICE, + ), + ) + + +def record_dependency( + context: Context, + source_id: str, + target_id: str, + relationship_type: str = "influences", +) -> None: + """Persist one dependency edge and commit it.""" + context._ddp_repo.record_dependency( + source_id, + target_id, + relationship_type=relationship_type, + ) + context._ddp_session.commit() + + +def get_edges(context: Context, plan_id: str = PLAN_ID_1) -> dict[str, list[str]]: + """Read influence edges after expiring the assertion session cache.""" + context._ddp_session.expire_all() + return context._ddp_repo.get_influence_edges(plan_id) + + +def create_three_level_tree(context: Context) -> None: + """Create a root -> child -> grandchild tree.""" + root = create_root(context) + child = create_child(context, root.decision_id, sequence_number=1) + grandchild = persist_decision( + context, + make_decision( + plan_id=PLAN_ID_1, + parent_decision_id=child.decision_id, + sequence_number=2, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + ), + ) + context._ddp_tree_root_id = root.decision_id + context._ddp_tree_child_id = child.decision_id + context._ddp_tree_grandchild_id = grandchild.decision_id + + +def record_three_level_edges(context: Context) -> None: + """Record root -> child and child -> grandchild dependencies.""" + record_dependency( + context, + context._ddp_tree_root_id, + context._ddp_tree_child_id, + ) + record_dependency( + context, + context._ddp_tree_child_id, + context._ddp_tree_grandchild_id, + ) + + +def affected_subtree(start_id: str, edges: dict[str, list[str]]) -> set[str]: + """Return all transitive dependents reachable from start_id.""" + affected: set[str] = set() + queue: deque[str] = deque(edges.get(start_id, [])) + while queue: + decision_id = queue.popleft() + if decision_id in affected: + continue + affected.add(decision_id) + queue.extend(edges.get(decision_id, [])) + return affected diff --git a/pyproject.toml b/pyproject.toml index dbeafd732..72d14970b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -156,6 +156,7 @@ strictDictionaryInference = true strictSetInference = true reportMissingImports = true reportMissingTypeStubs = false +reportMissingModuleSource = false reportMissingTypeArgument = true reportIncompatibleMethodOverride = true reportIncompatibleVariableOverride = true diff --git a/src/cleveragents/application/services/decision_service.py b/src/cleveragents/application/services/decision_service.py index 06a86212a..2055c5bd3 100644 --- a/src/cleveragents/application/services/decision_service.py +++ b/src/cleveragents/application/services/decision_service.py @@ -385,11 +385,19 @@ class DecisionService: correction_reason=correction_reason, ) - self._store_decision(decision) + dependency_ids = self._normalise_dependency_ids(dependency_decision_ids or []) - # Record influence edges in the dependency DAG - if dependency_decision_ids: - self._record_dependencies(decision.decision_id, dependency_decision_ids) + try: + if self._persisted and self.unit_of_work is not None: + self._store_decision_with_dependencies(decision, dependency_ids) + else: + self._store_decision(decision) + if dependency_ids: + self._record_dependencies(decision.decision_id, dependency_ids) + except Exception: + if self._plan_sequence.get(plan_id) == seq + 1: + self._plan_sequence[plan_id] = seq + raise self._logger.info( "decision.recorded", @@ -397,9 +405,7 @@ class DecisionService: plan_id=plan_id, decision_type=str(decision_type), sequence=seq, - dependency_count=len(dependency_decision_ids) - if dependency_decision_ids - else 0, + dependency_count=len(dependency_ids), ) if self.event_bus is not None: try: @@ -907,6 +913,56 @@ class DecisionService: # Store snapshot self.snapshots.store(decision_id, decision.context_snapshot) + def _store_decision_with_dependencies( + self, + decision: Decision, + dependency_ids: list[str], + ) -> None: + """Persist a decision and its dependency edges atomically.""" + decision_id = decision.decision_id + plan_id = decision.plan_id + + if decision_id in self._decisions: + raise DuplicateDecisionError(decision_id) + + if self.unit_of_work is None: # pragma: no cover - defensive guard + raise RuntimeError("UnitOfWork is required for persisted decisions") + + with self.unit_of_work.transaction() as ctx: + ctx.decisions.create(decision) + for upstream_id in dependency_ids: + ctx.decisions.record_dependency(upstream_id, decision_id) + + self._decisions[decision_id] = decision + self._plan_decisions.setdefault(plan_id, []).append(decision_id) + self.snapshots.store(decision_id, decision.context_snapshot) + self._cache_dependencies(decision_id, dependency_ids) + + @staticmethod + def _normalise_dependency_ids(upstream_ids: list[str]) -> list[str]: + """Return nonblank dependency IDs with duplicates removed in order.""" + dependency_ids: list[str] = [] + seen: set[str] = set() + for upstream_id in upstream_ids: + candidate = upstream_id.strip() + if not candidate or candidate in seen: + continue + dependency_ids.append(candidate) + seen.add(candidate) + return dependency_ids + + def _cache_dependencies(self, decision_id: str, upstream_ids: list[str]) -> None: + """Cache influence edges in memory after persistence succeeds.""" + for upstream_id in upstream_ids: + targets = self._dependencies.setdefault(upstream_id, []) + if decision_id not in targets: + targets.append(decision_id) + self._logger.debug( + "decision.dependency_recorded", + source=upstream_id, + target=decision_id, + ) + def _record_dependencies( self, decision_id: str, @@ -914,7 +970,7 @@ class DecisionService: ) -> None: """Record influence edges from upstream decisions to *decision_id*. - Each upstream decision is stored as a source → target edge in the + Each upstream decision is stored as a source \u2192 target edge in the in-memory ``_dependencies`` dict and, when a :class:`UnitOfWork` is wired, persisted to the ``decision_dependencies`` table. @@ -922,23 +978,21 @@ class DecisionService: 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, - ) + dependency_ids = self._normalise_dependency_ids(upstream_ids) + + # Persist to database when UnitOfWork is wired. + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + for uid in dependency_ids: + ctx.decisions.record_dependency(uid, decision_id) + + self._cache_dependencies(decision_id, dependency_ids) 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()``. + In persisted mode, merges in-memory edges with DB-persisted + edges for completeness after a service restart. Args: plan_id: ULID of the plan. @@ -954,6 +1008,24 @@ class DecisionService: plan_targets = [t for t in targets if t in plan_decision_ids] if plan_targets: edges[source] = plan_targets + + # Merge with DB-persisted edges when UnitOfWork is available. + if self._persisted and self.unit_of_work is not None: + try: + with self.unit_of_work.transaction() as ctx: + db_edges = ctx.decisions.get_influence_edges(plan_id) + for source, targets in db_edges.items(): + existing = set(edges.get(source, [])) + for t in targets: + if t not in existing: + edges.setdefault(source, []).append(t) + existing.add(t) + except Exception: # pragma: no cover + self._logger.exception( + "influence_edges_db_query_failed", + plan_id=plan_id, + ) + return edges @staticmethod diff --git a/src/cleveragents/domain/repositories/decision_repository.py b/src/cleveragents/domain/repositories/decision_repository.py index 25fca280a..9d1750e0c 100644 --- a/src/cleveragents/domain/repositories/decision_repository.py +++ b/src/cleveragents/domain/repositories/decision_repository.py @@ -92,6 +92,38 @@ class DecisionRepositoryProtocol(Protocol): """ ... + 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. Duplicate edges (same source and target) are silently ignored. + + 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"). + """ + return None + + 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 to targets for the plan. + """ + return {} + def delete(self, decision_id: str) -> bool: """Delete a decision by its ULID. diff --git a/src/cleveragents/infrastructure/database/repositories.py b/src/cleveragents/infrastructure/database/repositories.py index f00f3c1cb..76eb5422c 100644 --- a/src/cleveragents/infrastructure/database/repositories.py +++ b/src/cleveragents/infrastructure/database/repositories.py @@ -111,6 +111,7 @@ from cleveragents.infrastructure.database.models import ( ContextModel, CorrectionAttemptModel, DebugAttemptModel, + DecisionDependencyModel, DecisionModel, LifecycleActionModel, LifecyclePlanModel, @@ -5746,6 +5747,114 @@ class DecisionRepository(DecisionRepositoryProtocol): 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. + """ + if source_decision_id == target_decision_id: + raise DatabaseError("Decision dependency self-loops are not allowed") + 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() + existing = ( + session.query(DecisionDependencyModel) + .filter_by( + source_decision_id=source_decision_id, + target_decision_id=target_decision_id, + ) + .first() + ) + if existing is not None: + return + raise DatabaseError( + f"Failed to record dependency {source_decision_id} -> " + f"{target_decision_id}: {exc}", + ) from exc + except (OperationalError, SQLAlchemyDatabaseError) as exc: # pragma: no cover + 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. + """ + 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: # pragma: no cover + raise DatabaseError( + f"Failed to get influence edges for plan {plan_id}: {exc}", + ) from exc + # --------------------------------------------------------------------------- # Checkpoint Repository (Stage M6 - checkpointing and rollback) diff --git a/src/cleveragents/infrastructure/database/unit_of_work.py b/src/cleveragents/infrastructure/database/unit_of_work.py index 0d4d63598..a4b19bbb8 100644 --- a/src/cleveragents/infrastructure/database/unit_of_work.py +++ b/src/cleveragents/infrastructure/database/unit_of_work.py @@ -9,7 +9,7 @@ from collections.abc import Callable, Generator from contextlib import contextmanager from typing import TYPE_CHECKING, Any -from sqlalchemy import create_engine +from sqlalchemy import create_engine, event from sqlalchemy.orm import Session, sessionmaker from cleveragents.infrastructure.database.engine_cache import ( @@ -34,6 +34,18 @@ if TYPE_CHECKING: from sqlalchemy.engine import Engine +def _enable_sqlite_foreign_keys(engine: Engine) -> None: + """Enable SQLite foreign-key enforcement for every DBAPI connection.""" + + @event.listens_for(engine, "connect") + def _set_sqlite_pragma(dbapi_connection: Any, _connection_record: Any) -> None: + cursor = dbapi_connection.cursor() + try: + cursor.execute("PRAGMA foreign_keys=ON") + finally: + cursor.close() + + class UnitOfWork: """Unit of Work pattern for managing database transactions. @@ -125,9 +137,14 @@ class UnitOfWork: isolation_level="SERIALIZABLE", connect_args={"check_same_thread": False}, ) + _enable_sqlite_foreign_keys( + MEMORY_ENGINES[self.database_url], + ) # Always assign after the lock so cache-hits also # populate self._engine correctly. self._engine = MEMORY_ENGINES[self.database_url] + with self._engine.connect() as connection: + connection.exec_driver_sql("PRAGMA foreign_keys=ON") else: # File-based SQLite # Use SERIALIZABLE isolation for SQLite to ensure proper rollback @@ -138,6 +155,7 @@ class UnitOfWork: isolation_level="SERIALIZABLE", connect_args={"check_same_thread": False}, ) + _enable_sqlite_foreign_keys(self._engine) else: # PostgreSQL (or other non-SQLite): use connection pooling # suitable for multi-user server mode. diff --git a/tests/application/test_decision_service_dependencies.py b/tests/application/test_decision_service_dependencies.py new file mode 100644 index 000000000..2dabd920b --- /dev/null +++ b/tests/application/test_decision_service_dependencies.py @@ -0,0 +1,316 @@ +from __future__ import annotations + +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field + +import pytest + +from cleveragents.application.services.decision_service import ( + DecisionService, + DuplicateDecisionError, +) +from cleveragents.core.exceptions import DatabaseError +from cleveragents.domain.models.core.decision import Decision, DecisionType +from cleveragents.domain.models.core.plan import PlanPhase +from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES +from cleveragents.infrastructure.database.models import ( + LifecycleActionModel, + LifecyclePlanModel, +) +from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + + +@dataclass +class RecordingDecisionRepository: + created: list[Decision] = field(default_factory=list) + dependencies: list[tuple[str, str]] = field(default_factory=list) + influence_edges: dict[str, list[str]] = field(default_factory=dict) + + def create(self, decision: Decision) -> Decision: + self.created.append(decision) + return decision + + def get_max_sequence_number(self, plan_id: str) -> int | None: + return None + + def record_dependency( + self, + source_decision_id: str, + target_decision_id: str, + relationship_type: str = "influences", + ) -> None: + self.dependencies.append((source_decision_id, target_decision_id)) + + def get_influence_edges(self, plan_id: str) -> dict[str, list[str]]: + return self.influence_edges + + +@dataclass +class FakeUnitOfWorkContext: + decisions: RecordingDecisionRepository + + +@dataclass +class FakeUnitOfWork: + decisions: RecordingDecisionRepository + + @contextmanager + def transaction(self) -> Iterator[FakeUnitOfWorkContext]: + yield FakeUnitOfWorkContext(self.decisions) + + +def _seed_plan(uow: UnitOfWork, plan_id: str) -> None: + now = "2026-01-01T00:00:00+00:00" + action_name = "test/decision-dependencies" + with uow.engine.begin() as connection: + connection.execute( + LifecycleActionModel.__table__.insert().values( + namespaced_name=action_name, + namespace="test", + name="decision-dependencies", + description="Test action", + definition_of_done="Done", + strategy_actor="strategy", + execution_actor="execution", + state="available", + tags_json="[]", + created_at=now, + updated_at=now, + ) + ) + connection.execute( + LifecyclePlanModel.__table__.insert().values( + plan_id=plan_id, + root_plan_id=plan_id, + action_name=action_name, + namespaced_name="test/decision-dependencies-plan", + namespace="test", + description="Test plan", + definition_of_done="Done", + phase="execute", + processing_state="processing", + tags_json="[]", + created_at=now, + updated_at=now, + ) + ) + + +def test_record_decision_persists_only_nonblank_dependency_edges() -> None: + repository = RecordingDecisionRepository() + service = DecisionService(unit_of_work=FakeUnitOfWork(repository)) # type: ignore[arg-type] + + decision = service.record_decision( + plan_id="01J00000000000000000000001", + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which implementation path?", + chosen_option="Persist dependency edges", + plan_phase=PlanPhase.EXECUTE, + dependency_decision_ids=[ + "01J00000000000000000000002", + "", + " ", + "01J00000000000000000000003", + ], + ) + + assert repository.created == [decision] + assert repository.dependencies == [ + ("01J00000000000000000000002", decision.decision_id), + ("01J00000000000000000000003", decision.decision_id), + ] + + +def test_record_decision_caches_dependencies_without_unit_of_work() -> None: + service = DecisionService() + plan_id = "01J00000000000000000000001" + + source = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which source decision?", + chosen_option="The source exists", + plan_phase=PlanPhase.EXECUTE, + ) + + dependent = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which dependent decision?", + chosen_option="The dependent uses the source", + plan_phase=PlanPhase.EXECUTE, + dependency_decision_ids=[source.decision_id], + ) + + assert service.get_influence_edges(plan_id) == { + source.decision_id: [dependent.decision_id], + } + + +def test_record_decision_restores_sequence_when_store_fails() -> None: + service = DecisionService() + plan_id = "01J00000000000000000000001" + + source = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which source decision?", + chosen_option="The source exists", + plan_phase=PlanPhase.EXECUTE, + ) + + def fail_record_dependencies( + _decision_id: str, + _upstream_ids: list[str], + ) -> None: + raise RuntimeError("dependency cache failed") + + service._record_dependencies = fail_record_dependencies # type: ignore[method-assign] + + with pytest.raises(RuntimeError, match="dependency cache failed"): + service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which dependent decision fails?", + chosen_option="The dependency write fails", + plan_phase=PlanPhase.EXECUTE, + dependency_decision_ids=[source.decision_id], + ) + + assert service.get_next_sequence(plan_id) == 1 + + +def test_record_dependencies_persists_when_unit_of_work_is_wired() -> None: + repository = RecordingDecisionRepository() + service = DecisionService(unit_of_work=FakeUnitOfWork(repository)) # type: ignore[arg-type] + + service._record_dependencies( + "01J00000000000000000000003", + [ + "01J00000000000000000000002", + "01J00000000000000000000002", + "", + ], + ) + + assert repository.dependencies == [ + ("01J00000000000000000000002", "01J00000000000000000000003"), + ] + + +def test_store_decision_with_dependencies_rejects_duplicate_cached_id() -> None: + repository = RecordingDecisionRepository() + service = DecisionService(unit_of_work=FakeUnitOfWork(repository)) # type: ignore[arg-type] + decision = service.record_decision( + plan_id="01J00000000000000000000001", + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which decision?", + chosen_option="Keep the first decision", + plan_phase=PlanPhase.EXECUTE, + ) + + with pytest.raises(DuplicateDecisionError): + service._store_decision_with_dependencies(decision, []) + + +def test_get_influence_edges_merges_persisted_edges_after_restart() -> None: + source_id = "01J00000000000000000000002" + target_id = "01J00000000000000000000003" + repository = RecordingDecisionRepository( + influence_edges={source_id: [target_id]}, + ) + service = DecisionService(unit_of_work=FakeUnitOfWork(repository)) # type: ignore[arg-type] + plan_id = "01J00000000000000000000001" + + source = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which source?", + chosen_option="Source", + plan_phase=PlanPhase.EXECUTE, + ) + target = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which target?", + chosen_option="Target", + plan_phase=PlanPhase.EXECUTE, + ) + + repository.influence_edges = {source.decision_id: [target.decision_id]} + + assert service.get_influence_edges(plan_id) == { + source.decision_id: [target.decision_id], + } + + +def test_memory_sqlite_unit_of_work_enables_foreign_keys() -> None: + MEMORY_ENGINES.pop("sqlite:///:memory:", None) + uow = UnitOfWork("sqlite:///:memory:", require_confirmation=False) + + with uow.engine.connect() as connection: + foreign_keys_enabled = connection.exec_driver_sql( + "PRAGMA foreign_keys", + ).scalar() + + assert foreign_keys_enabled == 1 + + +def test_memory_sqlite_unit_of_work_cache_miss_registers_fk_listener() -> None: + MEMORY_ENGINES.pop("sqlite:///:memory:", None) + uow = UnitOfWork("sqlite:///:memory:", require_confirmation=False) + uow._database_initialized = True + + with uow.engine.connect() as connection: + foreign_keys_enabled = connection.exec_driver_sql( + "PRAGMA foreign_keys", + ).scalar() + + assert foreign_keys_enabled == 1 + + +def test_record_decision_rolls_back_invalid_persisted_dependency( + tmp_path, +) -> None: # type: ignore[no-untyped-def] + database_url = f"sqlite:///{tmp_path / 'decisions.db'}" + uow = UnitOfWork(database_url, require_confirmation=False) + service = DecisionService(unit_of_work=uow) + plan_id = "01J00000000000000000000001" + _seed_plan(uow, plan_id) + + source = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which source decision?", + chosen_option="The source exists", + plan_phase=PlanPhase.EXECUTE, + ) + + with pytest.raises(DatabaseError): + service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which dependent decision?", + chosen_option="This should not persist", + plan_phase=PlanPhase.EXECUTE, + dependency_decision_ids=["01J00000000000000000000099"], + ) + + assert service.list_decisions(plan_id) == [source] + assert service.get_influence_edges(plan_id) == {} + + dependent = service.record_decision( + plan_id=plan_id, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Which valid dependent decision?", + chosen_option="This should persist", + plan_phase=PlanPhase.EXECUTE, + dependency_decision_ids=[source.decision_id, source.decision_id], + ) + + assert dependent.sequence_number == 1 + assert service.get_influence_edges(plan_id) == { + source.decision_id: [dependent.decision_id], + } + assert service.list_decisions(plan_id) == [source, dependent] diff --git a/tests/infrastructure/test_decision_repository_dependencies.py b/tests/infrastructure/test_decision_repository_dependencies.py new file mode 100644 index 000000000..94df58c88 --- /dev/null +++ b/tests/infrastructure/test_decision_repository_dependencies.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import pytest +from sqlalchemy import Engine, create_engine, event +from sqlalchemy.orm import Session, sessionmaker + +from cleveragents.core.exceptions import DatabaseError +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + Decision, + DecisionType, +) +from cleveragents.domain.repositories.decision_repository import ( + DecisionRepositoryProtocol, +) +from cleveragents.infrastructure.database.models import ( + Base, + DecisionDependencyModel, + DecisionModel, + LifecycleActionModel, + LifecyclePlanModel, +) +from cleveragents.infrastructure.database.repositories import DecisionRepository + + +class ProtocolOnlyDecisionRepository(DecisionRepositoryProtocol): + pass + + +def _decision( + decision_id: str, + plan_id: str, + sequence_number: int, + decision_type: DecisionType = DecisionType.IMPLEMENTATION_CHOICE, +) -> Decision: + return Decision( + decision_id=decision_id, + plan_id=plan_id, + decision_type=decision_type, + sequence_number=sequence_number, + question=f"Question {sequence_number}?", + chosen_option=f"Option {sequence_number}", + context_snapshot=ContextSnapshot(hot_context_hash=f"hash-{sequence_number}"), + ) + + +def _repository() -> tuple[DecisionRepository, Session, Engine]: + engine = create_engine("sqlite:///:memory:", future=True) + + @event.listens_for(engine, "connect") + def _enable_foreign_keys(dbapi_connection, _connection_record) -> None: # type: ignore[no-untyped-def] + cursor = dbapi_connection.cursor() + try: + cursor.execute("PRAGMA foreign_keys=ON") + finally: + cursor.close() + + Base.metadata.create_all( + engine, + tables=[ + LifecycleActionModel.__table__, + LifecyclePlanModel.__table__, + DecisionModel.__table__, + DecisionDependencyModel.__table__, + ], + ) + session_factory = sessionmaker(bind=engine, expire_on_commit=False, class_=Session) + session = session_factory() + return DecisionRepository(lambda: session), session, engine + + +def _seed_plan(session: Session, plan_id: str) -> None: + now = "2026-01-01T00:00:00+00:00" + action_name = f"test/action-{plan_id[-4:]}" + session.add( + LifecycleActionModel( + namespaced_name=action_name, + namespace="test", + name=f"action-{plan_id[-4:]}", + description="Test action", + definition_of_done="Done", + strategy_actor="strategy", + execution_actor="execution", + state="available", + tags_json="[]", + created_at=now, + updated_at=now, + ) + ) + session.add( + LifecyclePlanModel( + plan_id=plan_id, + root_plan_id=plan_id, + action_name=action_name, + namespaced_name=f"test/plan-{plan_id[-4:]}", + namespace="test", + description="Test plan", + definition_of_done="Done", + phase="execute", + processing_state="processing", + tags_json="[]", + created_at=now, + updated_at=now, + ) + ) + + +def test_decision_repository_protocol_dependency_methods_have_default_bodies() -> None: + repository = ProtocolOnlyDecisionRepository() # type: ignore[abstract] + + assert repository.record_dependency("source", "target") is None + assert repository.get_influence_edges("plan") == {} + + +def test_record_dependency_ignores_duplicates_and_queries_plan_edges() -> None: + repository, session, engine = _repository() + plan_id = "01J00000000000000000000001" + source_id = "01J00000000000000000000002" + target_id = "01J00000000000000000000003" + outside_plan_id = "01J00000000000000000000004" + outside_decision_id = "01J00000000000000000000005" + + try: + _seed_plan(session, plan_id) + _seed_plan(session, outside_plan_id) + session.commit() + for decision in [ + _decision(source_id, plan_id, 0), + _decision(target_id, plan_id, 1), + _decision(outside_decision_id, outside_plan_id, 0), + ]: + repository.create(decision) + session.commit() + + repository.record_dependency(source_id, target_id) + session.commit() + + repository.record_dependency(source_id, target_id) + session.commit() + + repository.record_dependency(outside_decision_id, target_id) + session.commit() + + stored_edges = session.query(DecisionDependencyModel).all() + assert len(stored_edges) == 2 + assert repository.get_influence_edges(plan_id) == {source_id: [target_id]} + assert repository.get_influence_edges(outside_plan_id) == {} + finally: + session.close() + engine.dispose() + + +def test_record_dependency_surfaces_missing_decision_constraints() -> None: + repository, session, engine = _repository() + plan_id = "01J00000000000000000000001" + target_id = "01J00000000000000000000003" + missing_source_id = "01J00000000000000000000009" + + try: + _seed_plan(session, plan_id) + session.commit() + repository.create(_decision(target_id, plan_id, 0)) + session.commit() + + with pytest.raises(DatabaseError): + repository.record_dependency(missing_source_id, target_id) + + assert session.query(DecisionDependencyModel).count() == 0 + finally: + session.close() + engine.dispose() + + +def test_record_dependency_rejects_self_loops() -> None: + repository, session, engine = _repository() + decision_id = "01J00000000000000000000003" + + try: + with pytest.raises(DatabaseError, match="self-loops"): + repository.record_dependency(decision_id, decision_id) + finally: + session.close() + engine.dispose()