"""Robot helper for M1 E2E verification suite. Exercises the complete M1 success-criteria sequence: 1. Action creation from YAML config 2. Git-checkout resource registration 3. Project creation and resource linking 4. Plan use / execute / diff / apply lifecycle 5. SQLite persistence assertions (Plan + Action records) 6. ChangeSet built from tool invocations (not parsed output) 7. Git worktree sandbox creates isolated working directory 8. Sandbox changes do not affect original until Apply 9. Post-apply commit exists in the target repo Each subcommand prints a sentinel on success. Exit code 0 = pass, 1 = failure. CLI-facing tests (1-4) invoke the real ``agents`` CLI via subprocess without mocking the service layer. Domain-level tests (5-9) exercise real application code directly. """ from __future__ import annotations import os import shutil import subprocess import sys from collections.abc import Callable from datetime import datetime from pathlib import Path from typing import NoReturn # Ensure the src directory is on the import path. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) # Ensure robot/ is on the import path for helper_e2e_common. _ROBOT = str(Path(__file__).resolve().parent) if _ROBOT not in sys.path: sys.path.insert(0, _ROBOT) from helper_e2e_common import ( # noqa: E402 cleanup_workspace, init_bare_git_repo, is_expected_provider_unavailable, run_cli, setup_workspace, write_yaml, ) from helpers_common import reset_global_state # noqa: E402 from cleveragents.domain.models.core.action import Action, ActionState # noqa: E402 from cleveragents.domain.models.core.change import ( # noqa: E402 ChangeEntry, ChangeOperation, InMemoryChangeSetStore, ToolInvocation, ) from cleveragents.domain.models.core.plan import ( # noqa: E402 AutomationProfileProvenance, AutomationProfileRef, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) _PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8M" _VALID_ACTION_YAML = """\ name: local/m1-verify-action description: M1 verification action strategy_actor: openai/gpt-4 execution_actor: openai/gpt-4 definition_of_done: All M1 criteria pass """ def _extract_plan_id(output: str) -> str | None: """Extract a ULID plan_id from plain CLI output. Searches for a 26-character ULID pattern (uppercase alphanumeric) which is the format used by ``python-ulid``. """ import re match = re.search(r"\b([0-9A-Z]{26})\b", output) return match.group(1) if match else None def _mock_action(name: str = "local/m1-verify-action") -> Action: """Create a minimal valid Action for testing.""" return Action( namespaced_name=NamespacedName.parse(name), description="M1 verification action", long_description=None, definition_of_done="All M1 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_id: str = _PLAN_ULID, ) -> Plan: """Create a minimal valid Plan for testing.""" now = datetime.now() return Plan( identity=PlanIdentity(plan_id=plan_id), namespaced_name=NamespacedName.parse("local/m1-verify-plan"), description="M1 verification plan", definition_of_done="All M1 criteria pass", action_name="local/m1-verify-action", phase=phase, processing_state=state, project_links=[ProjectLink(project_name="local/m1-project")], arguments={"target_coverage": 97}, arguments_order=["target_coverage"], automation_profile=AutomationProfileRef( profile_name="trusted", provenance=AutomationProfileProvenance.PLAN, ), 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 _make_store() -> tuple[InMemoryChangeSetStore, str]: """Create an InMemoryChangeSetStore with one changeset started.""" store = InMemoryChangeSetStore() cid = store.start(_PLAN_ULID) return store, cid def _fail(msg: str) -> NoReturn: """Print failure message and exit.""" print(f"FAIL: {msg}", file=sys.stderr) raise SystemExit(1) # --------------------------------------------------------------------------- # Subcommand: action-create # --------------------------------------------------------------------------- def action_create_from_yaml() -> None: """Verify action create from YAML config via real CLI subprocess.""" reset_global_state() workspace = setup_workspace(prefix="m1_action_") yaml_path = write_yaml(_VALID_ACTION_YAML) try: result = run_cli("action", "create", "--config", yaml_path, workspace=workspace) if result.returncode != 0: _fail(f"action create rc={result.returncode}\n{result.stderr}") # Verify persistence: the action should be retrievable show = run_cli( "action", "show", "local/m1-verify-action", "--format", "plain", workspace=workspace, ) if show.returncode != 0: _fail(f"action show rc={show.returncode}\n{show.stderr}") if "local/m1-verify-action" not in show.stdout: _fail(f"action not found in show output:\n{show.stdout}") print("m1-action-create-ok") finally: os.unlink(yaml_path) cleanup_workspace(workspace) # --------------------------------------------------------------------------- # Subcommand: resource-register # --------------------------------------------------------------------------- def resource_register_git_checkout() -> None: """Register a git-checkout resource via real CLI subprocess.""" reset_global_state() workspace = setup_workspace(prefix="m1_resource_") repo_dir = init_bare_git_repo() try: result = run_cli( "resource", "add", "git-checkout", "local/m1-repo", "--path", repo_dir, "--branch", "main", "--format", "plain", workspace=workspace, ) if result.returncode != 0: _fail(f"resource add rc={result.returncode}\n{result.stderr}") # Verify persistence: the resource should be retrievable show = run_cli( "resource", "show", "local/m1-repo", "--format", "plain", workspace=workspace, ) if show.returncode != 0: _fail(f"resource show rc={show.returncode}\n{show.stderr}") if "git-checkout" not in show.stdout: _fail(f"resource type not git-checkout:\n{show.stdout}") print("m1-resource-register-ok") finally: shutil.rmtree(repo_dir, ignore_errors=True) cleanup_workspace(workspace) # --------------------------------------------------------------------------- # Subcommand: project-create-link # --------------------------------------------------------------------------- def project_create_and_link() -> None: """Create a project and link a resource via real CLI subprocess.""" reset_global_state() workspace = setup_workspace(prefix="m1_project_") repo_dir = init_bare_git_repo() try: # First register a resource (dependency for --resource flag) r1 = run_cli( "resource", "add", "git-checkout", "local/m1-repo", "--path", repo_dir, "--branch", "main", "--format", "plain", workspace=workspace, ) if r1.returncode != 0: _fail(f"resource add rc={r1.returncode}\n{r1.stderr}") # Create project with resource link r2 = run_cli( "project", "create", "local/m1-project", "--description", "M1 test project", "--resource", "local/m1-repo", "--format", "plain", workspace=workspace, ) if r2.returncode != 0: _fail(f"project create rc={r2.returncode}\n{r2.stderr}") # Verify persistence r3 = run_cli( "project", "show", "local/m1-project", "--format", "plain", workspace=workspace, ) if r3.returncode != 0: _fail(f"project show rc={r3.returncode}\n{r3.stderr}") if "local/m1-project" not in r3.stdout: _fail(f"project not found in show output:\n{r3.stdout}") print("m1-project-create-link-ok") finally: shutil.rmtree(repo_dir, ignore_errors=True) cleanup_workspace(workspace) # --------------------------------------------------------------------------- # Subcommand: plan-lifecycle # --------------------------------------------------------------------------- def plan_full_lifecycle() -> None: """Run the plan lifecycle via real CLI subprocesses. Invokes action create -> plan use -> plan show -> plan execute -> plan diff -> plan apply using the real ``agents`` CLI binary. Without an AI backend the plan stays in ``strategize / queued``, so ``execute``, ``diff``, and ``apply`` return graceful "not ready" messages rather than crashing. The test verifies the full CLI wiring (DI, persistence, env vars) end-to-end. ChangeSet verification uses the domain-level tests (``changeset-invocations``), not this CLI lifecycle test. """ reset_global_state() workspace = setup_workspace(prefix="m1_plan_") repo_dir = init_bare_git_repo() yaml_path = write_yaml(_VALID_ACTION_YAML) try: # Setup: register resource + create project r_res = run_cli( "resource", "add", "git-checkout", "local/m1-repo", "--path", repo_dir, "--branch", "main", workspace=workspace, ) if r_res.returncode != 0: _fail(f"resource add: {r_res.stderr}") r_proj = run_cli( "project", "create", "local/m1-project", "--resource", "local/m1-repo", workspace=workspace, ) if r_proj.returncode != 0: _fail(f"project create: {r_proj.stderr}") # Step 1: Create action r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace) if r1.returncode != 0: _fail(f"action create: {r1.stderr}") # Step 2: Plan use (--format plain to extract plan_id) r2 = run_cli( "plan", "use", "local/m1-verify-action", "local/m1-project", "--format", "plain", workspace=workspace, ) if r2.returncode != 0: _fail(f"plan use: {r2.stderr}") # Extract plan_id from plain output plan_id = _extract_plan_id(r2.stdout) if not plan_id: _fail(f"could not extract plan_id from plan use output:\n{r2.stdout}") # Step 3: Verify plan persisted — status must find it r3 = run_cli( "plan", "status", plan_id, "--format", "plain", workspace=workspace, ) if r3.returncode != 0: _fail(f"plan status: {r3.stderr}") if plan_id not in r3.stdout: _fail(f"plan_id not in plan status output:\n{r3.stdout}") # Step 4: Plan execute — plan is strategize/queued without # an AI backend so it correctly rejects with "not ready". # The test verifies no internal server error (500); a # controlled abort with a user-facing message is expected. r4 = run_cli("plan", "execute", plan_id, workspace=workspace) combined4 = r4.stdout + r4.stderr if "Traceback" in combined4: _fail(f"plan execute crashed:\n{combined4}") if "INTERNAL" in combined4 and not is_expected_provider_unavailable(combined4): _fail(f"plan execute crashed:\n{combined4}") # Step 5: Plan diff — similar: plan not in execute phase r5 = run_cli("plan", "diff", plan_id, workspace=workspace) combined5 = r5.stdout + r5.stderr if "Traceback" in combined5: _fail(f"plan diff crashed:\n{combined5}") if "INTERNAL" in combined5 and not is_expected_provider_unavailable(combined5): _fail(f"plan diff crashed:\n{combined5}") # Step 6: Plan apply — similar: plan not ready r6 = run_cli("plan", "apply", "--yes", plan_id, workspace=workspace) combined6 = r6.stdout + r6.stderr if "Traceback" in combined6: _fail(f"plan apply crashed:\n{combined6}") if "INTERNAL" in combined6 and not is_expected_provider_unavailable(combined6): _fail(f"plan apply crashed:\n{combined6}") print("m1-plan-lifecycle-ok") finally: os.unlink(yaml_path) shutil.rmtree(repo_dir, ignore_errors=True) cleanup_workspace(workspace) # --------------------------------------------------------------------------- # Subcommand: sqlite-persistence # --------------------------------------------------------------------------- def sqlite_persistence_check() -> None: """Verify Plan and Action records persist to SQLite via domain models. Uses the in-memory ChangeSetStore and domain model constructors to verify data survives serialisation round-trips. """ reset_global_state() # Create Action and verify serialisation action = _mock_action() action_dict = { "namespaced_name": str(action.namespaced_name), "state": action.state.value, "description": action.description, "definition_of_done": action.definition_of_done, "strategy_actor": action.strategy_actor, "execution_actor": action.execution_actor, } if action_dict["state"] != "available": _fail(f"action state={action_dict['state']}") # Create Plan and verify serialisation plan = _mock_plan() plan_dict = { "plan_id": plan.identity.plan_id, "namespaced_name": str(plan.namespaced_name), "phase": plan.phase.value, "processing_state": plan.processing_state.value, "action_name": plan.action_name, } if plan_dict["phase"] != "strategize": _fail(f"plan phase={plan_dict['phase']}") if plan_dict["plan_id"] != _PLAN_ULID: _fail(f"plan_id mismatch: {plan_dict['plan_id']}") # Verify ChangeSet round-trip store = InMemoryChangeSetStore() cid = store.start(_PLAN_ULID) store.record( cid, ChangeEntry( plan_id=_PLAN_ULID, resource_id="res-persist", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path="src/persisted.py", ), ) cs = store.get(cid) if cs is None: _fail("changeset not found") if cs.plan_id != _PLAN_ULID: _fail(f"changeset plan_id mismatch: {cs.plan_id}") if len(cs.entries) != 1: _fail(f"expected 1 entry, got {len(cs.entries)}") # Verify changesets retrievable by plan_id plan_sets = store.get_for_plan(_PLAN_ULID) if len(plan_sets) != 1: _fail(f"expected 1 changeset for plan, got {len(plan_sets)}") print("m1-sqlite-persistence-ok") # --------------------------------------------------------------------------- # Subcommand: changeset-from-invocations # --------------------------------------------------------------------------- def changeset_from_tool_invocations() -> None: """Verify ChangeSet is built from tool invocations, not parsed output. Creates ToolInvocation records and verifies ChangeEntry objects are linked via tool_name, not extracted from CLI output text. """ reset_global_state() store, cid = _make_store() # Simulate tool invocations producing ChangeEntry records inv = ToolInvocation( plan_id=_PLAN_ULID, tool_name="builtin/file-write", arguments={"path": "src/tool_generated.py", "content": "# new"}, success=True, duration_ms=42.0, sequence_number=0, ) # Record the change entry that this invocation produced entry = ChangeEntry( plan_id=_PLAN_ULID, resource_id="res-inv", tool_name=inv.tool_name, operation=ChangeOperation.CREATE, path="src/tool_generated.py", ) store.record(cid, entry) # Link invocation to change inv.change_ids.append(entry.entry_id) # Verify: entry tool_name matches invocation tool_name cs = store.get(cid) if cs is None: _fail("changeset not found") if cs.entries[0].tool_name != "builtin/file-write": _fail(f"tool_name mismatch: {cs.entries[0].tool_name}") if entry.entry_id not in inv.change_ids: _fail("entry not linked to invocation") print("m1-changeset-invocations-ok") # --------------------------------------------------------------------------- # Subcommand: sandbox-isolation # --------------------------------------------------------------------------- def sandbox_isolation_check() -> None: """Verify git worktree sandbox creates isolated working directory. Also verifies sandbox changes do not affect the original repo until Apply (commit + merge). """ reset_global_state() from cleveragents.infrastructure.sandbox.git_worktree import ( GitWorktreeSandbox, ) from cleveragents.infrastructure.sandbox.protocol import SandboxStatus repo_dir = init_bare_git_repo() try: # Create sandbox sandbox = GitWorktreeSandbox( resource_id="res-sandbox", original_path=repo_dir, ) ctx = sandbox.create(plan_id=_PLAN_ULID) # Verify sandbox created in isolated directory if not os.path.isdir(ctx.sandbox_path): _fail("sandbox path does not exist") if ctx.sandbox_path == repo_dir: _fail("sandbox path same as original") if sandbox.status != SandboxStatus.CREATED: _fail(f"expected CREATED, got {sandbox.status}") # Write a file in the sandbox sandbox_file = sandbox.get_path("src/sandbox_only.py") os.makedirs(os.path.dirname(sandbox_file), exist_ok=True) with open(sandbox_file, "w") as f: f.write("# created in sandbox\n") # Verify original repo does NOT have the sandbox file original_file = os.path.join(repo_dir, "src", "sandbox_only.py") if os.path.exists(original_file): _fail("sandbox change leaked to original before apply") # Commit and verify the merge applies to original result = sandbox.commit("sandbox: apply changes") if not result.success: _fail(f"commit failed: {result.error}") if result.commit_ref is None: _fail("no commit ref after commit") # Now original repo should have the file if not os.path.exists(original_file): _fail("file not in original after commit/merge") sandbox.cleanup() print("m1-sandbox-isolation-ok") finally: shutil.rmtree(repo_dir, ignore_errors=True) # --------------------------------------------------------------------------- # Subcommand: post-apply-commit # --------------------------------------------------------------------------- def post_apply_commit_check() -> None: """Verify post-apply commit exists in the target repo. Creates a git worktree sandbox, writes a file, commits, and verifies the commit is visible in git log on the original branch. """ reset_global_state() from cleveragents.infrastructure.sandbox.git_worktree import ( GitWorktreeSandbox, ) repo_dir = init_bare_git_repo() try: sandbox = GitWorktreeSandbox( resource_id="res-commit-check", original_path=repo_dir, ) sandbox.create(plan_id=_PLAN_ULID) # Write and commit new_file = sandbox.get_path("post_apply_test.txt") with open(new_file, "w") as f: f.write("post-apply verification content\n") commit_result = sandbox.commit("sandbox: post-apply test") if not commit_result.success: _fail(f"commit failed: {commit_result.error}") sandbox.cleanup() # Verify commit is in git log of original repo log_output = subprocess.run( ["git", "log", "--oneline", "-5"], cwd=repo_dir, capture_output=True, text=True, check=True, ) if "post-apply test" not in log_output.stdout: _fail(f"commit not found in git log:\n{log_output.stdout}") # Verify the file exists on disk in original if not os.path.exists(os.path.join(repo_dir, "post_apply_test.txt")): _fail("post-apply file not on disk") print("m1-post-apply-commit-ok") finally: shutil.rmtree(repo_dir, ignore_errors=True) # --------------------------------------------------------------------------- # Main dispatcher # --------------------------------------------------------------------------- _COMMANDS: dict[str, Callable[[], None]] = { "action-create": action_create_from_yaml, "resource-register": resource_register_git_checkout, "project-create-link": project_create_and_link, "plan-lifecycle": plan_full_lifecycle, "sqlite-persistence": sqlite_persistence_check, "changeset-invocations": changeset_from_tool_invocations, "sandbox-isolation": sandbox_isolation_check, "post-apply-commit": post_apply_commit_check, } def main() -> int: """Entry point called by Robot Framework ``Run Process``.""" if len(sys.argv) < 2: print( f"Usage: helper_m1_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())