"""Helper script for Robot Framework decision persistence smoke tests. Usage: python robot/helper_decision_persistence.py 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()