"""Helper script for WF04 multi-project dependency update Robot tests. Integration test for Specification Workflow Example 4: Multi-Project Dependency Update (spec §Workflow Example 4). Exercises the full workflow: 1. Register 4 git-checkout resources + create 4 projects + link them 2. Create action with args, invariants, automation_profile=supervised 3. Plan use with --arg flags across 4 projects 4. Strategize with child plan spawning 5. Execute children in dependency order (common-lib first) 6. Coordinated apply with validation gates Uses mocked LLM providers and in-memory SQLite database. """ from __future__ import annotations import sys import tempfile from pathlib import Path from unittest.mock import MagicMock from ulid import ULID _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from sqlalchemy import create_engine # noqa: E402 from sqlalchemy.orm import sessionmaker # noqa: E402 from cleveragents.application.services.multi_project_service import ( # noqa: E402 MultiProjectService, ) from cleveragents.application.services.plan_lifecycle_service import ( # noqa: E402 PlanLifecycleService, ) from cleveragents.application.services.resource_registry_service import ( # noqa: E402 ResourceRegistryService, ) from cleveragents.application.services.tool_registry_service import ( # noqa: E402 ToolRegistryService, ) from cleveragents.config.settings import Settings # noqa: E402 from cleveragents.domain.models.core.action import ( # noqa: E402 ActionArgument, ActionState, ) from cleveragents.domain.models.core.multi_project import ( # noqa: E402 ChangeSetSummary, ) from cleveragents.domain.models.core.plan import ( # noqa: E402 ExecutionMode, InvariantSource, PlanInvariant, PlanPhase, ProcessingState, ProjectLink, SubplanConfig, SubplanMergeStrategy, SubplanStatus, ) from cleveragents.domain.models.core.project import ( # noqa: E402 NamespacedProject, parse_namespaced_name, ) from cleveragents.domain.models.core.tool import Validation # noqa: E402 from cleveragents.infrastructure.database.models import Base # noqa: E402 from cleveragents.infrastructure.database.repositories import ( # noqa: E402 NamespacedProjectRepository, ProjectResourceLinkRepository, ToolRegistryRepository, ValidationAttachmentRepository, ) # --------------------------------------------------------------------------- # Spec constants (Specification §Workflow Example 4) # --------------------------------------------------------------------------- _PROJECTS = { "local/common-lib": { "description": "Shared common library", "resource": "local/common-lib-repo", "repo_desc": "Common library repository", }, "local/user-service": { "description": "User microservice", "resource": "local/user-service-repo", "repo_desc": "User service repository", }, "local/order-service": { "description": "Order microservice", "resource": "local/order-service-repo", "repo_desc": "Order service repository", }, "local/billing-service": { "description": "Billing microservice", "resource": "local/billing-service-repo", "repo_desc": "Billing service repository", }, } _VALIDATION_NAME = "local/pytest-mypy" _ACTION_NAME = "local/coordinated-dep-update" _ACTION_ARGS = [ ActionArgument.parse("library_name:string:required:Name of the shared library"), ActionArgument.parse("target_version:string:required:Target version to update to"), ActionArgument.parse("changelog_url:string:optional:URL to the library changelog"), ] _ACTION_INVARIANTS = [ "Each dependent project must be updated in its own child plan", "All child plans must pass validation before any can be applied", "The library update in common-lib must be applied first", ] _PLAN_ARGS = { "library_name": "authlib", "target_version": "2.0.0", "changelog_url": "https://authlib.org/changelog/2.0", } _PLAN_INVARIANTS = [ PlanInvariant( text="Each dependent project must be updated in its own child plan", source=InvariantSource.ACTION, ), PlanInvariant( text="All child plans must pass validation before any can be applied", source=InvariantSource.ACTION, ), PlanInvariant( text="The library update in common-lib must be applied first", source=InvariantSource.ACTION, ), ] _SUBPLAN_CONFIG = SubplanConfig( execution_mode=ExecutionMode.DEPENDENCY_ORDERED, merge_strategy=SubplanMergeStrategy.SEQUENTIAL_APPLY, max_parallel=3, fail_fast=True, ) _CHILD_IDS = [str(ULID()) for _ in range(4)] # --------------------------------------------------------------------------- # Session wrapper for in-memory SQLite # --------------------------------------------------------------------------- class _NoClose: """Prevent close() on shared in-memory session.""" __slots__ = ("_s",) def __init__(self, s: object) -> None: object.__setattr__(self, "_s", s) def close(self) -> None: object.__getattribute__(self, "_s").rollback() def __getattr__(self, n: str) -> object: return getattr(object.__getattribute__(self, "_s"), n) # --------------------------------------------------------------------------- # Service factories # --------------------------------------------------------------------------- def _setup_database() -> tuple[object, object]: """Create in-memory database and return (session_factory, engine).""" engine = create_engine("sqlite:///:memory:", echo=False) Base.metadata.create_all(engine) session = sessionmaker(bind=engine, expire_on_commit=False)() wrapper = _NoClose(session) return lambda: wrapper, engine def _setup_services() -> tuple[ PlanLifecycleService, ResourceRegistryService, NamespacedProjectRepository, ProjectResourceLinkRepository, MultiProjectService, ToolRegistryService, ]: """Set up all services with shared in-memory DB.""" factory, _ = _setup_database() registry = ResourceRegistryService(session_factory=factory) registry.bootstrap_builtin_types() proj_repo = NamespacedProjectRepository(session_factory=factory) link_repo = ProjectResourceLinkRepository(session_factory=factory) plan_svc = PlanLifecycleService(settings=Settings()) mock_decision = MagicMock() mock_decision.list_by_type.return_value = [] mp_svc = MultiProjectService(decision_service=mock_decision) tool_repo = ToolRegistryRepository(session_factory=factory) attach_repo = ValidationAttachmentRepository(session_factory=factory) tool_svc = ToolRegistryService( tool_repo=tool_repo, attachment_repo=attach_repo, ) return plan_svc, registry, proj_repo, link_repo, mp_svc, tool_svc def _make_child_statuses() -> list[SubplanStatus]: """Create SubplanStatus entries for all 4 projects.""" projects = list(_PROJECTS.keys()) return [ SubplanStatus( subplan_id=_CHILD_IDS[i], action_name=_ACTION_NAME, target_resources=[proj], status=ProcessingState.QUEUED, ) for i, proj in enumerate(projects) ] # --------------------------------------------------------------------------- # Step 1: Register resources + create projects + link # --------------------------------------------------------------------------- def cmd_register_projects() -> None: """Spec Step 1: Register 4 resources, create 4 projects, link them. Also registers a shared validation (local/pytest-mypy) and attaches it to all 4 projects per spec §Workflow Example 4. """ _, registry, proj_repo, link_repo, _, tool_svc = _setup_services() with tempfile.TemporaryDirectory() as tmpdir: for proj_name, info in _PROJECTS.items(): # Create a temp directory as repo location repo_path = Path(tmpdir) / info["resource"].split("/")[-1] repo_path.mkdir() # Register git-checkout resource res = registry.register_resource( type_name="git-checkout", name=info["resource"], location=str(repo_path), description=info["repo_desc"], properties={"path": str(repo_path), "branch": "main"}, ) assert res.name == info["resource"] # Create project parsed = parse_namespaced_name(proj_name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, description=info["description"], ) proj_repo.create(proj) fetched = proj_repo.get(proj.namespaced_name) assert fetched is not None assert fetched.namespaced_name == proj_name # Link resource to project link_repo.create_link( project_name=proj_name, resource_id=res.resource_id, ) links = link_repo.list_links(proj_name) assert len(links) == 1 # Register shared validation (spec: "agents validation add ... local/pytest-mypy") validation = Validation.from_config( { "name": _VALIDATION_NAME, "description": "Run pytest and mypy", "source": "custom", "code": "pytest tests/ && mypy src/", "mode": "required", "timeout": 300, } ) tool_svc.register_tool(validation) # Attach validation to all 4 projects (spec: "agents validation attach ...") for proj_name, info in _PROJECTS.items(): tool_svc.attach_validation( validation_name=_VALIDATION_NAME, resource_id=info["resource"], project_name=proj_name, ) # Verify attachments for proj_name, info in _PROJECTS.items(): attached = tool_svc.list_validations_for_resource( resource_id=info["resource"], project_name=proj_name, ) assert len(attached) == 1, ( f"Expected 1 validation for {proj_name}, got {len(attached)}" ) print("register-projects-ok") # --------------------------------------------------------------------------- # Step 2: Create action with full spec fields # --------------------------------------------------------------------------- def cmd_create_action() -> None: """Spec Step 2: Create action with args, invariants, supervised profile.""" plan_svc, *_ = _setup_services() action = plan_svc.create_action( name=_ACTION_NAME, description="Update a shared dependency across multiple projects", long_description=( "When a shared library introduces a breaking change, this " "action coordinates the update across all dependent projects." ), definition_of_done=( "The shared library is updated, all dependent projects compile " "and pass tests, API contracts remain compatible." ), strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", automation_profile="supervised", reusable=True, arguments=_ACTION_ARGS, invariants=_ACTION_INVARIANTS, ) assert action.state == ActionState.AVAILABLE assert str(action.namespaced_name) == _ACTION_NAME assert action.reusable is True assert action.automation_profile == "supervised" assert len(action.arguments) == 3 assert len(action.invariants) == 3 assert action.long_description is not None print("create-action-ok") # --------------------------------------------------------------------------- # Step 3: Plan use with args and supervised profile # --------------------------------------------------------------------------- def cmd_plan_use() -> None: """Spec Step 3: Plan use with --arg flags and 4 project links.""" plan_svc, *_ = _setup_services() plan_svc.create_action( name=_ACTION_NAME, description="Coordinated dep update", definition_of_done="All updated", strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", automation_profile="supervised", arguments=_ACTION_ARGS, invariants=_ACTION_INVARIANTS, ) project_links = [ProjectLink(project_name=p) for p in _PROJECTS] plan = plan_svc.use_action( action_name=_ACTION_NAME, project_links=project_links, arguments=_PLAN_ARGS, created_by="wf04-integration-test", ) assert plan.phase == PlanPhase.STRATEGIZE assert plan.state == ProcessingState.QUEUED assert len(plan.project_links) == 4 assert plan.arguments == _PLAN_ARGS # Action invariants are inherited automatically from the action assert len(plan.invariants) == 3, ( f"Expected 3 invariants (from action), got {len(plan.invariants)}" ) print("plan-use-ok") # --------------------------------------------------------------------------- # Step 3 continued: Strategize with child plan config # --------------------------------------------------------------------------- def cmd_strategize_cycle() -> None: """Drive strategize, attach subplan config and children.""" plan_svc, *_ = _setup_services() plan_svc.create_action( name=_ACTION_NAME, description="Coordinated dep update", definition_of_done="All updated", strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", arguments=_ACTION_ARGS, invariants=_ACTION_INVARIANTS, ) plan = plan_svc.use_action( action_name=_ACTION_NAME, project_links=[ProjectLink(project_name=p) for p in _PROJECTS], arguments=_PLAN_ARGS, invariants=_PLAN_INVARIANTS, ) plan_id = plan.identity.plan_id plan = plan_svc.start_strategize(plan_id) assert plan.state == ProcessingState.PROCESSING # Attach subplan config via model_copy (strategy output) plan_with_config = plan.model_copy( update={ "subplan_config": _SUBPLAN_CONFIG, "subplan_statuses": _make_child_statuses(), } ) assert plan_with_config.has_subplans assert len(plan_with_config.subplan_statuses) == 4 assert ( plan_with_config.subplan_config.execution_mode == ExecutionMode.DEPENDENCY_ORDERED ) plan = plan_svc.complete_strategize(plan_id) assert plan.state == ProcessingState.COMPLETE print("strategize-cycle-ok") # --------------------------------------------------------------------------- # Child plan spawning # --------------------------------------------------------------------------- def cmd_spawn_children() -> None: """Spawn 4 child plan statuses targeting each project.""" plan_svc, *_ = _setup_services() plan_svc.create_action( name=_ACTION_NAME, description="Coordinated dep update", definition_of_done="All updated", strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", ) plan = plan_svc.use_action( action_name=_ACTION_NAME, project_links=[ProjectLink(project_name=p) for p in _PROJECTS], ) plan = plan_svc.start_strategize(plan.identity.plan_id) children = _make_child_statuses() plan_with_children = plan.model_copy( update={ "subplan_config": _SUBPLAN_CONFIG, "subplan_statuses": children, } ) assert len(plan_with_children.subplan_statuses) == 4 assert plan_with_children.has_subplans targets = [s.target_resources[0] for s in plan_with_children.subplan_statuses] for project in _PROJECTS: assert project in targets, f"{project} not in subplan targets" print("spawn-children-ok") # --------------------------------------------------------------------------- # Execute children in dependency order # --------------------------------------------------------------------------- def cmd_execute_children() -> None: """Common-lib executes first, then services.""" plan_svc, *_ = _setup_services() plan_svc.create_action( name=_ACTION_NAME, description="Coordinated dep update", definition_of_done="All updated", strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", ) plan = plan_svc.use_action( action_name=_ACTION_NAME, project_links=[ProjectLink(project_name=p) for p in _PROJECTS], ) pid = plan.identity.plan_id plan_svc.start_strategize(pid) plan_svc.complete_strategize(pid) plan_svc.execute_plan(pid) plan = plan_svc.start_execute(pid) children = _make_child_statuses() # Phase 1: common-lib first (dependency root) for c in children: if "common-lib" in c.target_resources[0]: c.status = ProcessingState.COMPLETE c.files_changed = 4 # Phase 2: services after common-lib for c in children: if c.status != ProcessingState.COMPLETE: c.status = ProcessingState.COMPLETE c.files_changed = 5 plan_with_results = plan.model_copy( update={"subplan_config": _SUBPLAN_CONFIG, "subplan_statuses": children} ) assert all( s.status == ProcessingState.COMPLETE for s in plan_with_results.subplan_statuses ) common = next( s for s in plan_with_results.subplan_statuses if "common-lib" in s.target_resources[0] ) assert common.files_changed == 4 plan = plan_svc.complete_execute(pid) assert plan.state == ProcessingState.COMPLETE print("execute-children-ok") # --------------------------------------------------------------------------- # Coordinated apply # --------------------------------------------------------------------------- def cmd_apply_ordered() -> None: """Apply common-lib first, then services — coordinated apply.""" plan_svc, *_ = _setup_services() plan_svc.create_action( name=_ACTION_NAME, description="Coordinated dep update", definition_of_done="All updated", strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", ) plan = plan_svc.use_action( action_name=_ACTION_NAME, project_links=[ProjectLink(project_name=p) for p in _PROJECTS], ) pid = plan.identity.plan_id plan_svc.start_strategize(pid) plan_svc.complete_strategize(pid) plan_svc.execute_plan(pid) plan_svc.start_execute(pid) plan_svc.complete_execute(pid) plan = plan_svc.apply_plan(pid) assert plan.phase == PlanPhase.APPLY plan = plan_svc.start_apply(pid) children = _make_child_statuses() # Step 1: apply common-lib first (invariant #3) for c in children: if "common-lib" in c.target_resources[0]: c.status = ProcessingState.APPLIED applied_count = sum(1 for c in children if c.status == ProcessingState.APPLIED) assert applied_count == 1, "Only common-lib should be applied first" # Step 2: apply services after validation passes for c in children: if c.status != ProcessingState.APPLIED: c.status = ProcessingState.APPLIED assert all(c.status == ProcessingState.APPLIED for c in children) plan = plan_svc.complete_apply(pid) assert plan.phase == PlanPhase.APPLY assert plan.state == ProcessingState.APPLIED assert plan.is_terminal print("apply-ordered-ok") # --------------------------------------------------------------------------- # Full lifecycle end-to-end # --------------------------------------------------------------------------- def cmd_full_lifecycle() -> None: """End-to-end multi-project lifecycle with all spec requirements.""" plan_svc, registry, proj_repo, link_repo, mp_svc, tool_svc = _setup_services() # ── Step 1: Register resources, projects, link them ── with tempfile.TemporaryDirectory() as tmpdir: resource_ids = {} for proj_name, info in _PROJECTS.items(): repo_path = Path(tmpdir) / info["resource"].split("/")[-1] repo_path.mkdir() res = registry.register_resource( type_name="git-checkout", name=info["resource"], location=str(repo_path), description=info["repo_desc"], properties={"path": str(repo_path), "branch": "main"}, ) resource_ids[proj_name] = res.resource_id parsed = parse_namespaced_name(proj_name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, description=info["description"], ) proj_repo.create(proj) link_repo.create_link( project_name=proj_name, resource_id=res.resource_id, ) assert len(resource_ids) == 4 # Register shared validation and attach to all projects validation = Validation.from_config( { "name": _VALIDATION_NAME, "description": "Run pytest and mypy", "source": "custom", "code": "pytest tests/ && mypy src/", "mode": "required", "timeout": 300, } ) tool_svc.register_tool(validation) for proj_name, info in _PROJECTS.items(): tool_svc.attach_validation( validation_name=_VALIDATION_NAME, resource_id=info["resource"], project_name=proj_name, ) # ── Step 2: Create action with full spec fields ── action = plan_svc.create_action( name=_ACTION_NAME, description="Update a shared dependency across multiple projects", long_description=("Coordinates breaking-change updates across all dependents."), definition_of_done=( "All projects updated, tests pass, API contracts compatible." ), strategy_actor="anthropic/claude-3.5-sonnet", execution_actor="anthropic/claude-3.5-sonnet", automation_profile="supervised", reusable=True, arguments=_ACTION_ARGS, invariants=_ACTION_INVARIANTS, ) assert action.state == ActionState.AVAILABLE assert len(action.invariants) == 3 # ── Step 3: Plan use with args and 4 projects ── project_links = [ProjectLink(project_name=p) for p in _PROJECTS] plan = plan_svc.use_action( action_name=_ACTION_NAME, project_links=project_links, arguments=_PLAN_ARGS, created_by="wf04-integration-test", ) pid = plan.identity.plan_id assert len(plan.project_links) == 4 assert len(plan.invariants) == 3 # inherited from action # Initialize multi-project scopes resources = {p: [resource_ids[p]] for p in _PROJECTS} plan = mp_svc.initialize_scopes(plan, resources) assert plan.multi_project_metadata is not None assert len(plan.multi_project_metadata.project_scopes) == 4 # ── Strategize + spawn children ── plan_svc.start_strategize(pid) children = _make_child_statuses() plan_svc.complete_strategize(pid) # ── Execute: common-lib first, then services ── plan_svc.execute_plan(pid) plan_svc.start_execute(pid) for c in children: if "common-lib" in c.target_resources[0]: c.status = ProcessingState.COMPLETE c.files_changed = 4 for c in children: if c.status != ProcessingState.COMPLETE: c.status = ProcessingState.COMPLETE c.files_changed = 5 # Record changesets per project for proj_name in _PROJECTS: fc = 4 if "common-lib" in proj_name else 5 plan = mp_svc.record_changeset( plan, proj_name, ChangeSetSummary( project_name=proj_name, files_changed=fc, files_added=1, files_deleted=0, total_lines_changed=50, validation_passed=True, validation_errors=[], ), ) # Cross-project validation errors = mp_svc.validate_cross_project(plan) assert len(errors) == 0, f"Cross-project errors: {errors}" plan_svc.complete_execute(pid) # ── Apply in dependency order ── plan_svc.apply_plan(pid) plan_svc.start_apply(pid) for c in children: c.status = ProcessingState.APPLIED plan = plan_svc.complete_apply(pid) # ── Verify final state ── assert plan.phase == PlanPhase.APPLY assert plan.state == ProcessingState.APPLIED assert plan.is_terminal # Verify validations still attached to all projects for proj_name, info in _PROJECTS.items(): attached = tool_svc.list_validations_for_resource( resource_id=info["resource"], project_name=proj_name, ) assert len(attached) == 1, f"Validation missing for {proj_name} after lifecycle" print("full-lifecycle-ok") # --------------------------------------------------------------------------- # CLI dispatcher # --------------------------------------------------------------------------- _COMMANDS = { "register-projects": cmd_register_projects, "create-action": cmd_create_action, "plan-use": cmd_plan_use, "strategize-cycle": cmd_strategize_cycle, "spawn-children": cmd_spawn_children, "execute-children": cmd_execute_children, "apply-ordered": cmd_apply_ordered, "full-lifecycle": cmd_full_lifecycle, } def main() -> None: if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr) sys.exit(1) _COMMANDS[sys.argv[1]]() if __name__ == "__main__": main()