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.
381 lines
11 KiB
Python
381 lines
11 KiB
Python
"""Helper script for Robot Framework decision persistence smoke tests.
|
|
|
|
Usage:
|
|
python robot/helper_decision_persistence.py <subcommand>
|
|
|
|
Subcommands:
|
|
create-retrieve Create + retrieve a root decision
|
|
create-child Create a child under a root
|
|
json-roundtrip Verify JSON fields survive persistence
|
|
get-by-plan Get all decisions for a plan in order
|
|
tree-bfs Build a 3-level tree, retrieve via BFS
|
|
path-to-root Walk from leaf to root
|
|
superseded Mark a decision as superseded
|
|
delete Delete a decision
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
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 collections.abc import Callable
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, 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,
|
|
LifecyclePlanRepository,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PLAN_ID = "01HV000000000000000000RP01"
|
|
|
|
|
|
def _setup() -> tuple[Session, Callable[[], Session]]:
|
|
"""Create an in-memory SQLite DB with all tables."""
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
sm = sessionmaker(bind=engine)
|
|
session = sm()
|
|
factory: Callable[[], Session] = lambda: session # noqa: E731
|
|
return session, factory
|
|
|
|
|
|
def _create_prerequisites(session: Session, factory: Callable[[], Session]) -> None:
|
|
"""Create the action + plan needed to satisfy FK constraints."""
|
|
action_repo = ActionRepository(session_factory=factory)
|
|
action = Action(
|
|
namespaced_name=NamespacedName.parse("local/robot-decision-action"),
|
|
description="Robot prerequisite action",
|
|
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),
|
|
)
|
|
action_repo.create(action)
|
|
session.commit()
|
|
|
|
plan_repo = LifecyclePlanRepository(session_factory=factory)
|
|
now = datetime(2026, 3, 1, tzinfo=UTC)
|
|
plan = Plan(
|
|
identity=PlanIdentity(plan_id=_PLAN_ID, attempt=1),
|
|
namespaced_name=NamespacedName(namespace="local", name="robot-plan"),
|
|
action_name="local/robot-decision-action",
|
|
description="Robot decision test plan",
|
|
definition_of_done="All tests 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="robot-test",
|
|
tags=[],
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
plan_repo.create(plan)
|
|
session.commit()
|
|
|
|
|
|
def _make_decision(
|
|
plan_id: str = _PLAN_ID,
|
|
parent_decision_id: str | None = None,
|
|
sequence_number: int = 0,
|
|
decision_type: DecisionType = DecisionType.PROMPT_DEFINITION,
|
|
**kwargs: Any,
|
|
) -> Decision:
|
|
return Decision(
|
|
plan_id=plan_id,
|
|
parent_decision_id=parent_decision_id,
|
|
sequence_number=sequence_number,
|
|
decision_type=decision_type,
|
|
question=kwargs.get("question", "What approach?"),
|
|
chosen_option=kwargs.get("chosen_option", "Build API"),
|
|
**{k: v for k, v in kwargs.items() if k not in ("question", "chosen_option")},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _create_retrieve() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
d = _make_decision()
|
|
repo.create(d)
|
|
session.commit()
|
|
|
|
got = repo.get(d.decision_id)
|
|
assert got is not None, "decision not found"
|
|
assert got.decision_id == d.decision_id
|
|
assert got.is_root
|
|
assert str(got.decision_type) == "prompt_definition"
|
|
print("create-retrieve-ok")
|
|
|
|
|
|
def _create_child() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
root = _make_decision()
|
|
repo.create(root)
|
|
session.commit()
|
|
|
|
child = _make_decision(
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=1,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
repo.create(child)
|
|
session.commit()
|
|
|
|
got = repo.get(child.decision_id)
|
|
assert got is not None
|
|
assert not got.is_root
|
|
assert got.parent_decision_id == root.decision_id
|
|
print("create-child-ok")
|
|
|
|
|
|
def _json_roundtrip() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
snap = ContextSnapshot(
|
|
hot_context_hash="sha256:robot-test",
|
|
hot_context_ref="store://robot",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
|
|
ResourceRef(resource_id=str(ULID())),
|
|
],
|
|
actor_state_ref="checkpoint://robot",
|
|
)
|
|
d = _make_decision(
|
|
alternatives_considered=["Flask", "Django", "Express"],
|
|
confidence_score=0.92,
|
|
context_snapshot=snap,
|
|
artifacts_produced=[
|
|
ArtifactRef(artifact_path="src/api.py", artifact_type="file"),
|
|
ArtifactRef(artifact_path="patches/fix.patch", artifact_type="patch"),
|
|
],
|
|
downstream_decision_ids=[str(ULID())],
|
|
downstream_plan_ids=[str(ULID()), str(ULID())],
|
|
)
|
|
repo.create(d)
|
|
session.commit()
|
|
|
|
got = repo.get(d.decision_id)
|
|
assert got is not None
|
|
assert len(got.alternatives_considered) == 3
|
|
assert got.confidence_score == 0.92
|
|
assert got.context_snapshot.hot_context_hash == "sha256:robot-test"
|
|
assert len(got.context_snapshot.relevant_resources) == 2
|
|
assert len(got.artifacts_produced) == 2
|
|
assert len(got.downstream_decision_ids) == 1
|
|
assert len(got.downstream_plan_ids) == 2
|
|
print("json-roundtrip-ok")
|
|
|
|
|
|
def _get_by_plan() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
first_id: str | None = None
|
|
for i in range(5):
|
|
dt = DecisionType.PROMPT_DEFINITION if i == 0 else DecisionType.STRATEGY_CHOICE
|
|
parent = None if i == 0 else first_id
|
|
d = _make_decision(
|
|
parent_decision_id=parent,
|
|
sequence_number=i,
|
|
decision_type=dt,
|
|
)
|
|
repo.create(d)
|
|
if i == 0:
|
|
first_id = d.decision_id
|
|
session.commit()
|
|
|
|
results = repo.get_by_plan(_PLAN_ID)
|
|
assert len(results) == 5
|
|
for i in range(len(results) - 1):
|
|
assert results[i].sequence_number <= results[i + 1].sequence_number
|
|
print("get-by-plan-ok")
|
|
|
|
|
|
def _tree_bfs() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
root = _make_decision(sequence_number=0)
|
|
repo.create(root)
|
|
|
|
c1 = _make_decision(
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=1,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
repo.create(c1)
|
|
|
|
c2 = _make_decision(
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=2,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
repo.create(c2)
|
|
|
|
gc = _make_decision(
|
|
parent_decision_id=c1.decision_id,
|
|
sequence_number=3,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
)
|
|
repo.create(gc)
|
|
session.commit()
|
|
|
|
tree = repo.get_tree(root.decision_id)
|
|
assert len(tree) == 4, f"expected 4, got {len(tree)}"
|
|
assert tree[0].decision_id == root.decision_id
|
|
print("tree-bfs-ok")
|
|
|
|
|
|
def _path_to_root() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
root = _make_decision(sequence_number=0)
|
|
repo.create(root)
|
|
|
|
child = _make_decision(
|
|
parent_decision_id=root.decision_id,
|
|
sequence_number=1,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
)
|
|
repo.create(child)
|
|
|
|
leaf = _make_decision(
|
|
parent_decision_id=child.decision_id,
|
|
sequence_number=2,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
)
|
|
repo.create(leaf)
|
|
session.commit()
|
|
|
|
path = repo.get_path_to_root(leaf.decision_id)
|
|
assert len(path) == 3
|
|
assert path[0].decision_id == leaf.decision_id
|
|
assert path[-1].decision_id == root.decision_id
|
|
print("path-to-root-ok")
|
|
|
|
|
|
def _superseded() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
d = _make_decision()
|
|
repo.create(d)
|
|
session.commit()
|
|
|
|
new_id = str(ULID())
|
|
repo.update_superseded_by(d.decision_id, new_id)
|
|
session.commit()
|
|
|
|
got = repo.get(d.decision_id)
|
|
assert got is not None
|
|
assert got.is_superseded
|
|
assert got.superseded_by == new_id
|
|
print("superseded-ok")
|
|
|
|
|
|
def _delete() -> None:
|
|
session, factory = _setup()
|
|
_create_prerequisites(session, factory)
|
|
repo = DecisionRepository(session_factory=factory)
|
|
|
|
d = _make_decision()
|
|
repo.create(d)
|
|
session.commit()
|
|
|
|
result = repo.delete(d.decision_id)
|
|
session.commit()
|
|
assert result is True
|
|
|
|
got = repo.get(d.decision_id)
|
|
assert got is None
|
|
|
|
# Delete non-existent
|
|
result2 = repo.delete(str(ULID()))
|
|
assert result2 is False
|
|
|
|
print("delete-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_COMMANDS = {
|
|
"create-retrieve": _create_retrieve,
|
|
"create-child": _create_child,
|
|
"json-roundtrip": _json_roundtrip,
|
|
"get-by-plan": _get_by_plan,
|
|
"tree-bfs": _tree_bfs,
|
|
"path-to-root": _path_to_root,
|
|
"superseded": _superseded,
|
|
"delete": _delete,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 2:
|
|
raise SystemExit(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
|
|
command = sys.argv[1]
|
|
handler = _COMMANDS.get(command)
|
|
if handler is None:
|
|
raise SystemExit(f"Unknown command: {command}")
|
|
handler()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|