"""Helper script for Robot Framework decision recording smoke tests. Usage: python robot/helper_decision_recording.py Subcommands: record-retrieve Record + retrieve a decision via DecisionService sequencing Verify monotonic sequencing snapshot-capture Verify auto-captured context snapshot tree-bfs Build a tree and verify BFS via get_tree superseded Mark a decision as superseded via service delete Delete a decision and verify snapshot removal snapshot-dedup Verify hash-based deduplication in SnapshotStore """ from __future__ import annotations import sys from pathlib import Path # Ensure src is importable when run from workspace root sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.application.services.decision_service import ( DecisionService, SnapshotStore, ) from cleveragents.domain.models.core.decision import ( ContextSnapshot, DecisionType, ) _PLAN_ID = "01HV000000000000000000RS01" # --------------------------------------------------------------------------- # Subcommands # --------------------------------------------------------------------------- def _record_retrieve(): svc = DecisionService() d = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.PROMPT_DEFINITION, question="What approach?", chosen_option="Build API", ) got = svc.get_decision(d.decision_id) assert got.decision_id == d.decision_id assert got.is_root assert str(got.decision_type) == "prompt_definition" print("record-retrieve-ok") def _sequencing(): svc = DecisionService() d0 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.PROMPT_DEFINITION, question="First", chosen_option="Chosen first", ) d1 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.STRATEGY_CHOICE, question="Second", chosen_option="Chosen second", parent_decision_id=d0.decision_id, ) d2 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.IMPLEMENTATION_CHOICE, question="Third", chosen_option="Chosen third", parent_decision_id=d0.decision_id, ) assert d0.sequence_number == 0 assert d1.sequence_number == 1 assert d2.sequence_number == 2 assert svc.count_decisions(_PLAN_ID) == 3 print("sequencing-ok") def _snapshot_capture(): svc = DecisionService() d = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.STRATEGY_CHOICE, question="Snapshot test", chosen_option="Chosen", ) snap = svc.get_snapshot(d.decision_id) assert snap is not None assert snap.hot_context_hash.startswith("sha256:") print("snapshot-capture-ok") def _tree_bfs(): svc = DecisionService() root = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.PROMPT_DEFINITION, question="Root", chosen_option="Root choice", ) child1 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.STRATEGY_CHOICE, question="Child 1", chosen_option="C1", parent_decision_id=root.decision_id, ) svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.STRATEGY_CHOICE, question="Child 2", chosen_option="C2", parent_decision_id=root.decision_id, ) svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.IMPLEMENTATION_CHOICE, question="Grandchild", chosen_option="GC", parent_decision_id=child1.decision_id, ) tree = svc.get_tree(_PLAN_ID) assert len(tree) == 4, f"expected 4, got {len(tree)}" assert tree[0].is_root # Verify BFS level order: compute depths and check non-decreasing. by_id = {d.decision_id: d for d in tree} def _depth(node): depth = 0 current = node while current.parent_decision_id is not None: parent = by_id.get(current.parent_decision_id) if parent is None: break depth += 1 current = parent return depth depths = [_depth(d) for d in tree] for i in range(1, len(depths)): assert depths[i] >= depths[i - 1], ( f"BFS order violated at index {i}: " f"depth {depths[i]} follows depth {depths[i - 1]}. " f"Depths: {depths}" ) print("tree-bfs-ok") def _superseded(): svc = DecisionService() d1 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.PROMPT_DEFINITION, question="Original", chosen_option="Orig", ) d2 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.STRATEGY_CHOICE, question="Replacement", chosen_option="New", parent_decision_id=d1.decision_id, ) updated = svc.mark_superseded(d1.decision_id, d2.decision_id) assert updated.is_superseded assert updated.superseded_by == d2.decision_id superseded = svc.get_superseded(_PLAN_ID) assert len(superseded) == 1 print("superseded-ok") def _delete(): svc = DecisionService() d = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.PROMPT_DEFINITION, question="Ephemeral", chosen_option="Gone", ) did = d.decision_id assert svc.get_snapshot(did) is not None result = svc.delete_decision(did) assert result is True assert svc.get_snapshot(did) is None assert svc.count_decisions(_PLAN_ID) == 0 print("delete-ok") def _snapshot_dedup(): store = SnapshotStore() snap = ContextSnapshot( hot_context_hash="sha256:samehash", hot_context_ref="ref1", ) store.store("DEC_A", snap) store.store("DEC_B", snap) ids = store.get_by_hash("sha256:samehash") assert len(ids) == 2, f"expected 2, got {len(ids)}" assert "DEC_A" in ids assert "DEC_B" in ids # Remove one store.remove("DEC_A") ids2 = store.get_by_hash("sha256:samehash") assert len(ids2) == 1 assert "DEC_B" in ids2 print("snapshot-dedup-ok") # --------------------------------------------------------------------------- # Dispatcher # --------------------------------------------------------------------------- _COMMANDS = { "record-retrieve": _record_retrieve, "sequencing": _sequencing, "snapshot-capture": _snapshot_capture, "tree-bfs": _tree_bfs, "superseded": _superseded, "delete": _delete, "snapshot-dedup": _snapshot_dedup, } def main(): 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()