cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity
Files
master
cleveragents-core/robot/helper_wf04_multi_project_dependency.py
T
HAL9000 de1fd49594 fix(plan-lifecycle): fix integration test regressions from DoD gating 
Update integration test helpers to use definition_of_done values that
satisfy the new TextMatchEvaluator gate (criterion text must appear as
a substring of a plan argument key or value).  Also fix _evaluate_dod
to merge DoD metadata into plan.validation_summary without overwriting
Execute-phase validation counts, preserving the coverage/tool-validation
gate used by apply_with_validation_gate.  Update CONTRIBUTORS.md.

Fixes: wf02, wf04, wf06, wf07 integration test suites.

ISSUES CLOSED: #7927
2026-06-02 05:14:43 -04:00

764 lines
25 KiB
Python
Raw Permalink Blame History

"""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.COMPLETE
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="library_name",
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 ──
children = _make_child_statuses()
# ── 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()
Reference in New Issue View Git Blame Copy Permalink