khyari hamza
27bef69ddd
Add decision persistence layer with DecisionRepository, DecisionModel, and Alembic migration. Includes tree queries (BFS traversal via deque, path-to-root), superseded lookup, ordered decision path retrieval, concrete Decision type annotations (via TYPE_CHECKING), and comprehensive test coverage (Behave BDD, Robot Framework, ASV benchmarks). Updated database_schema.md, CHANGELOG.md, and CONTRIBUTORS.md. ISSUES CLOSED: #171
609 lines
21 KiB
Python
609 lines
21 KiB
Python
"""Step definitions for decision_persistence.feature.
|
|
|
|
Tests the DecisionRepository CRUD operations, JSON round-trips,
|
|
tree traversal, correction workflows, and constraint enforcement.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.action import Action, ActionState
|
|
from cleveragents.domain.models.core.decision import (
|
|
ArtifactRef,
|
|
ContextSnapshot,
|
|
Decision,
|
|
DecisionType,
|
|
ResourceRef,
|
|
)
|
|
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,
|
|
DuplicateDecisionError,
|
|
LifecyclePlanRepository,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PLAN_ID = "01HV000000000000000000DP01"
|
|
|
|
|
|
def _setup_db(context: Context) -> None:
|
|
"""Create an in-memory SQLite DB and attach repos."""
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
sm = sessionmaker(bind=engine)
|
|
session = sm()
|
|
context._dp_engine = engine
|
|
context._dp_session = session
|
|
context._dp_factory = lambda: session
|
|
context._dp_repo = DecisionRepository(session_factory=context._dp_factory)
|
|
context._dp_action_repo = ActionRepository(session_factory=context._dp_factory)
|
|
context._dp_plan_repo = LifecyclePlanRepository(
|
|
session_factory=context._dp_factory,
|
|
)
|
|
|
|
|
|
def _make_decision(
|
|
plan_id: str = _PLAN_ID,
|
|
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",
|
|
confidence_score: float | None = None,
|
|
context_snapshot: ContextSnapshot | None = None,
|
|
alternatives_considered: list[str] | None = None,
|
|
artifacts_produced: list[ArtifactRef] | None = None,
|
|
downstream_decision_ids: list[str] | None = None,
|
|
downstream_plan_ids: list[str] | None = None,
|
|
rationale: str = "",
|
|
is_correction: bool = False,
|
|
corrects_decision_id: str | None = None,
|
|
correction_reason: str | None = None,
|
|
superseded_by: str | None = None,
|
|
) -> Decision:
|
|
kwargs: dict[str, Any] = {
|
|
"plan_id": plan_id,
|
|
"sequence_number": sequence_number,
|
|
"decision_type": decision_type,
|
|
"question": question,
|
|
"chosen_option": chosen_option,
|
|
"rationale": rationale,
|
|
"is_correction": is_correction,
|
|
}
|
|
if parent_decision_id is not None:
|
|
kwargs["parent_decision_id"] = parent_decision_id
|
|
if confidence_score is not None:
|
|
kwargs["confidence_score"] = confidence_score
|
|
if context_snapshot is not None:
|
|
kwargs["context_snapshot"] = context_snapshot
|
|
if alternatives_considered is not None:
|
|
kwargs["alternatives_considered"] = alternatives_considered
|
|
if artifacts_produced is not None:
|
|
kwargs["artifacts_produced"] = artifacts_produced
|
|
if downstream_decision_ids is not None:
|
|
kwargs["downstream_decision_ids"] = downstream_decision_ids
|
|
if downstream_plan_ids is not None:
|
|
kwargs["downstream_plan_ids"] = downstream_plan_ids
|
|
if corrects_decision_id is not None:
|
|
kwargs["corrects_decision_id"] = corrects_decision_id
|
|
if correction_reason is not None:
|
|
kwargs["correction_reason"] = correction_reason
|
|
if superseded_by is not None:
|
|
kwargs["superseded_by"] = superseded_by
|
|
return Decision(**kwargs)
|
|
|
|
|
|
def _create_prerequisite_action(context: Context) -> None:
|
|
ns = NamespacedName.parse("local/decision-action")
|
|
action = Action(
|
|
namespaced_name=ns,
|
|
description="Prerequisite action for decision 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._dp_action_repo.create(action)
|
|
context._dp_session.commit()
|
|
|
|
|
|
def _create_prerequisite_plan(context: Context) -> None:
|
|
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-decision",
|
|
tags=[],
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
context._dp_plan_repo.create(plan)
|
|
context._dp_session.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh in-memory decision persistence database")
|
|
def step_fresh_db(context: Context) -> None:
|
|
_setup_db(context)
|
|
|
|
|
|
@given('a prerequisite action "{action_name}" exists for decisions')
|
|
def step_prereq_action(context: Context, action_name: str) -> None:
|
|
_create_prerequisite_action(context)
|
|
|
|
|
|
@given("a prerequisite plan exists for decisions")
|
|
def step_prereq_plan(context: Context) -> None:
|
|
_create_prerequisite_plan(context)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create and retrieve
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a new root decision with sequence {seq:d}")
|
|
def step_new_root(context: Context, seq: int) -> None:
|
|
context._dp_decision = _make_decision(sequence_number=seq)
|
|
|
|
|
|
@given("a persisted root decision")
|
|
def step_persisted_root(context: Context) -> None:
|
|
d = _make_decision()
|
|
context._dp_repo.create(d)
|
|
context._dp_session.commit()
|
|
context._dp_root_decision = d
|
|
context._dp_root_id = d.decision_id
|
|
|
|
|
|
@given('a new child decision of type "{dtype}" with sequence {seq:d}')
|
|
def step_new_child(context: Context, dtype: str, seq: int) -> None:
|
|
context._dp_decision = _make_decision(
|
|
parent_decision_id=context._dp_root_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType(dtype),
|
|
)
|
|
|
|
|
|
@given("a new root decision with all fields populated")
|
|
def step_new_full_root(context: Context) -> None:
|
|
context._dp_decision = _make_decision(
|
|
question="Full question?",
|
|
chosen_option="Full option",
|
|
confidence_score=0.85,
|
|
rationale="Thorough analysis",
|
|
alternatives_considered=["A", "B"],
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="sha256:full",
|
|
hot_context_ref="store://full",
|
|
relevant_resources=[ResourceRef(resource_id=str(ULID()))],
|
|
actor_state_ref="checkpoint://full",
|
|
),
|
|
)
|
|
|
|
|
|
@when("I persist the decision via the decision repository")
|
|
def step_persist_decision(context: Context) -> None:
|
|
context._dp_repo.create(context._dp_decision)
|
|
context._dp_session.commit()
|
|
|
|
|
|
@then("I can retrieve the decision by its ID")
|
|
def step_retrieve_by_id(context: Context) -> None:
|
|
d = context._dp_repo.get(context._dp_decision.decision_id)
|
|
assert d is not None, "Decision not found"
|
|
context._dp_retrieved = d
|
|
|
|
|
|
@then('the persisted decision type should be "{dtype}"')
|
|
def step_check_type(context: Context, dtype: str) -> None:
|
|
assert str(context._dp_retrieved.decision_type) == dtype
|
|
|
|
|
|
@then("the persisted decision should be a root")
|
|
def step_check_is_root(context: Context) -> None:
|
|
assert context._dp_retrieved.is_root
|
|
|
|
|
|
@then("the persisted decision should not be a root")
|
|
def step_check_not_root(context: Context) -> None:
|
|
assert not context._dp_retrieved.is_root
|
|
|
|
|
|
@then("the persisted decision parent should match the root")
|
|
def step_check_parent(context: Context) -> None:
|
|
assert context._dp_retrieved.parent_decision_id == context._dp_root_id
|
|
|
|
|
|
@then("the persisted decision question should match")
|
|
def step_check_question(context: Context) -> None:
|
|
assert context._dp_retrieved.question == context._dp_decision.question
|
|
|
|
|
|
@then("the persisted decision chosen_option should match")
|
|
def step_check_chosen(context: Context) -> None:
|
|
assert context._dp_retrieved.chosen_option == context._dp_decision.chosen_option
|
|
|
|
|
|
@then("the persisted decision confidence_score should be {score}")
|
|
def step_check_confidence(context: Context, score: str) -> None:
|
|
expected = float(score)
|
|
assert context._dp_retrieved.confidence_score == expected, (
|
|
f"Expected {expected}, got {context._dp_retrieved.confidence_score}"
|
|
)
|
|
|
|
|
|
@then("the persisted decision rationale should match")
|
|
def step_check_rationale(context: Context) -> None:
|
|
assert context._dp_retrieved.rationale == context._dp_decision.rationale
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# JSON round-trip
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a new decision with {count:d} alternatives")
|
|
def step_new_with_alts(context: Context, count: int) -> None:
|
|
alts = [f"Alternative {i}" for i in range(count)]
|
|
context._dp_decision = _make_decision(alternatives_considered=alts)
|
|
|
|
|
|
@then("the persisted decision should have {count:d} alternatives")
|
|
def step_check_alt_count(context: Context, count: int) -> None:
|
|
assert len(context._dp_retrieved.alternatives_considered) == count
|
|
|
|
|
|
@given("a new decision with a populated context snapshot")
|
|
def step_new_with_snapshot(context: Context) -> None:
|
|
snap = ContextSnapshot(
|
|
hot_context_hash="sha256:persist-test",
|
|
hot_context_ref="store://persist-test",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
|
|
ResourceRef(resource_id=str(ULID())),
|
|
],
|
|
actor_state_ref="checkpoint://persist",
|
|
)
|
|
context._dp_decision = _make_decision(context_snapshot=snap)
|
|
|
|
|
|
@then('the persisted decision context hash should be "{expected}"')
|
|
def step_check_ctx_hash(context: Context, expected: str) -> None:
|
|
assert context._dp_retrieved.context_snapshot.hot_context_hash == expected
|
|
|
|
|
|
@then("the persisted decision context should have {count:d} resources")
|
|
def step_check_ctx_resources(context: Context, count: int) -> None:
|
|
assert len(context._dp_retrieved.context_snapshot.relevant_resources) == count
|
|
|
|
|
|
@given("a new decision with {count:d} artifacts")
|
|
def step_new_with_artifacts(context: Context, count: int) -> None:
|
|
arts = [
|
|
ArtifactRef(artifact_path=f"src/file_{i}.py", artifact_type="file")
|
|
for i in range(count)
|
|
]
|
|
context._dp_decision = _make_decision(artifacts_produced=arts)
|
|
|
|
|
|
@then("the persisted decision should have {count:d} artifacts")
|
|
def step_check_artifact_count(context: Context, count: int) -> None:
|
|
assert len(context._dp_retrieved.artifacts_produced) == count
|
|
|
|
|
|
@given("a new decision with downstream IDs")
|
|
def step_new_with_downstream(context: Context) -> None:
|
|
context._dp_decision = _make_decision(
|
|
downstream_decision_ids=[str(ULID()), str(ULID())],
|
|
downstream_plan_ids=[str(ULID())],
|
|
)
|
|
|
|
|
|
@then("the persisted decision should have downstream decision IDs")
|
|
def step_check_downstream_decisions(context: Context) -> None:
|
|
assert len(context._dp_retrieved.downstream_decision_ids) > 0
|
|
|
|
|
|
@then("the persisted decision should have downstream plan IDs")
|
|
def step_check_downstream_plans(context: Context) -> None:
|
|
assert len(context._dp_retrieved.downstream_plan_ids) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Get by plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("{count:d} persisted decisions for the same plan")
|
|
def step_n_decisions(context: Context, count: int) -> None:
|
|
for i in range(count):
|
|
dt = DecisionType.PROMPT_DEFINITION if i == 0 else DecisionType.STRATEGY_CHOICE
|
|
parent = None if i == 0 else context._dp_first_id
|
|
d = _make_decision(
|
|
sequence_number=i,
|
|
decision_type=dt,
|
|
parent_decision_id=parent,
|
|
)
|
|
context._dp_repo.create(d)
|
|
if i == 0:
|
|
context._dp_first_id = d.decision_id
|
|
context._dp_session.commit()
|
|
|
|
|
|
@when("I get decisions by plan ID")
|
|
def step_get_by_plan(context: Context) -> None:
|
|
context._dp_plan_decisions = context._dp_repo.get_by_plan(_PLAN_ID)
|
|
|
|
|
|
@then("I should get {count:d} decisions in sequence order")
|
|
def step_check_plan_decisions(context: Context, count: int) -> None:
|
|
decisions = context._dp_plan_decisions
|
|
assert len(decisions) == count, f"Expected {count}, got {len(decisions)}"
|
|
for i in range(len(decisions) - 1):
|
|
assert decisions[i].sequence_number <= decisions[i + 1].sequence_number
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Get tree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a persisted decision tree with 3 levels")
|
|
def step_decision_tree(context: Context) -> None:
|
|
# Root (level 0)
|
|
root = _make_decision(sequence_number=0)
|
|
context._dp_repo.create(root)
|
|
context._dp_tree_root_id = root.decision_id
|
|
|
|
# Level 1 — two children
|
|
child1 = _make_decision(
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=1,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
context._dp_repo.create(child1)
|
|
|
|
child2 = _make_decision(
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=2,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
context._dp_repo.create(child2)
|
|
|
|
# Level 2 — one grandchild under child1
|
|
grandchild = _make_decision(
|
|
parent_decision_id=child1.decision_id,
|
|
sequence_number=3,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
)
|
|
context._dp_repo.create(grandchild)
|
|
context._dp_tree_leaf_id = grandchild.decision_id
|
|
|
|
context._dp_session.commit()
|
|
|
|
|
|
@when("I get the decision tree from the root")
|
|
def step_get_tree(context: Context) -> None:
|
|
context._dp_tree = context._dp_repo.get_tree(context._dp_tree_root_id)
|
|
|
|
|
|
@then("the tree should have {count:d} decisions")
|
|
def step_check_tree_count(context: Context, count: int) -> None:
|
|
assert len(context._dp_tree) == count, (
|
|
f"Expected {count}, got {len(context._dp_tree)}"
|
|
)
|
|
|
|
|
|
@then("the first decision in the tree should be the root")
|
|
def step_check_tree_root(context: Context) -> None:
|
|
assert context._dp_tree[0].decision_id == context._dp_tree_root_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Get path to root
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I get the path from a leaf to root")
|
|
def step_get_path(context: Context) -> None:
|
|
context._dp_path = context._dp_repo.get_path_to_root(
|
|
context._dp_tree_leaf_id,
|
|
)
|
|
|
|
|
|
@then("the path should start with the leaf")
|
|
def step_check_path_leaf(context: Context) -> None:
|
|
assert context._dp_path[0].decision_id == context._dp_tree_leaf_id
|
|
|
|
|
|
@then("the path should end with the root")
|
|
def step_check_path_root(context: Context) -> None:
|
|
assert context._dp_path[-1].decision_id == context._dp_tree_root_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Correction and superseded
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a correction decision that corrects the root")
|
|
def step_correction_decision(context: Context) -> None:
|
|
context._dp_decision = _make_decision(
|
|
parent_decision_id=context._dp_root_id,
|
|
sequence_number=1,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
is_correction=True,
|
|
corrects_decision_id=context._dp_root_id,
|
|
correction_reason="Root was suboptimal",
|
|
)
|
|
|
|
|
|
@then("the persisted decision is_correction should be true")
|
|
def step_check_is_correction(context: Context) -> None:
|
|
assert context._dp_retrieved.is_correction
|
|
|
|
|
|
@when("I update the root decision superseded_by to a new ULID")
|
|
def step_update_superseded(context: Context) -> None:
|
|
new_id = str(ULID())
|
|
context._dp_superseding_id = new_id
|
|
context._dp_repo.update_superseded_by(context._dp_root_id, new_id)
|
|
context._dp_session.commit()
|
|
|
|
|
|
@then("the root decision should be marked as superseded")
|
|
def step_check_superseded(context: Context) -> None:
|
|
d = context._dp_repo.get(context._dp_root_id)
|
|
assert d is not None
|
|
assert d.is_superseded
|
|
assert d.superseded_by == context._dp_superseding_id
|
|
|
|
|
|
@given("a persisted root decision that is superseded")
|
|
def step_superseded_root(context: Context) -> None:
|
|
d = _make_decision(superseded_by=str(ULID()))
|
|
context._dp_repo.create(d)
|
|
context._dp_session.commit()
|
|
context._dp_root_id = d.decision_id
|
|
|
|
|
|
@when("I get superseded decisions for the plan")
|
|
def step_get_superseded(context: Context) -> None:
|
|
context._dp_superseded = context._dp_repo.get_superseded(_PLAN_ID)
|
|
|
|
|
|
@then("I should get {count:d} superseded decision")
|
|
def step_check_superseded_count(context: Context, count: int) -> None:
|
|
assert len(context._dp_superseded) == count
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List by type
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('{count:d} persisted decisions of type "{dtype}"')
|
|
def step_n_typed_decisions(context: Context, count: int, dtype: str) -> None:
|
|
dt = DecisionType(dtype)
|
|
for i in range(count):
|
|
# Use sequence offset to avoid collision with any prior root
|
|
seq = 10 + i
|
|
parent = None
|
|
if dt != DecisionType.PROMPT_DEFINITION:
|
|
# Need a root first
|
|
if not hasattr(context, "_dp_type_root_id"):
|
|
root = _make_decision(sequence_number=9)
|
|
context._dp_repo.create(root)
|
|
context._dp_session.commit()
|
|
context._dp_type_root_id = root.decision_id
|
|
parent = context._dp_type_root_id
|
|
d = _make_decision(
|
|
parent_decision_id=parent,
|
|
sequence_number=seq,
|
|
decision_type=dt,
|
|
)
|
|
context._dp_repo.create(d)
|
|
context._dp_session.commit()
|
|
|
|
|
|
@given('{count:d} persisted decision of type "{dtype}"')
|
|
def step_one_typed_decision(context: Context, count: int, dtype: str) -> None:
|
|
step_n_typed_decisions(context, count, dtype)
|
|
|
|
|
|
@when('I list decisions by type "{dtype}"')
|
|
def step_list_by_type(context: Context, dtype: str) -> None:
|
|
context._dp_typed = context._dp_repo.list_by_type(_PLAN_ID, dtype)
|
|
|
|
|
|
@then("I should get {count:d} decisions of that type")
|
|
def step_check_typed_count(context: Context, count: int) -> None:
|
|
assert len(context._dp_typed) == count, (
|
|
f"Expected {count}, got {len(context._dp_typed)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I delete the root decision")
|
|
def step_delete_root(context: Context) -> None:
|
|
context._dp_delete_result = context._dp_repo.delete(context._dp_root_id)
|
|
context._dp_session.commit()
|
|
|
|
|
|
@then("the root decision should no longer exist")
|
|
def step_check_deleted(context: Context) -> None:
|
|
assert context._dp_delete_result is True
|
|
d = context._dp_repo.get(context._dp_root_id)
|
|
assert d is None
|
|
|
|
|
|
@when("I try to delete a non-existent decision")
|
|
def step_delete_missing(context: Context) -> None:
|
|
context._dp_delete_result = context._dp_repo.delete(str(ULID()))
|
|
|
|
|
|
@then("the decision delete result should be false")
|
|
def step_check_delete_false(context: Context) -> None:
|
|
assert context._dp_delete_result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Duplicate detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I try to persist the same decision again")
|
|
def step_persist_duplicate(context: Context) -> None:
|
|
try:
|
|
context._dp_repo.create(context._dp_root_decision)
|
|
context._dp_session.commit()
|
|
context._dp_dup_error = None
|
|
except DuplicateDecisionError as exc:
|
|
context._dp_session.rollback()
|
|
context._dp_dup_error = exc
|
|
|
|
|
|
@then("a DuplicateDecisionError should be raised")
|
|
def step_check_duplicate(context: Context) -> None:
|
|
assert context._dp_dup_error is not None
|
|
assert isinstance(context._dp_dup_error, DuplicateDecisionError)
|