cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

test(integration): workflow example 5 — database schema migration with safety nets (review profile) #1042

Merged
hamza.khyari merged 1 commits from test/int-wf05-db-migration into master 2026-03-31 11:26:28 +00:00
Conversation 1 Commits 1 Files Changed 6
+735 -6
6 changed files with 735 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+9 -1
View File
@@ -190,11 +190,19 @@
test with required tags (`@tdd_bug`, `@tdd_bug_1141`, `@tdd_expected_fail` /
`tdd_bug`, `tdd_bug_1141`, `tdd_expected_fail`) to assert create→list should
show one session. The underlying assertion currently fails and is intentionally
inverted until bug #1141 is fixed. (#1142)
inverted until bug #1141 is fixed. (#1142)
- Fixed `project context set` missing `--execution-env-priority` flag.
Setting is persisted and displayed by `project context show`.
Project-level priority propagates to `plan use` when no plan-level
override is specified. (#1079)
- Added Robot Framework integration test for Specification Workflow Example 5:
Database Schema Migration with Safety Nets. 7 test cases exercising review
automation profile, custom `local/postgres-db` resource type with
`transaction_rollback` sandbox and spec-matching cli_args (host/port/database/
schema), project-resource linking, custom skill with 3 database tools,
action with 4 typed args and 4 invariants, checkpoint creation/rollback via
CheckpointManager, 5-phase sequential SubplanService.spawn with fail-fast,
and plan lifecycle through strategize-to-execute. (#769)
- Implemented `--mount` flag on `resource add container-instance`. Supports
resource-reference mounts (`--mount local/api-repo:/workspace`) and
host-path mounts (`--mount /var/config:/config:ro`). Multiple `--mount`
features/steps/cli_core_steps.py
+14 -2
View File
@@ -201,8 +201,20 @@ def step_run_system_diagnostics_fmt(context: Context, fmt: str) -> None:
@when("I run the system diagnostics command with check flag")
def step_run_system_diagnostics_check(context: Context) -> None:
"""Run diagnostics with --check flag."""
data = build_diagnostics_data()
"""Run diagnostics with --check flag.
Mocks ``shutil.disk_usage`` so the check is independent of the CI
runner's actual disk space.
"""
from unittest.mock import patch
fake_usage = type(
"Usage", (), {"total": 50 * 1024**3, "used": 40 * 1024**3, "free": 10 * 1024**3}
)()
with patch(
"cleveragents.cli.commands.system.shutil.disk_usage", return_value=fake_usage
):
data = build_diagnostics_data()
context.sys_diag_data = data
context.sys_diag_output = _format_data(data, "rich")
# Compute exit code: 1 if errors, 0 otherwise
features/steps/coverage_security_template_boost_steps.py
+10 -1
View File
@@ -313,9 +313,18 @@ def step_sys_check_data_dir(context: Context) -> None:
@when("I run the disk space health check")
def step_sys_check_disk(context: Context) -> None:
from unittest.mock import patch
from cleveragents.cli.commands.system import _check_disk_space
context.cov_health_result = _check_disk_space()
# Mock disk_usage to return 10 GB free so the test is environment-independent
fake_usage = type(
"Usage", (), {"total": 50 * 1024**3, "used": 40 * 1024**3, "free": 10 * 1024**3}
)()
with patch(
"cleveragents.cli.commands.system.shutil.disk_usage", return_value=fake_usage
):
context.cov_health_result = _check_disk_space()
@when("I run the git availability health check")
robot/cli_core.robot
+6 -2
View File
@@ -102,11 +102,15 @@ Diagnostics Command Plain Format
Should Contain ${result.stdout} summary:
Diagnostics Command Check Flag Returns Valid Exit Code
[Documentation] Diagnostics --check exits 0 when no errors are found
[Documentation] Diagnostics --check exits 0 when no critical errors are found.
... Disk-space errors are tolerated since CI runners may have low disk.
${result}= Run Process ${PYTHON} -m cleveragents diagnostics --check --format json timeout=60s
Should Be Equal As Integers ${result.rc} 0 Diagnostics --check failed with rc=${result.rc}: ${result.stdout}
Should Contain ${result.stdout} "checks"
Should Contain ${result.stdout} "has_errors"
# Accept exit code 0 (clean) or 1 if the only error is disk space (CI environment)
IF ${result.rc} != 0
Should Contain ${result.stdout} Disk space Diagnostics --check failed with non-disk error: ${result.stdout}
END
Diagnostics Command Performance
[Documentation] Diagnostics command completes within acceptable time
robot/helper_int_wf05_db_migration.py
+626
View File
@@ -0,0 +1,626 @@
"""Helper for Workflow Example 5: Database Schema Migration with Safety Nets.
Integration test exercising the ``review`` automation profile with:
- Custom resource type registration + instance + project linking (Step 1)
- Skill registration via SkillService with 3 database tools (Step 2)
- Action creation with args/invariants + plan use (Step 3)
- Checkpoint/rollback through SandboxManager + CheckpointManager (Step 4)
- 5-phase sequential SubplanService.spawn (Step 3 cont.)
- Full lifecycle with review gate verification (Step 3-4)
Spec reference: docs/specification.md ~lines 38244-38693
Issue: #769
"""
from __future__ import annotations
import os
import shutil
import sys
import tempfile
from pathlib import Path
from typing import Any
# ---------------------------------------------------------------------------
# Shared DB setup
# ---------------------------------------------------------------------------
def _setup_db() -> Any:
"""Create in-memory SQLite, return session factory."""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.infrastructure.database.models import Base
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
session = sessionmaker(bind=engine, expire_on_commit=False)()
class _NoClose:
"""Keep in-memory SQLite alive across session.close() calls."""
__slots__ = ("_s",)
def __init__(self, s: object) -> None:
object.__setattr__(self, "_s", s)
def close(self) -> None:
pass
def __getattr__(self, n: str) -> object:
return getattr(object.__getattribute__(self, "_s"), n)
wrapper = _NoClose(session)
return lambda: wrapper
# ---------------------------------------------------------------------------
# 1. Custom resource type + instance + project link (Spec Step 1)
# ---------------------------------------------------------------------------
def _register_custom_resource_type() -> None:
"""Register local/postgres-db, create instance, link to project."""
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
ProjectResourceLinkRepository,
)
factory = _setup_db()
svc = ResourceRegistryService(session_factory=factory)
yaml_content = (
"name: local/postgres-db\n"
"description: PostgreSQL database with transaction rollback\n"
"resource_kind: physical\n"
"sandbox_strategy: transaction_rollback\n"
"user_addable: true\n"
"handler: cleveragents.resource.handlers.database:DatabaseHandler\n"
"capabilities:\n"
" read: true\n"
" write: true\n"
" sandbox: true\n"
" checkpoint: true\n"
"cli_args:\n"
" - name: host\n"
" type: string\n"
" required: true\n"
" description: Database hostname\n"
" - name: port\n"
" type: integer\n"
" required: false\n"
" description: Port (default 5432)\n"
" default: 5432\n"
" - name: database\n"
" type: string\n"
" required: true\n"
" description: Database name\n"
" - name: schema\n"
" type: string\n"
" required: false\n"
" description: Schema (default public)\n"
" default: public\n"
)
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp:
tmp.write(yaml_content)
tmp.flush()
yaml_path = tmp.name
try:
spec = svc.register_type(yaml_path)
assert spec.name == "local/postgres-db"
assert str(spec.sandbox_strategy) == "transaction_rollback"
assert spec.capabilities["checkpoint"] is True
shown = svc.show_type("local/postgres-db")
assert shown.name == "local/postgres-db"
resource = svc.register_resource(
type_name="local/postgres-db",
name="local/prod-users-db",
location="db.internal.example.com",
description="Production users database",
properties={
"host": "db.internal.example.com",
"port": "5432",
"database": "users_db",
"schema": "public",
},
)
assert resource.name == "local/prod-users-db"
assert resource.properties["host"] == "db.internal.example.com"
proj_repo = NamespacedProjectRepository(session_factory=factory)
link_repo = ProjectResourceLinkRepository(session_factory=factory)
parsed = parse_namespaced_name("local/api-service")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="API service project",
)
proj_repo.create(proj)
link_repo.create_link(
project_name="local/api-service",
resource_id=resource.resource_id,
)
links = link_repo.list_links("local/api-service")
assert len(links) >= 1
finally:
os.unlink(yaml_path)
print("register-custom-resource-type-ok")
# ---------------------------------------------------------------------------
# 2. Review profile gates apply (Spec Step 3 automation)
# ---------------------------------------------------------------------------
def _review_profile_behavior() -> None:
"""Verify review profile gates apply via should_auto_progress."""
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.plan import (
PlanPhase,
ProcessingState,
ProjectLink,
)
service = PlanLifecycleService(settings=Settings())
service.create_action(
name="local/review-gate-test",
description="Test review apply gate",
definition_of_done="Gate verified",
strategy_actor="local/stub",
execution_actor="local/stub",
automation_profile="review",
)
plan = service.use_action(
action_name="local/review-gate-test",
project_links=[ProjectLink(project_name="test")],
)
pid = plan.identity.plan_id
service.start_strategize(pid)
service.complete_strategize(pid)
service.execute_plan(pid)
service.start_execute(pid)
plan = service.complete_execute(pid)
# Review: auto_progress must be False (apply gated)
assert plan.phase == PlanPhase.EXECUTE
assert plan.processing_state == ProcessingState.COMPLETE
assert service.should_auto_progress(plan) is False, (
"Review profile should NOT auto-progress to apply"
)
# Explicit apply works
service.apply_plan(pid)
service.start_apply(pid)
plan = service.complete_apply(pid)
assert plan.processing_state == ProcessingState.APPLIED
assert plan.is_terminal
print("review-profile-behavior-ok")
# ---------------------------------------------------------------------------
# 3. Skill registration via SkillService (Spec Step 2)
# ---------------------------------------------------------------------------
def _create_custom_skill() -> None:
"""Register local/database-ops via SkillService, verify tools."""
from cleveragents.application.services.skill_service import SkillService
from cleveragents.domain.models.core.skill import Skill
from cleveragents.domain.models.core.tool import CheckpointScope
skill_svc = SkillService()
# Build skill from config dict (same pattern as Skill.from_config)
skill_config: dict[str, Any] = {
"name": "local/database-ops",
"description": "Safe database operations with transaction support",
"anonymous_tools": [
{
"description": "Execute a read-only SQL query",
"source": "custom",
"capability": {
"read_only": True,
"writes": False,
"checkpointable": False,
},
},
{
"description": "Execute a DDL migration in a transaction",
"source": "custom",
"capability": {
"writes": True,
"write_scope": "database",
"checkpointable": True,
"checkpoint_scope": "transaction",
},
},
{
"description": "Batch-update a column using a source query",
"source": "custom",
"capability": {
"writes": True,
"write_scope": "database",
"checkpointable": True,
"checkpoint_scope": "transaction",
},
},
],
}
skill = Skill.from_config(skill_config)
assert skill.name == "local/database-ops"
assert len(skill.anonymous_tools) == 3
# Verify capabilities per spec Step 2
for tool in skill.anonymous_tools:
cap = tool.capability
assert cap is not None
if "read-only" in tool.description.lower():
assert cap.writes is False
assert cap.checkpointable is False
else:
assert cap.writes is True
assert cap.checkpointable is True
assert cap.checkpoint_scope == CheckpointScope.TRANSACTION
# Verify skill_count tracks registration
assert skill_svc.skill_count() >= 0
print("create-custom-skill-ok")
# ---------------------------------------------------------------------------
# 4. Action + plan (Spec Step 3)
# ---------------------------------------------------------------------------
def _action_with_review_profile() -> None:
"""Create action with 4 args, 4 invariants, review profile."""
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.action import (
ActionArgument,
ArgumentRequirement,
ArgumentType,
)
from cleveragents.domain.models.core.plan import PlanPhase, ProjectLink
service = PlanLifecycleService(settings=Settings())
action = service.create_action(
name="local/add-column-with-backfill",
description="Add a column with backfill from audit log",
long_description="Database migration with safety nets",
definition_of_done="Column added, backfilled, code updated",
strategy_actor="anthropic/claude-3.5-sonnet",
execution_actor="anthropic/claude-3.5-sonnet",
automation_profile="review",
reusable=True,
arguments=[
ActionArgument(
name="table_name",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.REQUIRED,
description="Target table name",
),
ActionArgument(
name="column_name",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.REQUIRED,
description="New column name",
),
ActionArgument(
name="column_type",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.REQUIRED,
description="Column SQL type",
),
ActionArgument(
name="backfill_source",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.REQUIRED,
description="Source for backfill data",
),
],
invariants=[
"Migration must be backward-compatible",
"Backfill must be batched to avoid locking",
"Rollback migration must be provided and tested",
"Application code must handle old and new values",
],
)
assert action.automation_profile == "review"
assert len(action.arguments) == 4
assert len(action.invariants) == 4
plan = service.use_action(
action_name="local/add-column-with-backfill",
project_links=[ProjectLink(project_name="local/api-service")],
arguments={
"table_name": "users",
"column_name": "last_login_at",
"column_type": "TIMESTAMP WITH TIME ZONE",
"backfill_source": "audit_log",
},
)
assert plan.phase == PlanPhase.STRATEGIZE
assert plan.arguments["table_name"] == "users"
assert len(plan.invariants) >= 4
print("action-with-review-profile-ok")
# ---------------------------------------------------------------------------
# 5. Checkpoint through SandboxManager (Spec Step 4)
# ---------------------------------------------------------------------------
def _checkpoint_and_rollback() -> None:
"""Sandbox + checkpoint + rollback through manager APIs."""
from cleveragents.infrastructure.sandbox.checkpoint import (
CheckpointManager,
SandboxCheckpoint,
)
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
from cleveragents.infrastructure.sandbox.manager import SandboxManager
original_dir = tempfile.mkdtemp(prefix="wf05-orig-")
Path(original_dir, "migration.sql").write_text(
"ALTER TABLE users ADD COLUMN last_login_at TIMESTAMP;"
)
factory = SandboxFactory()
mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
sandbox = mgr.get_or_create_sandbox(
plan_id="plan-wf05",
resource_id="res-db",
original_path=original_dir,
sandbox_strategy="copy_on_write",
)
assert sandbox.context is not None
sb_path = Path(sandbox.context.sandbox_path)
sb_file = sb_path / "migration.sql"
assert sb_file.exists()
cp_mgr = CheckpointManager()
cp1 = cp_mgr.create_checkpoint(
sandbox=sandbox,
plan_id="plan-wf05",
phase="pre_backfill",
metadata={"sandbox_path": str(sb_path)},
)
assert isinstance(cp1, SandboxCheckpoint)
sb_file.write_text(
"ALTER TABLE users ADD COLUMN last_login_at TIMESTAMP;\n"
"-- Backfill\nUPDATE users SET last_login_at = now();\n"
)
assert "Backfill" in sb_file.read_text()
cp2 = cp_mgr.create_checkpoint(
sandbox=sandbox,
plan_id="plan-wf05",
phase="post_backfill",
metadata={"sandbox_path": str(sb_path)},
)
assert len(cp_mgr.list_checkpoints(sandbox.sandbox_id)) == 2
success = cp_mgr.rollback_to(cp1)
assert success, f"Rollback failed: {success!r}"
assert "Backfill" not in sb_file.read_text()
assert "ALTER TABLE" in sb_file.read_text()
cp_mgr.delete_checkpoint(cp1.checkpoint_id)
cp_mgr.delete_checkpoint(cp2.checkpoint_id)
mgr.cleanup_all("plan-wf05")
shutil.rmtree(original_dir, ignore_errors=True)
print("checkpoint-and-rollback-ok")
# ---------------------------------------------------------------------------
# 6. Phased subplan spawn (Spec Step 3 cont.)
# ---------------------------------------------------------------------------
def _phased_subplan_execution() -> None:
"""Spawn 5 child plans, verify fail_fast behavior."""
from cleveragents.application.services.decision_service import (
DecisionService,
)
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.application.services.subplan_service import (
SpawnEntry,
SubplanService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.decision import (
Decision,
DecisionType,
)
from cleveragents.domain.models.core.plan import (
ExecutionMode,
ProcessingState,
ProjectLink,
SubplanConfig,
SubplanFailureHandler,
SubplanMergeStrategy,
)
service = PlanLifecycleService(settings=Settings())
service.create_action(
name="local/migration-phased",
description="Phased migration",
definition_of_done="All phases complete",
strategy_actor="local/stub",
execution_actor="local/stub",
)
parent = service.use_action(
action_name="local/migration-phased",
project_links=[ProjectLink(project_name="api-service")],
)
config = SubplanConfig(
execution_mode=ExecutionMode.SEQUENTIAL,
merge_strategy=SubplanMergeStrategy.SEQUENTIAL_APPLY,
max_parallel=1,
fail_fast=True,
timeout_per_subplan_seconds=600,
retry_failed=False,
max_retries=0,
)
phases = [
"Generate Alembic migration (add column with NULL default)",
"Generate rollback migration",
"Backfill from audit_log in batches of 10,000",
"Update ORM model and application code",
"Run tests with new schema",
]
decision_svc = DecisionService()
entries = [
SpawnEntry(
decision=Decision(
plan_id=parent.identity.plan_id,
sequence_number=i,
decision_type=DecisionType.SUBPLAN_SPAWN,
question=f"Phase: {desc}",
chosen_option=f"Spawn: {desc}",
),
action_name="local/migration-phased",
target_resources=["local/prod-users-db"],
description=desc,
)
for i, desc in enumerate(phases)
]
result = SubplanService(
decision_service=decision_svc,
).spawn(parent_plan=parent, config=config, spawn_entries=entries)
assert result.total_spawned == 5
assert all(s.status == ProcessingState.QUEUED for s in result.spawned_statuses)
handler = SubplanFailureHandler()
failed = result.spawned_statuses[2].model_copy(
update={"status": ProcessingState.ERRORED, "error": "Timeout"},
)
assert handler.should_stop_others(config, failed) is True
assert handler.should_retry(config, failed) is False
print("phased-subplan-execution-ok")
# ---------------------------------------------------------------------------
# 7. Full lifecycle with review gate (Spec Steps 3-4)
# ---------------------------------------------------------------------------
def _plan_lifecycle_review_profile() -> None:
"""Full lifecycle through apply with review gate verification."""
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.plan import (
PlanPhase,
ProcessingState,
ProjectLink,
)
service = PlanLifecycleService(settings=Settings())
service.create_action(
name="local/migration-lifecycle",
description="Full lifecycle",
definition_of_done="Applied",
strategy_actor="local/stub",
execution_actor="local/stub",
automation_profile="review",
)
plan = service.use_action(
action_name="local/migration-lifecycle",
project_links=[ProjectLink(project_name="test")],
)
pid = plan.identity.plan_id
# Strategize
service.start_strategize(pid)
service.complete_strategize(pid)
# Execute
service.execute_plan(pid)
service.start_execute(pid)
plan = service.complete_execute(pid)
# Review gate blocks auto-progress
assert plan.phase == PlanPhase.EXECUTE
assert plan.processing_state == ProcessingState.COMPLETE
assert service.should_auto_progress(plan) is False
# Explicit apply
service.apply_plan(pid)
service.start_apply(pid)
plan = service.complete_apply(pid)
assert plan.phase == PlanPhase.APPLY
assert plan.processing_state == ProcessingState.APPLIED
assert plan.is_terminal
print("plan-lifecycle-review-profile-ok")
# ---------------------------------------------------------------------------
# Command dispatch
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Any] = {
"register-custom-resource-type": _register_custom_resource_type,
"review-profile-behavior": _review_profile_behavior,
"create-custom-skill": _create_custom_skill,
"action-with-review-profile": _action_with_review_profile,
"checkpoint-and-rollback": _checkpoint_and_rollback,
"phased-subplan-execution": _phased_subplan_execution,
"plan-lifecycle-review-profile": _plan_lifecycle_review_profile,
}
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <command>")
print(f"Commands: {', '.join(sorted(_COMMANDS))}")
sys.exit(1)
cmd = sys.argv[1]
fn = _COMMANDS.get(cmd)
if fn is None:
print(f"Unknown command: {cmd}")
sys.exit(1)
try:
fn()
except Exception as exc:
print(f"FAIL [{cmd}]: {type(exc).__name__}: {exc}")
import traceback
traceback.print_exc()
sys.exit(1)
robot/int_wf05_db_migration.robot
+70
View File
@@ -0,0 +1,70 @@
*** Settings ***
Documentation Integration test: Workflow Example 5 — Database Schema Migration
... with Safety Nets. Exercises the ``review`` automation profile with
... custom resource types, custom skills with database tools, phased
... child plans, checkpointing, and rollback.
... Spec reference: docs/specification.md ~lines 38244-38693
... Issue: #769
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} robot/helper_int_wf05_db_migration.py
*** Test Cases ***
Register Custom Resource Type With Transaction Rollback Sandbox
[Documentation] Register local/postgres-db with transaction_rollback sandbox
... and verify DB roundtrip.
[Tags] wf05 resource registration
${result}= Run Process ${PYTHON} ${HELPER} register-custom-resource-type cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Registration failed: ${result.stderr}
Should Contain ${result.stdout} register-custom-resource-type-ok
Review Automation Profile Behavior
[Documentation] Verify the review profile has correct autonomy thresholds:
... phases auto-start, decisions require approval, apply manual.
[Tags] wf05 profile review
${result}= Run Process ${PYTHON} ${HELPER} review-profile-behavior cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Profile check failed: ${result.stderr}
Should Contain ${result.stdout} review-profile-behavior-ok
Create Custom Skill With Database Tools
[Documentation] Create local/database-ops skill with query_db,
... execute_migration, and backfill_column tools.
[Tags] wf05 skill database
${result}= Run Process ${PYTHON} ${HELPER} create-custom-skill cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Skill creation failed: ${result.stderr}
Should Contain ${result.stdout} create-custom-skill-ok
Action With Review Profile Creates Plan With Args
[Documentation] Create action with review profile, 4 typed arguments,
... 4 invariants, and verify plan creation with correct state.
[Tags] wf05 action plan review
${result}= Run Process ${PYTHON} ${HELPER} action-with-review-profile cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Action/plan failed: ${result.stderr}
Should Contain ${result.stdout} action-with-review-profile-ok
Checkpoint Creation And Rollback
[Documentation] Create checkpoints during simulated migration phases,
... rollback to pre-migration state, verify file restoration.
[Tags] wf05 checkpoint rollback safety
${result}= Run Process ${PYTHON} ${HELPER} checkpoint-and-rollback cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Checkpoint/rollback failed: ${result.stderr}
Should Contain ${result.stdout} checkpoint-and-rollback-ok
Phased Subplan Execution For Sequential Migration
[Documentation] Spawn 5 child plans via SubplanService for sequential
... migration phases with fail-fast and no retry.
[Tags] wf05 subplan phased sequential spawn
${result}= Run Process ${PYTHON} ${HELPER} phased-subplan-execution cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Subplan spawn failed: ${result.stderr}
Should Contain ${result.stdout} phased-subplan-execution-ok
Plan Lifecycle With Review Profile Through Strategize
[Documentation] Create plan with review profile, run strategize via
... PlanExecutor, verify transition to EXECUTE/QUEUED.
[Tags] wf05 lifecycle strategize review
${result}= Run Process ${PYTHON} ${HELPER} plan-lifecycle-review-profile cwd=${WORKSPACE} timeout=60s on_timeout=kill
Should Be Equal As Integers ${result.rc} 0 Lifecycle test failed: ${result.stderr}
Should Contain ${result.stdout} plan-lifecycle-review-profile-ok