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.
283 lines
9.4 KiB
Python
283 lines
9.4 KiB
Python
"""Helper script for Robot Framework decision persistence integration tests.
|
|
|
|
Each test function exercises a serialization round-trip path for
|
|
the Decision domain model (model_dump / model_validate, JSON) and
|
|
prints a deterministic tag on success.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# Ensure src is importable when run from workspace root
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
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",
|
|
}
|
|
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(d: Decision) -> Decision:
|
|
data = d.model_dump()
|
|
return Decision.model_validate(data)
|
|
|
|
|
|
def _roundtrip_json(d: Decision) -> Decision:
|
|
json_str = d.model_dump_json()
|
|
raw = json.loads(json_str)
|
|
return Decision.model_validate(raw)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test functions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _test_roundtrip_root() -> None:
|
|
"""Root decision survives model_dump round-trip."""
|
|
original = _make_decision()
|
|
restored = _roundtrip_dump(original)
|
|
assert restored.decision_id == original.decision_id
|
|
assert restored.decision_type == DecisionType.PROMPT_DEFINITION
|
|
assert restored.is_root
|
|
assert restored.question == original.question
|
|
assert restored.chosen_option == original.chosen_option
|
|
print("decision-persist-roundtrip-root-ok")
|
|
|
|
|
|
def _test_roundtrip_child() -> None:
|
|
"""Child decision with all fields survives model_dump round-trip."""
|
|
parent_id = str(ULID())
|
|
original = _make_decision(
|
|
parent_decision_id=parent_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework?",
|
|
chosen_option="FastAPI",
|
|
alternatives_considered=["Flask", "Django", "Starlette"],
|
|
confidence_score=0.88,
|
|
rationale="Async support and auto docs",
|
|
)
|
|
restored = _roundtrip_dump(original)
|
|
assert restored.decision_id == original.decision_id
|
|
assert restored.parent_decision_id == parent_id
|
|
assert not restored.is_root
|
|
assert len(restored.alternatives_considered) == 3
|
|
assert restored.confidence_score == 0.88
|
|
assert restored.rationale == "Async support and auto docs"
|
|
print("decision-persist-roundtrip-child-ok")
|
|
|
|
|
|
def _test_snapshot() -> None:
|
|
"""Context snapshot with resources survives round-trip."""
|
|
snap = ContextSnapshot(
|
|
hot_context_hash="sha256:persist_snap",
|
|
hot_context_ref="store://snapshots/persist_snap",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
|
|
ResourceRef(resource_id=str(ULID()), path="src/utils.py"),
|
|
],
|
|
actor_state_ref="checkpoint://actor/persist",
|
|
)
|
|
original = _make_decision(
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
context_snapshot=snap,
|
|
)
|
|
restored = _roundtrip_dump(original)
|
|
assert (
|
|
restored.context_snapshot.hot_context_hash
|
|
== original.context_snapshot.hot_context_hash
|
|
)
|
|
assert len(restored.context_snapshot.relevant_resources) == 2
|
|
assert restored.context_snapshot.actor_state_ref == "checkpoint://actor/persist"
|
|
print("decision-persist-snapshot-ok")
|
|
|
|
|
|
def _test_json_roundtrip() -> None:
|
|
"""Decision survives JSON serialization round-trip."""
|
|
original = _make_decision(
|
|
decision_type=DecisionType.RESOURCE_SELECTION,
|
|
artifacts_produced=[
|
|
ArtifactRef(artifact_path="src/api.py", artifact_type="file"),
|
|
ArtifactRef(artifact_path="patches/fix.patch", artifact_type="patch"),
|
|
],
|
|
)
|
|
restored = _roundtrip_json(original)
|
|
assert restored.decision_id == original.decision_id
|
|
assert restored.decision_type == DecisionType.RESOURCE_SELECTION
|
|
assert len(restored.artifacts_produced) == 2
|
|
print("decision-persist-json-roundtrip-ok")
|
|
|
|
|
|
def _test_correction_chain() -> None:
|
|
"""Correction chain of 3 decisions persists and reconstructs."""
|
|
plan_id = str(ULID())
|
|
decisions: list[Decision] = []
|
|
|
|
original = _make_decision(
|
|
plan_id=plan_id,
|
|
sequence_number=0,
|
|
question="Initial approach?",
|
|
chosen_option="Monolith",
|
|
)
|
|
decisions.append(original)
|
|
|
|
prev_id = original.decision_id
|
|
for i in range(1, 3):
|
|
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}",
|
|
)
|
|
decisions[-1] = decisions[-1].with_superseded_by(correction.decision_id)
|
|
decisions.append(correction)
|
|
prev_id = correction.decision_id
|
|
|
|
restored = [_roundtrip_dump(d) for d in decisions]
|
|
for orig, rest in zip(decisions, 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
|
|
|
|
for idx in range(1, len(restored)):
|
|
assert restored[idx].corrects_decision_id == restored[idx - 1].decision_id
|
|
assert restored[idx - 1].superseded_by == restored[idx].decision_id
|
|
|
|
print("decision-persist-correction-chain-ok")
|
|
|
|
|
|
def _test_tree_reconstruction() -> None:
|
|
"""3-level tree (7 nodes) reconstructs from serialized data."""
|
|
plan_id = str(ULID())
|
|
nodes: list[Decision] = []
|
|
seq = 0
|
|
|
|
root = _make_decision(
|
|
plan_id=plan_id,
|
|
sequence_number=seq,
|
|
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
|
|
|
|
restored = [_roundtrip_dump(n) for n in nodes]
|
|
assert len(restored) == 7
|
|
id_set = {n.decision_id for n in restored}
|
|
root_count = 0
|
|
for n in restored:
|
|
if n.parent_decision_id is None:
|
|
root_count += 1
|
|
else:
|
|
assert n.parent_decision_id in id_set
|
|
assert root_count == 1
|
|
print("decision-persist-tree-reconstruction-ok")
|
|
|
|
|
|
def _test_all_types() -> None:
|
|
"""All decision types survive model_dump round-trip."""
|
|
plan_id = str(ULID())
|
|
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)
|
|
assert restored.decision_id == original.decision_id
|
|
assert restored.decision_type == original.decision_type
|
|
print("decision-persist-all-types-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TESTS = {
|
|
"roundtrip_root": _test_roundtrip_root,
|
|
"roundtrip_child": _test_roundtrip_child,
|
|
"snapshot": _test_snapshot,
|
|
"json_roundtrip": _test_json_roundtrip,
|
|
"correction_chain": _test_correction_chain,
|
|
"tree_reconstruction": _test_tree_reconstruction,
|
|
"all_types": _test_all_types,
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} <test_name>")
|
|
print(f"Available tests: {', '.join(sorted(_TESTS))}")
|
|
sys.exit(1)
|
|
|
|
test_name = sys.argv[1]
|
|
if test_name not in _TESTS:
|
|
print(f"Unknown test: {test_name}")
|
|
print(f"Available: {', '.join(sorted(_TESTS))}")
|
|
sys.exit(1)
|
|
|
|
_TESTS[test_name]()
|