brent.edwards
32b81793a2
Rename our serialization-focused decision persistence suites to *_serialization to avoid add/add conflicts with the repository-based decision persistence suites that landed on master independently: - decision_persistence.feature -> decision_persistence_serialization.feature - decision_persistence_steps.py -> decision_persistence_serialization_steps.py - decision_persistence_bench.py -> decision_persistence_serialization_bench.py - decision_persistence.robot -> decision_persistence_serialization.robot - helper_decision_persistence.py -> helper_decision_persistence_serialization.py Updated robot helper path, step docstring, and testing.md references.
523 lines
19 KiB
Python
523 lines
19 KiB
Python
"""Step definitions for decision_persistence_serialization.feature.
|
|
|
|
Tests Decision serialization round-trips (model_dump / model_validate,
|
|
JSON), context-snapshot persistence, correction-chain reconstruction,
|
|
and decision-tree reconstruction from serialized data.
|
|
|
|
All step names are prefixed with ``decision persistence`` to avoid
|
|
AmbiguousStep conflicts with decision_model_steps.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.decision import (
|
|
ArtifactRef,
|
|
ContextSnapshot,
|
|
Decision,
|
|
DecisionType,
|
|
ResourceRef,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ALL_DECISION_TYPES: list[DecisionType] = list(DecisionType)
|
|
|
|
|
|
def _make_decision(**overrides: Any) -> Decision:
|
|
"""Build a Decision with sensible defaults, merged with *overrides*."""
|
|
defaults: dict[str, Any] = {
|
|
"plan_id": str(ULID()),
|
|
"sequence_number": 0,
|
|
"decision_type": DecisionType.PROMPT_DEFINITION,
|
|
"question": "What approach should we take?",
|
|
"chosen_option": "Build a REST API",
|
|
}
|
|
# prompt_definition must not have a parent
|
|
dt = overrides.get("decision_type", defaults["decision_type"])
|
|
if dt != DecisionType.PROMPT_DEFINITION and "parent_decision_id" not in overrides:
|
|
defaults["parent_decision_id"] = str(ULID())
|
|
defaults.update(overrides)
|
|
return Decision(**defaults)
|
|
|
|
|
|
def _roundtrip_dump(decision: Decision) -> Decision:
|
|
"""Serialise via model_dump and deserialise back."""
|
|
data = decision.model_dump()
|
|
return Decision.model_validate(data)
|
|
|
|
|
|
def _roundtrip_json(decision: Decision) -> Decision:
|
|
"""Serialise to JSON string and deserialise back."""
|
|
json_str = decision.model_dump_json()
|
|
raw = json.loads(json_str)
|
|
return Decision.model_validate(raw)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a decision persistence plan ULID")
|
|
def step_dp_plan_ulid(context: Context) -> None:
|
|
context.dp_plan_id = str(ULID())
|
|
|
|
|
|
@given("a decision persistence parent ULID")
|
|
def step_dp_parent_ulid(context: Context) -> None:
|
|
context.dp_parent_id = str(ULID())
|
|
|
|
|
|
@given("a decision persistence corrected decision ULID")
|
|
def step_dp_corrected_ulid(context: Context) -> None:
|
|
context.dp_corrected_id = str(ULID())
|
|
|
|
|
|
@given("a decision persistence context snapshot with {count:d} resources")
|
|
def step_dp_snapshot_with_resources(context: Context, count: int) -> None:
|
|
resources = [
|
|
ResourceRef(resource_id=str(ULID()), path=f"src/file_{i}.py")
|
|
for i in range(count)
|
|
]
|
|
context.dp_snapshot = ContextSnapshot(
|
|
hot_context_hash="sha256:persist_test",
|
|
hot_context_ref="store://snapshots/persist_test",
|
|
relevant_resources=resources,
|
|
actor_state_ref="checkpoint://actor/persist",
|
|
)
|
|
|
|
|
|
@given('a decision persistence snapshot with actor state ref "{ref}"')
|
|
def step_dp_snapshot_actor_ref(context: Context, ref: str) -> None:
|
|
context.dp_snapshot = ContextSnapshot(
|
|
hot_context_hash="sha256:actor_ref_test",
|
|
hot_context_ref="store://snapshots/actor_ref_test",
|
|
relevant_resources=[ResourceRef(resource_id=str(ULID()))],
|
|
actor_state_ref=ref,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I decision persistence create a root prompt_definition with sequence {seq:d}")
|
|
def step_dp_create_root(context: Context, seq: int) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a full strategy_choice child at sequence {seq:d}")
|
|
def step_dp_create_full_child(context: Context, seq: int) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
parent_decision_id=context.dp_parent_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework?",
|
|
chosen_option="FastAPI",
|
|
alternatives_considered=["Flask", "Django", "Starlette"],
|
|
confidence_score=0.88,
|
|
rationale="FastAPI has async support",
|
|
)
|
|
|
|
|
|
@when("I decision persistence create an implementation_choice with the snapshot")
|
|
def step_dp_create_impl_with_snapshot(context: Context) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
parent_decision_id=context.dp_parent_id,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
context_snapshot=context.dp_snapshot,
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a strategy_choice with the snapshot")
|
|
def step_dp_create_strategy_with_snapshot(context: Context) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
parent_decision_id=context.dp_parent_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
context_snapshot=context.dp_snapshot,
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a decision with {count:d} artifacts")
|
|
def step_dp_create_with_artifacts(context: Context, count: int) -> None:
|
|
artifacts = [
|
|
ArtifactRef(artifact_path=f"src/artifact_{i}.py", artifact_type="file")
|
|
for i in range(count)
|
|
]
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
parent_decision_id=context.dp_parent_id,
|
|
decision_type=DecisionType.TOOL_INVOCATION,
|
|
artifacts_produced=artifacts,
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a decision with no alternatives")
|
|
def step_dp_create_no_alternatives(context: Context) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
alternatives_considered=[],
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a decision with confidence {value}")
|
|
def step_dp_create_with_confidence(context: Context, value: str) -> None:
|
|
conf: float | None = None if value == "None" else float(value)
|
|
kwargs: dict[str, Any] = {
|
|
"plan_id": context.dp_plan_id,
|
|
"decision_type": DecisionType.PROMPT_DEFINITION,
|
|
}
|
|
if conf is not None:
|
|
kwargs["confidence_score"] = conf
|
|
context.dp_decision = _make_decision(**kwargs)
|
|
|
|
|
|
@when("I decision persistence create a decision with {count:d} downstream IDs")
|
|
def step_dp_create_downstream_ids(context: Context, count: int) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
downstream_decision_ids=[str(ULID()) for _ in range(count)],
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a decision with {count:d} downstream plan IDs")
|
|
def step_dp_create_downstream_plan_ids(context: Context, count: int) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
downstream_plan_ids=[str(ULID()) for _ in range(count)],
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a correction decision")
|
|
def step_dp_create_correction(context: Context) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
is_correction=True,
|
|
corrects_decision_id=context.dp_corrected_id,
|
|
correction_reason="Original approach was suboptimal",
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a decision superseded by another")
|
|
def step_dp_create_superseded(context: Context) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
superseded_by=str(ULID()),
|
|
)
|
|
|
|
|
|
@when("I decision persistence create a decision with a {length:d} character rationale")
|
|
def step_dp_create_long_rationale(context: Context, length: int) -> None:
|
|
context.dp_decision = _make_decision(
|
|
plan_id=context.dp_plan_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
rationale="R" * length,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — round-trip operations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I decision persistence round-trip the decision through model_dump")
|
|
def step_dp_roundtrip_dump(context: Context) -> None:
|
|
context.dp_restored = _roundtrip_dump(context.dp_decision)
|
|
|
|
|
|
@when("I decision persistence round-trip the decision through JSON")
|
|
def step_dp_roundtrip_json(context: Context) -> None:
|
|
context.dp_restored = _roundtrip_json(context.dp_decision)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — correction chain
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I decision persistence build a correction chain of {count:d} decisions")
|
|
def step_dp_correction_chain(context: Context, count: int) -> None:
|
|
plan_id = context.dp_plan_id
|
|
decisions: list[Decision] = []
|
|
|
|
# First decision: the original
|
|
original = _make_decision(
|
|
plan_id=plan_id,
|
|
sequence_number=0,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
question="Initial approach?",
|
|
chosen_option="Monolith",
|
|
)
|
|
decisions.append(original)
|
|
|
|
prev_id = original.decision_id
|
|
for i in range(1, count):
|
|
correction = _make_decision(
|
|
plan_id=plan_id,
|
|
sequence_number=i,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question=f"Correction {i}?",
|
|
chosen_option=f"Approach v{i + 1}",
|
|
is_correction=True,
|
|
corrects_decision_id=prev_id,
|
|
correction_reason=f"Reason {i}",
|
|
)
|
|
# Mark previous as superseded
|
|
decisions[-1] = decisions[-1].with_superseded_by(correction.decision_id)
|
|
decisions.append(correction)
|
|
prev_id = correction.decision_id
|
|
|
|
context.dp_chain = decisions
|
|
|
|
|
|
@when("I decision persistence round-trip all decisions through model_dump")
|
|
def step_dp_roundtrip_all(context: Context) -> None:
|
|
context.dp_chain_restored = [_roundtrip_dump(d) for d in context.dp_chain]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — decision tree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I decision persistence build a 3-level decision tree")
|
|
def step_dp_build_tree(context: Context) -> None:
|
|
"""Build: root -> 2 children -> 2 grandchildren each = 7 nodes."""
|
|
plan_id = context.dp_plan_id
|
|
nodes: list[Decision] = []
|
|
seq = 0
|
|
|
|
root = _make_decision(
|
|
plan_id=plan_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
question="Root?",
|
|
chosen_option="Start",
|
|
)
|
|
nodes.append(root)
|
|
seq += 1
|
|
|
|
for ci in range(2):
|
|
child = _make_decision(
|
|
plan_id=plan_id,
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question=f"Child {ci}?",
|
|
chosen_option=f"Option C{ci}",
|
|
)
|
|
nodes.append(child)
|
|
seq += 1
|
|
|
|
for gi in range(2):
|
|
grandchild = _make_decision(
|
|
plan_id=plan_id,
|
|
parent_decision_id=child.decision_id,
|
|
sequence_number=seq,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
question=f"Grandchild {ci}.{gi}?",
|
|
chosen_option=f"Impl {ci}.{gi}",
|
|
)
|
|
nodes.append(grandchild)
|
|
seq += 1
|
|
|
|
context.dp_tree = nodes
|
|
|
|
|
|
@when("I decision persistence serialize and deserialize all tree nodes")
|
|
def step_dp_roundtrip_tree(context: Context) -> None:
|
|
context.dp_tree_restored = [_roundtrip_dump(n) for n in context.dp_tree]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps — all types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I decision persistence create and round-trip all decision types")
|
|
def step_dp_all_types(context: Context) -> None:
|
|
plan_id = context.dp_plan_id
|
|
results: list[tuple[Decision, Decision]] = []
|
|
for i, dt in enumerate(_ALL_DECISION_TYPES):
|
|
kwargs: dict[str, Any] = {
|
|
"plan_id": plan_id,
|
|
"sequence_number": i,
|
|
"decision_type": dt,
|
|
"question": f"Q for {dt.value}?",
|
|
"chosen_option": f"A for {dt.value}",
|
|
}
|
|
if dt != DecisionType.PROMPT_DEFINITION:
|
|
kwargs["parent_decision_id"] = str(ULID())
|
|
original = Decision(**kwargs)
|
|
restored = _roundtrip_dump(original)
|
|
results.append((original, restored))
|
|
context.dp_all_types_results = results
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the decision persistence restored decision should match the original")
|
|
def step_dp_match_original(context: Context) -> None:
|
|
orig = context.dp_decision
|
|
rest = context.dp_restored
|
|
assert rest.decision_id == orig.decision_id
|
|
assert rest.plan_id == orig.plan_id
|
|
assert rest.parent_decision_id == orig.parent_decision_id
|
|
assert rest.sequence_number == orig.sequence_number
|
|
assert rest.decision_type == orig.decision_type
|
|
assert rest.question == orig.question
|
|
assert rest.chosen_option == orig.chosen_option
|
|
assert rest.confidence_score == orig.confidence_score
|
|
assert rest.is_correction == orig.is_correction
|
|
assert rest.corrects_decision_id == orig.corrects_decision_id
|
|
assert rest.superseded_by == orig.superseded_by
|
|
assert rest.rationale == orig.rationale
|
|
|
|
|
|
@then("the decision persistence restored alternatives should have {count:d} entries")
|
|
def step_dp_alternatives_count(context: Context, count: int) -> None:
|
|
assert len(context.dp_restored.alternatives_considered) == count
|
|
|
|
|
|
@then("the decision persistence restored snapshot hash should match")
|
|
def step_dp_snapshot_hash_match(context: Context) -> None:
|
|
assert (
|
|
context.dp_restored.context_snapshot.hot_context_hash
|
|
== context.dp_decision.context_snapshot.hot_context_hash
|
|
)
|
|
|
|
|
|
@then("the decision persistence restored snapshot should have {count:d} resources")
|
|
def step_dp_snapshot_resource_count(context: Context, count: int) -> None:
|
|
assert len(context.dp_restored.context_snapshot.relevant_resources) == count
|
|
|
|
|
|
@then("the decision persistence restored snapshot hash should be empty")
|
|
def step_dp_snapshot_hash_empty(context: Context) -> None:
|
|
assert context.dp_restored.context_snapshot.hot_context_hash == ""
|
|
|
|
|
|
@then('the decision persistence restored actor state ref should be "{ref}"')
|
|
def step_dp_actor_state_ref(context: Context, ref: str) -> None:
|
|
assert context.dp_restored.context_snapshot.actor_state_ref == ref
|
|
|
|
|
|
@then("the decision persistence restored decision should have {count:d} artifacts")
|
|
def step_dp_artifact_count(context: Context, count: int) -> None:
|
|
assert len(context.dp_restored.artifacts_produced) == count
|
|
|
|
|
|
@then("the decision persistence restored is_correction should be true")
|
|
def step_dp_is_correction_true(context: Context) -> None:
|
|
assert context.dp_restored.is_correction is True
|
|
|
|
|
|
@then("the decision persistence restored corrects_decision_id should match")
|
|
def step_dp_corrects_id_match(context: Context) -> None:
|
|
assert (
|
|
context.dp_restored.corrects_decision_id
|
|
== context.dp_decision.corrects_decision_id
|
|
)
|
|
|
|
|
|
@then("the decision persistence restored is_superseded should be true")
|
|
def step_dp_is_superseded_true(context: Context) -> None:
|
|
assert context.dp_restored.is_superseded is True
|
|
|
|
|
|
@then("the decision persistence correction chain should be reconstructable")
|
|
def step_dp_chain_reconstructable(context: Context) -> None:
|
|
originals = context.dp_chain
|
|
restored = context.dp_chain_restored
|
|
assert len(restored) == len(originals)
|
|
|
|
for orig, rest in zip(originals, restored, strict=True):
|
|
assert rest.decision_id == orig.decision_id
|
|
assert rest.is_correction == orig.is_correction
|
|
assert rest.corrects_decision_id == orig.corrects_decision_id
|
|
assert rest.superseded_by == orig.superseded_by
|
|
|
|
# Verify the chain linkage
|
|
for i in range(1, len(restored)):
|
|
assert restored[i].corrects_decision_id == restored[i - 1].decision_id
|
|
assert restored[i - 1].superseded_by == restored[i].decision_id
|
|
|
|
|
|
@then("the decision persistence tree parent links should be intact")
|
|
def step_dp_tree_parent_links(context: Context) -> None:
|
|
nodes = context.dp_tree_restored
|
|
id_set = {n.decision_id for n in nodes}
|
|
root_count = 0
|
|
for n in nodes:
|
|
if n.parent_decision_id is None:
|
|
root_count += 1
|
|
else:
|
|
assert n.parent_decision_id in id_set, (
|
|
f"Parent {n.parent_decision_id} not found in tree"
|
|
)
|
|
assert root_count == 1, f"Expected 1 root, found {root_count}"
|
|
|
|
|
|
@then("the decision persistence tree should have {count:d} nodes")
|
|
def step_dp_tree_node_count(context: Context, count: int) -> None:
|
|
assert len(context.dp_tree_restored) == count
|
|
|
|
|
|
@then("the decision persistence restored should have {count:d} downstream IDs")
|
|
def step_dp_downstream_ids_count(context: Context, count: int) -> None:
|
|
assert len(context.dp_restored.downstream_decision_ids) == count
|
|
|
|
|
|
@then("the decision persistence restored should have {count:d} downstream plan IDs")
|
|
def step_dp_downstream_plan_ids_count(context: Context, count: int) -> None:
|
|
assert len(context.dp_restored.downstream_plan_ids) == count
|
|
|
|
|
|
@then("the decision persistence restored confidence should be {value}")
|
|
def step_dp_confidence_value(context: Context, value: str) -> None:
|
|
if value == "None":
|
|
assert context.dp_restored.confidence_score is None
|
|
else:
|
|
assert context.dp_restored.confidence_score == float(value)
|
|
|
|
|
|
@then("all decision persistence round-trips should succeed")
|
|
def step_dp_all_types_ok(context: Context) -> None:
|
|
results = context.dp_all_types_results
|
|
assert len(results) == len(DecisionType)
|
|
for original, restored in results:
|
|
assert restored.decision_id == original.decision_id
|
|
assert restored.decision_type == original.decision_type
|
|
assert restored.question == original.question
|
|
assert restored.chosen_option == original.chosen_option
|
|
|
|
|
|
@then("the decision persistence restored rationale length should be {length:d}")
|
|
def step_dp_rationale_length(context: Context, length: int) -> None:
|
|
assert len(context.dp_restored.rationale) == length
|