fix(decision_service): persist decision dependency edges to database #8182
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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}"
|
||||
)
|
||||
@@ -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"
|
||||
)
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Shared Behave step helpers."""
|
||||
@@ -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
|
||||
@@ -156,6 +156,7 @@ strictDictionaryInference = true
|
||||
strictSetInference = true
|
||||
reportMissingImports = true
|
||||
reportMissingTypeStubs = false
|
||||
reportMissingModuleSource = false
|
||||
reportMissingTypeArgument = true
|
||||
reportIncompatibleMethodOverride = true
|
||||
reportIncompatibleVariableOverride = true
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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]
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user