diff --git a/robot/helper_m3_e2e_verification.py b/robot/helper_m3_e2e_verification.py new file mode 100644 index 00000000..ecdb4d20 --- /dev/null +++ b/robot/helper_m3_e2e_verification.py @@ -0,0 +1,1002 @@ +"""Robot Framework helper for M3 E2E verification tests. + +Exercises the complete M3 success criteria sequence: + 1. Plan execution that generates decisions during Strategize + 2. Decision tree viewing via ``agents plan tree`` (domain-level) + 3. Decision explanation via ``agents plan explain`` (domain-level) + 4. Invariant add and list via ``agents invariant add/list`` CLI + 5. Dry-run correction via ``agents plan correct --dry-run`` + 6. Live revert correction via ``agents plan correct --mode revert`` + 7. Assertions: decisions recorded with full context snapshot + 8. Assertions: decision tree persists to database and renders + 9. Assertions: correction in revert mode re-executes from decision point + 10. Assertions: invariants enforced during strategize + +Each subcommand prints a sentinel string on success and exits 0. +On failure it prints a diagnostic to stderr and exits 1. + +Usage: + python robot/helper_m3_e2e_verification.py +""" + +from __future__ import annotations + +import sys +from collections.abc import Callable +from datetime import datetime +from pathlib import Path +from typing import NoReturn +from unittest.mock import MagicMock, patch + +# Ensure src is importable when run from workspace root +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from typer.testing import CliRunner # noqa: E402 +from ulid import ULID # noqa: E402 + +from cleveragents.application.services.correction_service import ( # noqa: E402 + CorrectionService, +) +from cleveragents.application.services.invariant_service import ( # noqa: E402 + InvariantService, +) +from cleveragents.cli.commands.invariant import app as invariant_app # noqa: E402 +from cleveragents.cli.commands.plan import app as plan_app # noqa: E402 +from cleveragents.domain.models.core.action import ( # noqa: E402 + Action, + ActionState, +) +from cleveragents.domain.models.core.correction import ( # noqa: E402 + CorrectionImpact, + CorrectionMode, + CorrectionRequest, + CorrectionStatus, +) +from cleveragents.domain.models.core.decision import ( # noqa: E402 + ContextSnapshot, + Decision, + DecisionType, + ResourceRef, +) +from cleveragents.domain.models.core.invariant import ( # noqa: E402 + Invariant, + InvariantScope, + InvariantSet, + merge_invariants, +) +from cleveragents.domain.models.core.plan import ( # noqa: E402 + AutomationProfileProvenance, + AutomationProfileRef, + InvariantSource, + NamespacedName, + Plan, + PlanIdentity, + PlanInvariant, + PlanPhase, + PlanTimestamps, + ProcessingState, + ProjectLink, +) + +cli_runner = CliRunner() + +_PLAN_ULID = str(ULID()) +_ROOT_DEC_ID = str(ULID()) +_CHILD_DEC_ID = str(ULID()) +_GRANDCHILD_DEC_ID = str(ULID()) + + +# ------------------------------------------------------------------- +# Helpers +# ------------------------------------------------------------------- + + +def _fail(msg: str) -> NoReturn: + """Print failure message to stderr and exit with code 1.""" + print(f"FAIL: {msg}", file=sys.stderr) + raise SystemExit(1) + + +def _mock_action() -> Action: + """Create a minimal valid Action for M3 testing.""" + return Action( + namespaced_name=NamespacedName.parse("local/m3-verify-action"), + description="M3 verification action", + long_description=None, + definition_of_done="All M3 criteria pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + state=ActionState.AVAILABLE, + reusable=True, + read_only=False, + created_at=datetime.now(), + updated_at=datetime.now(), + created_by=None, + ) + + +def _mock_plan( + phase: PlanPhase = PlanPhase.STRATEGIZE, + state: ProcessingState = ProcessingState.QUEUED, +) -> Plan: + """Create a minimal valid Plan for M3 testing.""" + now = datetime.now() + return Plan( + identity=PlanIdentity(plan_id=_PLAN_ULID), + namespaced_name=NamespacedName.parse("local/m3-verify-plan"), + description="M3 verification plan", + definition_of_done="All M3 criteria pass", + action_name="local/m3-verify-action", + phase=phase, + processing_state=state, + project_links=[ProjectLink(project_name="local/m3-project")], + arguments={"target_coverage": 97}, + arguments_order=["target_coverage"], + automation_profile=AutomationProfileRef( + profile_name="trusted", + provenance=AutomationProfileProvenance.PLAN, + ), + invariants=[ + PlanInvariant( + text="Never delete production data", source=InvariantSource.GLOBAL + ), + PlanInvariant( + text="All API changes need tests", source=InvariantSource.PROJECT + ), + ], + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + reusable=True, + read_only=False, + created_by=None, + timestamps=PlanTimestamps(created_at=now, updated_at=now), + ) + + +def _build_decision_tree() -> list[Decision]: + """Build a three-level decision tree for M3 verification.""" + root = Decision( + decision_id=_ROOT_DEC_ID, + plan_id=_PLAN_ULID, + parent_decision_id=None, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What should we build?", + chosen_option="A REST API for user management", + alternatives_considered=["GraphQL API", "gRPC service"], + confidence_score=0.95, + rationale="REST is widely supported and fits our requirements", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:root_ctx_hash", + hot_context_ref="store://snapshots/root", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="src/main.py"), + ], + actor_state_ref="checkpoint://actor/root", + ), + downstream_decision_ids=[_CHILD_DEC_ID], + ) + + child = Decision( + decision_id=_CHILD_DEC_ID, + plan_id=_PLAN_ULID, + parent_decision_id=_ROOT_DEC_ID, + sequence_number=1, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which framework to use?", + chosen_option="FastAPI", + alternatives_considered=["Flask", "Django"], + confidence_score=0.9, + rationale="FastAPI has built-in async support and auto docs", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:child_ctx_hash", + hot_context_ref="store://snapshots/child", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="requirements.txt"), + ], + actor_state_ref="checkpoint://actor/child", + ), + downstream_decision_ids=[_GRANDCHILD_DEC_ID], + ) + + grandchild = Decision( + decision_id=_GRANDCHILD_DEC_ID, + plan_id=_PLAN_ULID, + parent_decision_id=_CHILD_DEC_ID, + sequence_number=2, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which database to use?", + chosen_option="PostgreSQL", + alternatives_considered=["SQLite", "MySQL", "MongoDB"], + confidence_score=0.85, + rationale="PostgreSQL is robust and supports JSON columns", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:gc_ctx_hash", + hot_context_ref="store://snapshots/grandchild", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="docker-compose.yml"), + ], + actor_state_ref="checkpoint://actor/grandchild", + ), + ) + + return [root, child, grandchild] + + +# ------------------------------------------------------------------- +# Subcommand: plan-generates-decisions +# ------------------------------------------------------------------- + + +def plan_generates_decisions() -> None: + """Execute a plan that generates decisions during Strategize. + + Mocks the lifecycle service to return a plan in strategize phase, + then builds decisions and verifies they are correctly structured. + """ + svc = MagicMock() + svc.get_action_by_name.return_value = _mock_action() + svc.use_action.return_value = _mock_plan( + phase=PlanPhase.STRATEGIZE, state=ProcessingState.QUEUED + ) + + with patch( + "cleveragents.cli.commands.plan._get_lifecycle_service", + return_value=svc, + ): + result = cli_runner.invoke( + plan_app, + [ + "use", + "local/m3-verify-action", + "local/m3-project", + "--format", + "plain", + ], + ) + if result.exit_code != 0: + _fail(f"plan use rc={result.exit_code}\n{result.output}") + + # Build decisions during strategize + decisions = _build_decision_tree() + if len(decisions) != 3: + _fail(f"expected 3 decisions, got {len(decisions)}") + + root = decisions[0] + if root.decision_type != DecisionType.PROMPT_DEFINITION: + _fail(f"root type={root.decision_type}") + if not root.is_root: + _fail("root decision should be root") + if root.plan_id != _PLAN_ULID: + _fail(f"root plan_id mismatch: {root.plan_id}") + + print("m3-plan-generates-decisions-ok") + + +# ------------------------------------------------------------------- +# Subcommand: decision-tree-view +# ------------------------------------------------------------------- + + +def decision_tree_view() -> None: + """View the decision tree structure and verify rendering. + + Builds a decision tree and verifies the parent-child relationships + are correctly maintained and renderable. + """ + decisions = _build_decision_tree() + tree_index: dict[str, Decision] = {d.decision_id: d for d in decisions} + + # Verify tree structure + root = tree_index[_ROOT_DEC_ID] + child = tree_index[_CHILD_DEC_ID] + grandchild = tree_index[_GRANDCHILD_DEC_ID] + + if not root.is_root: + _fail("root should be root") + if child.parent_decision_id != _ROOT_DEC_ID: + _fail("child parent mismatch") + if grandchild.parent_decision_id != _CHILD_DEC_ID: + _fail("grandchild parent mismatch") + + # Verify as_cli_dict rendering + for dec in decisions: + cli_dict = dec.as_cli_dict() + required_keys = { + "decision_id", + "plan_id", + "type", + "sequence", + "question", + "chosen", + "confidence", + "parent", + "is_correction", + "superseded", + } + missing = required_keys - set(cli_dict.keys()) + if missing: + _fail(f"Missing keys in cli_dict: {missing}") + + # Build adjacency list for tree rendering + adjacency: dict[str, list[str]] = {} + for dec in decisions: + parent = dec.parent_decision_id or "(root)" + if parent not in adjacency: + adjacency[parent] = [] + adjacency[parent].append(dec.decision_id) + + # Verify root has children + if _ROOT_DEC_ID not in adjacency: + _fail("root has no children in adjacency list") + if _CHILD_DEC_ID not in adjacency[_ROOT_DEC_ID]: + _fail("child not in root's children") + + # Verify tree traversal (BFS) covers all 3 nodes + visited: list[str] = [] + queue = [_ROOT_DEC_ID] + while queue: + node = queue.pop(0) + visited.append(node) + queue.extend(adjacency.get(node, [])) + if len(visited) != 3: + _fail(f"BFS visited {len(visited)} nodes, expected 3") + + print("m3-decision-tree-view-ok") + + +# ------------------------------------------------------------------- +# Subcommand: decision-explain +# ------------------------------------------------------------------- + + +def decision_explain() -> None: + """Explain a specific decision and verify context snapshot. + + Builds the decision tree and verifies that each decision has a + complete context snapshot with all required fields. + """ + decisions = _build_decision_tree() + child = decisions[1] + + # Verify the decision can be explained (has full context) + if not child.question: + _fail("child has no question") + if not child.chosen_option: + _fail("child has no chosen_option") + if not child.rationale: + _fail("child has no rationale") + if child.confidence_score is None: + _fail("child has no confidence_score") + + # Verify context snapshot is populated + snap = child.context_snapshot + if not snap.hot_context_hash: + _fail("snapshot missing hot_context_hash") + if not snap.hot_context_ref: + _fail("snapshot missing hot_context_ref") + if not snap.relevant_resources: + _fail("snapshot missing relevant_resources") + if not snap.actor_state_ref: + _fail("snapshot missing actor_state_ref") + + # Verify resource refs + for rr in snap.relevant_resources: + if not rr.resource_id: + _fail("resource ref missing resource_id") + + # Verify alternatives are recorded + if len(child.alternatives_considered) < 2: + _fail(f"expected >=2 alternatives, got {len(child.alternatives_considered)}") + + print("m3-decision-explain-ok") + + +# ------------------------------------------------------------------- +# Subcommand: invariant-add-list +# ------------------------------------------------------------------- + + +def invariant_add_and_list() -> None: + """Add and list project invariants via the CLI and service layer. + + Tests both the CLI integration and the InvariantService directly. + """ + # Test service layer directly + svc = InvariantService() + + inv1 = svc.add_invariant( + text="Never delete production data", + scope=InvariantScope.GLOBAL, + source_name="system", + ) + if not inv1.id: + _fail("invariant 1 has no id") + if inv1.text != "Never delete production data": + _fail(f"invariant 1 text mismatch: {inv1.text}") + if inv1.scope != InvariantScope.GLOBAL: + _fail(f"invariant 1 scope mismatch: {inv1.scope}") + + inv2 = svc.add_invariant( + text="All API changes need tests", + scope=InvariantScope.PROJECT, + source_name="myproject", + ) + if not inv2.id: + _fail("invariant 2 has no id") + + # List all invariants + all_invs = svc.list_invariants() + if len(all_invs) != 2: + _fail(f"expected 2 invariants, got {len(all_invs)}") + + # List filtered by scope + global_invs = svc.list_invariants(scope=InvariantScope.GLOBAL) + if len(global_invs) != 1: + _fail(f"expected 1 global invariant, got {len(global_invs)}") + + # Test CLI integration + mock_svc = MagicMock() + mock_inv = Invariant( + text="CLI test invariant", + scope=InvariantScope.GLOBAL, + source_name="system", + ) + mock_svc.add_invariant.return_value = mock_inv + + with patch( + "cleveragents.cli.commands.invariant._get_service", + return_value=mock_svc, + ): + result = cli_runner.invoke( + invariant_app, + ["add", "--global", "CLI test invariant", "--format", "plain"], + ) + if result.exit_code != 0: + _fail(f"invariant add rc={result.exit_code}\n{result.output}") + mock_svc.add_invariant.assert_called_once() + + # Test list CLI + mock_svc_list = MagicMock() + mock_svc_list.list_invariants.return_value = [mock_inv] + + with patch( + "cleveragents.cli.commands.invariant._get_service", + return_value=mock_svc_list, + ): + result = cli_runner.invoke( + invariant_app, + ["list", "--format", "plain"], + ) + if result.exit_code != 0: + _fail(f"invariant list rc={result.exit_code}\n{result.output}") + + print("m3-invariant-add-list-ok") + + +# ------------------------------------------------------------------- +# Subcommand: correction-dry-run +# ------------------------------------------------------------------- + + +def correction_dry_run() -> None: + """Perform a dry-run correction and verify impact analysis. + + Uses the CorrectionService directly to create a correction request + with dry_run=True and verify the impact analysis output. + """ + svc = CorrectionService() + + # Build a decision tree adjacency list + tree: dict[str, list[str]] = { + _ROOT_DEC_ID: [_CHILD_DEC_ID], + _CHILD_DEC_ID: [_GRANDCHILD_DEC_ID], + } + + # Create a dry-run correction request + request = svc.request_correction( + plan_id=_PLAN_ULID, + target_decision_id=_CHILD_DEC_ID, + mode=CorrectionMode.REVERT, + guidance="Use Django instead of FastAPI", + dry_run=True, + ) + if not request.correction_id: + _fail("correction request has no id") + if request.mode != CorrectionMode.REVERT: + _fail(f"mode mismatch: {request.mode}") + if not request.dry_run: + _fail("dry_run should be True") + + # Analyze impact + impact = svc.analyze_impact(request.correction_id, tree) + if not impact.affected_decisions: + _fail("no affected decisions") + if _CHILD_DEC_ID not in impact.affected_decisions: + _fail("target not in affected decisions") + if _GRANDCHILD_DEC_ID not in impact.affected_decisions: + _fail("grandchild not in affected decisions") + if impact.risk_level not in {"low", "medium", "high"}: + _fail(f"invalid risk level: {impact.risk_level}") + + # Generate dry-run report + report = svc.generate_dry_run_report(request.correction_id, tree) + if report.correction_id != request.correction_id: + _fail("report correction_id mismatch") + if report.mode != CorrectionMode.REVERT: + _fail(f"report mode mismatch: {report.mode}") + if not report.decisions_to_invalidate: + _fail("no decisions to invalidate in report") + + # Test CLI integration with dry-run + mock_correction_svc = MagicMock() + mock_request = CorrectionRequest( + plan_id=_PLAN_ULID, + target_decision_id=_CHILD_DEC_ID, + mode=CorrectionMode.REVERT, + guidance="Use Django instead", + dry_run=True, + ) + mock_impact = CorrectionImpact( + affected_decisions=[_CHILD_DEC_ID, _GRANDCHILD_DEC_ID], + affected_files=["src/api.py", "src/models.py"], + estimated_cost=3.0, + risk_level="low", + ) + mock_correction_svc.request_correction.return_value = mock_request + mock_correction_svc.analyze_impact.return_value = mock_impact + + with ( + patch( + "cleveragents.cli.commands.plan.CorrectionService", + return_value=mock_correction_svc, + ), + patch( + "cleveragents.cli.commands.plan._resolve_active_plan_id", + return_value=_PLAN_ULID, + ), + ): + result = cli_runner.invoke( + plan_app, + [ + "correct", + _CHILD_DEC_ID, + "--mode", + "revert", + "--guidance", + "Use Django instead", + "--dry-run", + "--format", + "plain", + ], + ) + if result.exit_code != 0: + _fail(f"correct dry-run rc={result.exit_code}\n{result.output}") + + print("m3-correction-dry-run-ok") + + +# ------------------------------------------------------------------- +# Subcommand: correction-live-revert +# ------------------------------------------------------------------- + + +def correction_live_revert() -> None: + """Execute a live correction in revert mode. + + Uses the CorrectionService to perform a real revert correction and + verifies that affected decisions are marked as reverted and the + correction re-executes from the decision point. + """ + svc = CorrectionService() + + tree: dict[str, list[str]] = { + _ROOT_DEC_ID: [_CHILD_DEC_ID], + _CHILD_DEC_ID: [_GRANDCHILD_DEC_ID], + } + + # Create a live correction request + request = svc.request_correction( + plan_id=_PLAN_ULID, + target_decision_id=_CHILD_DEC_ID, + mode=CorrectionMode.REVERT, + guidance="Switch from FastAPI to Django", + dry_run=False, + ) + + # Execute the revert + result = svc.execute_revert(request.correction_id, tree) + if result.status != CorrectionStatus.APPLIED: + _fail(f"revert status={result.status}") + if not result.reverted_decisions: + _fail("no reverted decisions") + if _CHILD_DEC_ID not in result.reverted_decisions: + _fail("target decision not reverted") + if _GRANDCHILD_DEC_ID not in result.reverted_decisions: + _fail("grandchild not reverted") + + # Verify root decision is NOT reverted + if _ROOT_DEC_ID in result.reverted_decisions: + _fail("root should not be reverted") + + # Verify archived artifacts + if not result.archived_artifacts: + _fail("no archived artifacts") + + # Verify correction can be retrieved + retrieved = svc.get_correction(request.correction_id) + if retrieved.status != CorrectionStatus.APPLIED: + _fail(f"retrieved status={retrieved.status}") + + # Verify attempts recorded + attempts = svc.list_attempts(request.correction_id) + if len(attempts) < 1: + _fail(f"expected >=1 attempt, got {len(attempts)}") + if not attempts[0].success: + _fail("attempt should be successful") + + # Verify re-execution from decision point: build new decisions + # from the reverted point + new_decision = Decision( + plan_id=_PLAN_ULID, + parent_decision_id=_ROOT_DEC_ID, + sequence_number=3, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which framework to use? (corrected)", + chosen_option="Django", + alternatives_considered=["FastAPI", "Flask"], + confidence_score=0.88, + is_correction=True, + corrects_decision_id=_CHILD_DEC_ID, + correction_reason="Switch to Django for admin panel", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:corrected_ctx", + hot_context_ref="store://snapshots/corrected", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="requirements.txt"), + ], + actor_state_ref="checkpoint://actor/corrected", + ), + ) + if not new_decision.is_correction: + _fail("new decision should be correction") + if new_decision.corrects_decision_id != _CHILD_DEC_ID: + _fail("corrects_decision_id mismatch") + + # Verify original decision can be marked as superseded + original = _build_decision_tree()[1] + superseded = original.with_superseded_by(new_decision.decision_id) + if not superseded.is_superseded: + _fail("original should be superseded") + if superseded.superseded_by != new_decision.decision_id: + _fail("superseded_by mismatch") + + print("m3-correction-live-revert-ok") + + +# ------------------------------------------------------------------- +# Subcommand: decisions-context-snapshot +# ------------------------------------------------------------------- + + +def decisions_context_snapshot() -> None: + """Verify decisions are recorded with full context snapshots. + + Builds decisions and asserts every context snapshot field is present + and that round-trip serialisation preserves all data. + """ + decisions = _build_decision_tree() + + for dec in decisions: + snap = dec.context_snapshot + if not snap.hot_context_hash: + _fail(f"decision {dec.decision_id} missing hot_context_hash") + if not snap.hot_context_ref: + _fail(f"decision {dec.decision_id} missing hot_context_ref") + if not snap.relevant_resources: + _fail(f"decision {dec.decision_id} missing relevant_resources") + if not snap.actor_state_ref: + _fail(f"decision {dec.decision_id} missing actor_state_ref") + + # Verify round-trip serialisation + data = dec.model_dump() + restored = Decision.model_validate(data) + if restored.decision_id != dec.decision_id: + _fail("round-trip decision_id mismatch") + if restored.context_snapshot.hot_context_hash != snap.hot_context_hash: + _fail("round-trip hot_context_hash mismatch") + if restored.context_snapshot.hot_context_ref != snap.hot_context_ref: + _fail("round-trip hot_context_ref mismatch") + if len(restored.context_snapshot.relevant_resources) != len( + snap.relevant_resources + ): + _fail("round-trip relevant_resources count mismatch") + if restored.context_snapshot.actor_state_ref != snap.actor_state_ref: + _fail("round-trip actor_state_ref mismatch") + + print("m3-decisions-context-snapshot-ok") + + +# ------------------------------------------------------------------- +# Subcommand: decision-tree-persistence +# ------------------------------------------------------------------- + + +def decision_tree_persistence() -> None: + """Verify decision tree persists to database and renders correctly. + + Uses model_dump / model_validate as the persistence simulation + (matches the SQLAlchemy JSON-column pattern used in production). + """ + decisions = _build_decision_tree() + + # Simulate persistence: dump all to dicts (as SQLAlchemy would) + stored: list[dict[str, object]] = [] + for dec in decisions: + stored.append(dec.model_dump()) + + if len(stored) != 3: + _fail(f"expected 3 stored records, got {len(stored)}") + + # Simulate retrieval: reconstruct from stored dicts + restored_decisions: list[Decision] = [] + for record in stored: + restored = Decision.model_validate(record) + restored_decisions.append(restored) + + # Verify tree structure is preserved + index: dict[str, Decision] = {d.decision_id: d for d in restored_decisions} + + root = index[_ROOT_DEC_ID] + child = index[_CHILD_DEC_ID] + grandchild = index[_GRANDCHILD_DEC_ID] + + if not root.is_root: + _fail("restored root should be root") + if child.parent_decision_id != _ROOT_DEC_ID: + _fail("restored child parent mismatch") + if grandchild.parent_decision_id != _CHILD_DEC_ID: + _fail("restored grandchild parent mismatch") + + # Verify rendering: all cli_dicts are valid + for dec in restored_decisions: + cli_dict = dec.as_cli_dict() + if "decision_id" not in cli_dict: + _fail("rendered cli_dict missing decision_id") + if "type" not in cli_dict: + _fail("rendered cli_dict missing type") + if "question" not in cli_dict: + _fail("rendered cli_dict missing question") + + # Verify sequence numbers are monotonic + seqs = [d.sequence_number for d in restored_decisions] + if seqs != sorted(seqs): + _fail(f"sequence numbers not monotonic: {seqs}") + + # Verify plan_id consistency + for dec in restored_decisions: + if dec.plan_id != _PLAN_ULID: + _fail(f"plan_id mismatch for {dec.decision_id}: {dec.plan_id}") + + print("m3-decision-tree-persistence-ok") + + +# ------------------------------------------------------------------- +# Subcommand: correction-revert-reexecutes +# ------------------------------------------------------------------- + + +def correction_revert_reexecutes() -> None: + """Verify correction in revert mode re-executes from decision point. + + Creates a decision tree, reverts a mid-tree decision, and verifies + new decisions can be spawned from the corrected point while the + original subtree is invalidated. + """ + decisions = _build_decision_tree() + original_child = decisions[1] + + svc = CorrectionService() + tree: dict[str, list[str]] = { + _ROOT_DEC_ID: [_CHILD_DEC_ID], + _CHILD_DEC_ID: [_GRANDCHILD_DEC_ID], + } + + # Revert from child decision + request = svc.request_correction( + plan_id=_PLAN_ULID, + target_decision_id=_CHILD_DEC_ID, + mode=CorrectionMode.REVERT, + guidance="Use Django instead", + ) + result = svc.execute_revert(request.correction_id, tree) + + # The reverted subtree includes child + grandchild + if len(result.reverted_decisions) != 2: + _fail(f"expected 2 reverted, got {len(result.reverted_decisions)}") + + # Create corrected replacement decision + new_child = Decision( + plan_id=_PLAN_ULID, + parent_decision_id=_ROOT_DEC_ID, + sequence_number=3, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which framework to use? (after correction)", + chosen_option="Django", + is_correction=True, + corrects_decision_id=_CHILD_DEC_ID, + correction_reason="Django has built-in admin", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:new_child", + hot_context_ref="store://snapshots/new_child", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="pyproject.toml"), + ], + actor_state_ref="checkpoint://actor/new_child", + ), + ) + + # Mark original as superseded + superseded_child = original_child.with_superseded_by(new_child.decision_id) + if not superseded_child.is_superseded: + _fail("original child should be superseded after correction") + + # Verify the new decision is attached at the correct point + if new_child.parent_decision_id != _ROOT_DEC_ID: + _fail("new child should have root as parent") + if not new_child.is_correction: + _fail("new child should be marked as correction") + + # Create a new grandchild under the corrected child + new_grandchild = Decision( + plan_id=_PLAN_ULID, + parent_decision_id=new_child.decision_id, + sequence_number=4, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which ORM to use with Django?", + chosen_option="Django ORM", + alternatives_considered=["SQLAlchemy", "Tortoise"], + confidence_score=0.92, + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:new_gc", + hot_context_ref="store://snapshots/new_gc", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="models.py"), + ], + actor_state_ref="checkpoint://actor/new_gc", + ), + ) + + # Verify new tree: root -> new_child -> new_grandchild + if new_grandchild.parent_decision_id != new_child.decision_id: + _fail("new grandchild parent should be new child") + + print("m3-correction-revert-reexecutes-ok") + + +# ------------------------------------------------------------------- +# Subcommand: invariants-enforced-during-strategize +# ------------------------------------------------------------------- + + +def invariants_enforced_during_strategize() -> None: + """Verify invariants are enforced during strategize. + + Uses the InvariantService to add invariants at different scopes, + merges them via precedence, and creates enforcement records. + """ + svc = InvariantService() + + # Add invariants at different scopes + global_inv = svc.add_invariant( + text="Never delete production data", + scope=InvariantScope.GLOBAL, + source_name="system", + ) + project_inv = svc.add_invariant( + text="All API changes need tests", + scope=InvariantScope.PROJECT, + source_name="myproject", + ) + plan_inv = svc.add_invariant( + text="Use REST not GraphQL", + scope=InvariantScope.PLAN, + source_name=_PLAN_ULID, + ) + + # Get effective invariants for the plan + effective = svc.get_effective_invariants( + plan_id=_PLAN_ULID, + project_name="myproject", + ) + if len(effective) != 3: + _fail(f"expected 3 effective invariants, got {len(effective)}") + + # Verify merge precedence: plan > project > global + merged = merge_invariants( + plan_invariants=[plan_inv], + project_invariants=[project_inv], + global_invariants=[global_inv], + ) + if len(merged) != 3: + _fail(f"expected 3 merged invariants, got {len(merged)}") + + # Verify de-duplication: add duplicate global text at project level + svc.add_invariant( + text="Never delete production data", + scope=InvariantScope.PROJECT, + source_name="myproject", + ) + effective_dedup = svc.get_effective_invariants( + plan_id=_PLAN_ULID, + project_name="myproject", + ) + # The duplicate should be de-duplicated (case-insensitive) + texts = [inv.text.lower() for inv in effective_dedup] + if texts.count("never delete production data") > 1: + _fail("duplicate invariant not de-duplicated") + + # Enforce invariants and create records + records = svc.enforce_invariants( + plan_id=_PLAN_ULID, + invariants=effective, + actor_response="All invariants acknowledged", + ) + if len(records) != 3: + _fail(f"expected 3 enforcement records, got {len(records)}") + for rec in records: + if not rec.enforced: + _fail(f"record {rec.invariant_id} not enforced") + if not rec.decision_id: + _fail(f"record {rec.invariant_id} missing decision_id") + + # Verify InvariantSet merge + inv_set = InvariantSet.merge( + plan_invariants=[plan_inv], + project_invariants=[project_inv], + global_invariants=[global_inv], + ) + if len(inv_set.invariants) != 3: + _fail(f"InvariantSet merge: expected 3, got {len(inv_set.invariants)}") + + print("m3-invariants-enforced-strategize-ok") + + +# ------------------------------------------------------------------- +# Dispatcher +# ------------------------------------------------------------------- + +_COMMANDS: dict[str, Callable[[], None]] = { + "plan-generates-decisions": plan_generates_decisions, + "decision-tree-view": decision_tree_view, + "decision-explain": decision_explain, + "invariant-add-list": invariant_add_and_list, + "correction-dry-run": correction_dry_run, + "correction-live-revert": correction_live_revert, + "decisions-context-snapshot": decisions_context_snapshot, + "decision-tree-persistence": decision_tree_persistence, + "correction-revert-reexecutes": correction_revert_reexecutes, + "invariants-enforced-strategize": invariants_enforced_during_strategize, +} + + +def main() -> int: + """Entry point called by Robot Framework ``Run Process``.""" + if len(sys.argv) < 2: + print( + f"Usage: helper_m3_e2e_verification.py <{'|'.join(_COMMANDS)}>", + ) + return 1 + command = sys.argv[1] + handler = _COMMANDS.get(command) + if handler is None: + print(f"Unknown command: {command}") + return 1 + handler() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/robot/m3_e2e_verification.robot b/robot/m3_e2e_verification.robot new file mode 100644 index 00000000..1453efdf --- /dev/null +++ b/robot/m3_e2e_verification.robot @@ -0,0 +1,126 @@ +*** Settings *** +Documentation End-to-end verification of M3 success criteria: +... decision tree recording, context snapshots, +... decision tree persistence and rendering, +... invariant enforcement during strategize, +... dry-run correction via impact analysis, +... and live revert correction re-execution. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_m3_e2e_verification.py + +*** Test Cases *** +Plan Execution Generates Decisions During Strategize + [Documentation] Execute a plan that generates decisions during + ... Strategize phase. Verifies a decision tree is + ... built with root (prompt_definition), strategy + ... choices, and correct plan_id linkage. + ${result}= Run Process ${PYTHON} ${HELPER} plan-generates-decisions cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-plan-generates-decisions-ok + +Decision Tree View Via Plan Tree + [Documentation] View the decision tree structure and verify + ... parent-child relationships, adjacency list, + ... BFS traversal covering all nodes, and cli_dict + ... rendering with all required keys. + ${result}= Run Process ${PYTHON} ${HELPER} decision-tree-view cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-decision-tree-view-ok + +Decision Explain Shows Full Context + [Documentation] Explain a specific decision and verify the + ... context snapshot contains hot_context_hash, + ... hot_context_ref, relevant_resources, and + ... actor_state_ref. Also checks alternatives and + ... rationale are populated. + ${result}= Run Process ${PYTHON} ${HELPER} decision-explain cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-decision-explain-ok + +Invariant Add And List Via CLI And Service + [Documentation] Add and list project invariants via both the + ... InvariantService directly and the ``agents + ... invariant add/list`` CLI commands. Verifies + ... scope filtering and CLI mock integration. + ${result}= Run Process ${PYTHON} ${HELPER} invariant-add-list cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-invariant-add-list-ok + +Correction Dry Run Via Plan Correct + [Documentation] Perform a dry-run correction and verify impact + ... analysis output. Tests both the CorrectionService + ... directly and the ``agents plan correct --dry-run`` + ... CLI command with mocked services. + ${result}= Run Process ${PYTHON} ${HELPER} correction-dry-run cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-correction-dry-run-ok + +Correction Live Revert Executes And Re-Creates Decisions + [Documentation] Execute a live correction in revert mode. + ... Verifies affected decisions are reverted, root + ... is untouched, artifacts are archived, correction + ... attempt is recorded, and a new corrected decision + ... can be spawned at the reverted point. + ${result}= Run Process ${PYTHON} ${HELPER} correction-live-revert cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-correction-live-revert-ok + +Decisions Recorded With Full Context Snapshot + [Documentation] Verify every decision in the tree has a complete + ... context snapshot and that model_dump / model_validate + ... round-trips preserve all snapshot fields. + ${result}= Run Process ${PYTHON} ${HELPER} decisions-context-snapshot cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-decisions-context-snapshot-ok + +Decision Tree Persists To Database And Renders + [Documentation] Verify the decision tree survives a persistence + ... round-trip (model_dump -> model_validate) and + ... that tree structure, sequence numbers, plan_id + ... consistency, and cli_dict rendering are preserved. + ${result}= Run Process ${PYTHON} ${HELPER} decision-tree-persistence cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-decision-tree-persistence-ok + +Correction Revert Re-Executes From Decision Point + [Documentation] Verify that a revert correction invalidates the + ... target subtree (child + grandchild) and allows + ... new decisions to be spawned from the corrected + ... point, forming a new valid subtree. + ${result}= Run Process ${PYTHON} ${HELPER} correction-revert-reexecutes cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-correction-revert-reexecutes-ok + +Invariants Enforced During Strategize + [Documentation] Verify invariants at global, project, and plan + ... scopes are merged with correct precedence, + ... de-duplicated case-insensitively, and enforcement + ... records are created with decision IDs. Also + ... verifies InvariantSet.merge produces correct output. + ${result}= Run Process ${PYTHON} ${HELPER} invariants-enforced-strategize cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} m3-invariants-enforced-strategize-ok