freemo/cleveragents-core
0
0
Fork 0
forked from cleveragents/cleveragents-core
Code Pull Requests Activity

Merge pull request 'test(e2e): verify M6 success criteria — Firefox-scale autonomous porting' (#457) from test/m6-e2e-verification into master

Browse Source 
Reviewed-on: cleveragents/cleveragents-core#457
Reviewed-by: Luis Mendes <luis.mendes@cleverthis.com>
This commit is contained in:
Brent Edwards
2026-02-27 23:00:14 +00:00
committed by Forgejo
parent 7235d46ade ea060ec7a9
commit 332eb142b3
2 changed files with 1014 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
robot/helper_m6_e2e_verification.py
+891
View File
@@ -0,0 +1,891 @@
"""Robot Framework helper for M6 E2E verification tests.
Exercises the complete M6 success criteria sequence:
1. Porting action creation from YAML config
2. Plan use + execute via CLI with mocked lifecycle service
3. Hierarchical decomposition: 4+ levels of subplans
4. Decision correction recomputes only affected subtree
5. Parallel execution scales to 10+ concurrent subplans
6. Realistic porting task completes autonomously (lifecycle)
7. Plan apply transitions to APPLIED state
8. SubplanFailureHandler retry and stop-others logic
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_m6_e2e_verification.py <command>
"""
from __future__ import annotations
import sys
import tempfile
from collections.abc import Callable
from datetime import UTC, 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.cli.commands.action import app as action_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,
)
from cleveragents.domain.models.core.correction import ( # noqa: E402
CorrectionDryRunReport,
CorrectionImpact,
CorrectionMode,
CorrectionRequest,
CorrectionStatus,
)
from cleveragents.domain.models.core.decision import ( # noqa: E402
Decision,
DecisionType,
)
from cleveragents.domain.models.core.plan import ( # noqa: E402
ExecutionMode,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
ProcessingState,
SubplanConfig,
SubplanFailureHandler,
SubplanMergeStrategy,
SubplanStatus,
)
cli_runner = CliRunner()
# -------------------------------------------------------------------
# Valid ULID constants
# -------------------------------------------------------------------
_ROOT_PLAN_ID = str(ULID())
_CHILD_PLAN_IDS = [str(ULID()) for _ in range(4)]
_GRANDCHILD_PLAN_IDS = [str(ULID()) for _ in range(8)]
_GREAT_GRANDCHILD_IDS = [str(ULID()) for _ in range(4)]
_LEAF_IDS = [str(ULID()) for _ in range(4)]
_ACTION_ULID = str(ULID())
_ROOT_DEC_ID = str(ULID())
_CHILD_DEC_IDS = [str(ULID()) for _ in range(4)]
# -------------------------------------------------------------------
# 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 _make_plan(
*,
plan_id: str = _ROOT_PLAN_ID,
parent_plan_id: str | None = None,
root_plan_id: str | None = None,
phase: PlanPhase = PlanPhase.STRATEGIZE,
processing_state: ProcessingState = ProcessingState.QUEUED,
subplan_config: SubplanConfig | None = None,
subplan_statuses: list[SubplanStatus] | None = None,
action_name: str = "local/port-to-typescript",
description: str = "Port Python to TypeScript",
) -> Plan:
"""Create a Plan for testing with minimal required fields."""
return Plan(
identity=PlanIdentity(
plan_id=plan_id,
parent_plan_id=parent_plan_id,
root_plan_id=root_plan_id,
),
namespaced_name=NamespacedName(
server=None,
namespace="local",
name="port-to-typescript",
),
description=description,
action_name=action_name,
phase=phase,
processing_state=processing_state,
subplan_config=subplan_config,
subplan_statuses=subplan_statuses or [],
)
def _make_action() -> Action:
"""Create a porting action for testing."""
return Action.from_config(
{
"name": "local/port-to-typescript",
"description": "Autonomous Python to TypeScript porting action",
"long_description": "Converts all Python files to TypeScript with tests",
"strategy_actor": "local/architect",
"execution_actor": "local/coder",
"definition_of_done": (
"All Python files converted to TypeScript with tests"
),
"reusable": True,
"read_only": False,
"created_by": "robot-m6-test",
}
)
# -------------------------------------------------------------------
# Subcommand: action-create-porting
# -------------------------------------------------------------------
def action_create_porting() -> None:
"""Create a porting action from YAML config via CLI.
Writes a temp YAML file, invokes ``agents action create --config``,
and verifies the action is created with the expected attributes.
"""
action = _make_action()
mock_service = MagicMock()
mock_service.create_action.return_value = action
with patch(
"cleveragents.cli.commands.action._get_lifecycle_service",
return_value=mock_service,
):
# Write YAML config to temp file
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".yaml",
delete=False,
) as f:
f.write(
"name: local/port-to-typescript\n"
"description: Autonomous Python to TypeScript porting action\n"
"long_description: Converts all Python files to TypeScript\n"
"strategy_actor: local/architect\n"
"execution_actor: local/coder\n"
"definition_of_done: >-\n"
" All Python files converted to TypeScript with tests\n"
"reusable: true\n"
"read_only: false\n"
)
yaml_path = f.name
result = cli_runner.invoke(action_app, ["create", "--config", yaml_path])
if result.exit_code != 0:
_fail(
f"action create exit_code={result.exit_code}: "
f"{result.output}\n{result.exception}"
)
# Verify service was called
mock_service.create_action.assert_called_once()
call_kwargs = mock_service.create_action.call_args
if call_kwargs is None:
_fail("create_action was not called with keyword args")
# Clean up
Path(yaml_path).unlink(missing_ok=True)
print("m6-action-create-porting-ok")
# -------------------------------------------------------------------
# Subcommand: plan-use-execute
# -------------------------------------------------------------------
def plan_use_execute() -> None:
"""Use and execute a porting plan via CLI.
Mocks _get_lifecycle_service to verify the plan transitions
through use → execute.
"""
action = _make_action()
plan_use = _make_plan(
phase=PlanPhase.STRATEGIZE,
processing_state=ProcessingState.QUEUED,
)
plan_exec = _make_plan(
phase=PlanPhase.EXECUTE,
processing_state=ProcessingState.PROCESSING,
)
mock_service = MagicMock()
mock_service.get_action_by_name.return_value = action
mock_service.use_action.return_value = plan_use
mock_service.execute_plan.return_value = plan_exec
# For auto-discovery in execute
mock_service.list_plans.return_value = [plan_use]
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
# Plan use
result_use = cli_runner.invoke(
plan_app,
[
"use",
"local/port-to-typescript",
"local/large-project",
],
)
if result_use.exit_code != 0:
_fail(
f"plan use exit_code={result_use.exit_code}: "
f"{result_use.output}\n{result_use.exception}"
)
mock_service.use_action.assert_called_once()
# Plan execute
result_exec = cli_runner.invoke(
plan_app,
["execute", _ROOT_PLAN_ID],
)
if result_exec.exit_code != 0:
_fail(
f"plan execute exit_code={result_exec.exit_code}: "
f"{result_exec.output}\n{result_exec.exception}"
)
mock_service.execute_plan.assert_called_once()
print("m6-plan-use-execute-ok")
# -------------------------------------------------------------------
# Subcommand: hierarchical-decomposition
# -------------------------------------------------------------------
def hierarchical_decomposition() -> None:
"""Verify hierarchical decomposition creates 4+ levels of subplans.
Constructs a tree: root → L1 (4 children) → L2 (8 grandchildren)
→ L3 (4 great-grandchildren) → L4 (4 leaves).
Verifies identity flags and root_plan_id propagation.
"""
# Level 0: root plan
root = _make_plan(plan_id=_ROOT_PLAN_ID, root_plan_id=_ROOT_PLAN_ID)
if not root.is_root_plan:
_fail("root should be is_root_plan")
if root.is_subplan:
_fail("root should not be is_subplan")
# Level 1: 4 children
level1 = []
for cid in _CHILD_PLAN_IDS:
p = _make_plan(
plan_id=cid,
parent_plan_id=_ROOT_PLAN_ID,
root_plan_id=_ROOT_PLAN_ID,
description=f"L1 subplan {cid[:8]}",
)
if not p.is_subplan:
_fail(f"L1 {cid} should be subplan")
if p.identity.root_plan_id != _ROOT_PLAN_ID:
_fail(f"L1 root_plan_id mismatch: {p.identity.root_plan_id}")
level1.append(p)
if len(level1) < 4:
_fail(f"expected >= 4 L1 subplans, got {len(level1)}")
# Level 2: 8 grandchildren (2 per L1)
level2 = []
for i, gcid in enumerate(_GRANDCHILD_PLAN_IDS):
parent = _CHILD_PLAN_IDS[i % len(_CHILD_PLAN_IDS)]
p = _make_plan(
plan_id=gcid,
parent_plan_id=parent,
root_plan_id=_ROOT_PLAN_ID,
description=f"L2 subplan {gcid[:8]}",
)
if p.identity.root_plan_id != _ROOT_PLAN_ID:
_fail(f"L2 root mismatch: {p.identity.root_plan_id}")
level2.append(p)
if len(level2) < 4:
_fail(f"expected >= 4 L2 subplans, got {len(level2)}")
# Level 3: 4 great-grandchildren
level3 = []
for i, ggid in enumerate(_GREAT_GRANDCHILD_IDS):
parent = _GRANDCHILD_PLAN_IDS[i % len(_GRANDCHILD_PLAN_IDS)]
p = _make_plan(
plan_id=ggid,
parent_plan_id=parent,
root_plan_id=_ROOT_PLAN_ID,
description=f"L3 subplan {ggid[:8]}",
)
level3.append(p)
# Level 4: 4 leaves
level4 = []
for i, lid in enumerate(_LEAF_IDS):
parent = _GREAT_GRANDCHILD_IDS[i % len(_GREAT_GRANDCHILD_IDS)]
p = _make_plan(
plan_id=lid,
parent_plan_id=parent,
root_plan_id=_ROOT_PLAN_ID,
description=f"L4 leaf subplan {lid[:8]}",
)
level4.append(p)
total = 1 + len(level1) + len(level2) + len(level3) + len(level4)
levels = 5 # L0, L1, L2, L3, L4
if levels < 5:
_fail(f"expected >= 5 levels (4+ subplan levels), got {levels}")
# Verify all non-root plans propagate root_plan_id
for plan in [*level1, *level2, *level3, *level4]:
if plan.identity.root_plan_id != _ROOT_PLAN_ID:
_fail(f"root_plan_id not propagated: {plan.identity.plan_id}")
if not plan.is_subplan:
_fail(f"should be subplan: {plan.identity.plan_id}")
if total < 21:
_fail(f"expected >= 21 total plans, got {total}")
print("m6-hierarchical-decomposition-ok")
# -------------------------------------------------------------------
# Subcommand: correction-affected-subtree
# -------------------------------------------------------------------
def correction_affected_subtree() -> None:
"""Verify decision correction recomputes only affected subtree.
Creates a decision tree (root → 4 children), then simulates
correction targeting one child. Verifies CorrectionImpact
contains only the affected subtree, not the entire tree.
"""
# Root decision
root = Decision(
decision_id=_ROOT_DEC_ID,
plan_id=_ROOT_PLAN_ID,
parent_decision_id=None,
sequence_number=0,
decision_type=DecisionType.PROMPT_DEFINITION,
question="How to port this project?",
chosen_option="Module-by-module porting strategy",
rationale="Systematic approach for large codebase",
downstream_decision_ids=_CHILD_DEC_IDS,
)
if root.parent_decision_id is not None:
_fail("root decision should have no parent")
if len(root.downstream_decision_ids) != len(_CHILD_DEC_IDS):
_fail(f"root downstream count: {len(root.downstream_decision_ids)}")
# 4 child decisions (one per module)
children = []
for i, cid in enumerate(_CHILD_DEC_IDS):
d = Decision(
decision_id=cid,
plan_id=_ROOT_PLAN_ID,
parent_decision_id=_ROOT_DEC_ID,
sequence_number=i + 1,
decision_type=DecisionType.SUBPLAN_SPAWN,
question=f"How to port module {i}?",
chosen_option=f"Direct translation for module {i}",
rationale=f"Module {i} has 1:1 Python→TS mapping",
)
children.append(d)
# Target child[1] for correction
target_id = _CHILD_DEC_IDS[1]
# Build CorrectionImpact scoped to just the target subtree
impact = CorrectionImpact(
affected_decisions=[target_id],
affected_files=[f"src/module_1/file_{j}.py" for j in range(5)],
affected_child_plans=[],
risk_level="low",
rollback_tier="phase",
artifacts_to_archive=[],
)
# Verify only 1 decision affected (not the root or other children)
if len(impact.affected_decisions) != 1:
_fail(f"expected 1 affected decision, got {len(impact.affected_decisions)}")
if impact.affected_decisions[0] != target_id:
_fail(f"wrong affected decision: {impact.affected_decisions[0]}")
# Verify root is NOT affected
if _ROOT_DEC_ID in impact.affected_decisions:
_fail("root should not be in affected decisions")
# Verify other children not affected
for i, cid in enumerate(_CHILD_DEC_IDS):
if i == 1:
continue
if cid in impact.affected_decisions:
_fail(f"child {i} should not be affected")
# Verify rollback_tier supports subtree-only
if impact.rollback_tier not in ("full", "phase", "append_only"):
_fail(f"unexpected rollback_tier: {impact.rollback_tier}")
# Build dry-run report
report = CorrectionDryRunReport(
correction_id=str(ULID()),
mode=CorrectionMode.REVERT,
impact=impact,
decisions_to_invalidate=[target_id],
child_plans_to_rollback=[],
)
if len(report.decisions_to_invalidate) != 1:
_fail(
f"expected 1 decision to invalidate, "
f"got {len(report.decisions_to_invalidate)}"
)
# Verify CorrectionRequest
request = CorrectionRequest(
plan_id=_ROOT_PLAN_ID,
target_decision_id=target_id,
mode=CorrectionMode.REVERT,
guidance="Re-evaluate module 1 porting strategy",
dry_run=True,
)
if request.status != CorrectionStatus.PENDING:
_fail(f"initial status should be PENDING: {request.status}")
print("m6-correction-affected-subtree-ok")
# -------------------------------------------------------------------
# Subcommand: parallel-execution-scale
# -------------------------------------------------------------------
def parallel_execution_scale() -> None:
"""Verify parallel execution scales to 10+ concurrent subplans.
Constructs a SubplanConfig with PARALLEL mode and max_parallel=15,
creates 15 SubplanStatuses, and verifies the plan model handles
them correctly.
"""
config = SubplanConfig(
execution_mode=ExecutionMode.PARALLEL,
merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY,
max_parallel=15,
fail_fast=False,
retry_failed=True,
max_retries=2,
)
if config.execution_mode != ExecutionMode.PARALLEL:
_fail(f"execution_mode mismatch: {config.execution_mode}")
if config.max_parallel != 15:
_fail(f"max_parallel mismatch: {config.max_parallel}")
# Create 15 subplan statuses (simulating 15 concurrent subplans)
statuses: list[SubplanStatus] = []
for i in range(15):
st = SubplanStatus(
subplan_id=str(ULID()),
action_name="local/port-module",
target_resources=[f"src/module_{i:02d}/"],
status=ProcessingState.PROCESSING,
)
statuses.append(st)
if len(statuses) < 10:
_fail(f"expected >= 10 concurrent subplans, got {len(statuses)}")
# Create plan with parallel config
plan = _make_plan(
subplan_config=config,
subplan_statuses=statuses,
phase=PlanPhase.EXECUTE,
processing_state=ProcessingState.PROCESSING,
)
if not plan.has_subplans:
_fail("plan should have subplans")
if len(plan.subplan_statuses) != 15:
_fail(f"expected 15 subplans, got {len(plan.subplan_statuses)}")
if plan.subplan_config is None:
_fail("subplan_config is None")
if plan.subplan_config.max_parallel < 10:
_fail(f"max_parallel should be >= 10: {plan.subplan_config.max_parallel}")
# Verify CLI dict includes subplan info
cli = plan.as_cli_dict()
if "subplan_count" not in cli:
_fail("as_cli_dict missing subplan_count")
if cli["subplan_count"] != 15:
_fail(f"subplan_count mismatch: {cli['subplan_count']}")
print("m6-parallel-execution-scale-ok")
# -------------------------------------------------------------------
# Subcommand: porting-task-autonomous
# -------------------------------------------------------------------
def porting_task_autonomous() -> None:
"""Verify a realistic porting task completes autonomously.
Simulates the full lifecycle: ACTION → STRATEGIZE → EXECUTE →
APPLY with subplan decomposition and completion.
"""
# Phase 1: ACTION → STRATEGIZE
plan_action = _make_plan(
phase=PlanPhase.ACTION,
processing_state=ProcessingState.COMPLETE,
)
if not plan_action.can_transition_to_next_phase:
_fail("ACTION/COMPLETE should allow transition")
# Phase 2: STRATEGIZE with subplan config
plan_strat = _make_plan(
phase=PlanPhase.STRATEGIZE,
processing_state=ProcessingState.COMPLETE,
subplan_config=SubplanConfig(
execution_mode=ExecutionMode.PARALLEL,
max_parallel=10,
),
)
if not plan_strat.can_transition_to_next_phase:
_fail("STRATEGIZE/COMPLETE should allow transition")
# Phase 3: EXECUTE with 10 subplans all COMPLETE
exec_statuses = []
for i in range(10):
exec_statuses.append(
SubplanStatus(
subplan_id=str(ULID()),
action_name="local/port-module",
target_resources=[f"src/module_{i}/"],
status=ProcessingState.COMPLETE,
started_at=datetime.now(tz=UTC),
completed_at=datetime.now(tz=UTC),
files_changed=i * 10 + 5,
)
)
plan_exec = _make_plan(
phase=PlanPhase.EXECUTE,
processing_state=ProcessingState.COMPLETE,
subplan_config=SubplanConfig(
execution_mode=ExecutionMode.PARALLEL,
max_parallel=10,
),
subplan_statuses=exec_statuses,
)
if not plan_exec.can_transition_to_next_phase:
_fail("EXECUTE/COMPLETE should allow transition")
if not plan_exec.has_subplans:
_fail("execute plan should have subplans")
# Verify all subplans completed
for st in plan_exec.subplan_statuses:
if st.status != ProcessingState.COMPLETE:
_fail(f"subplan {st.subplan_id} not complete: {st.status}")
# Phase 4: APPLY
plan_apply = _make_plan(
phase=PlanPhase.APPLY,
processing_state=ProcessingState.APPLIED,
subplan_config=plan_exec.subplan_config,
subplan_statuses=exec_statuses,
)
if not plan_apply.is_terminal:
_fail("APPLIED should be terminal")
# Verify total files changed across all subplans
total_files = sum(st.files_changed for st in plan_apply.subplan_statuses)
if total_files < 50:
_fail(f"expected >= 50 files changed, got {total_files}")
print("m6-porting-task-autonomous-ok")
# -------------------------------------------------------------------
# Subcommand: plan-apply-lifecycle
# -------------------------------------------------------------------
def plan_apply_lifecycle() -> None:
"""Verify plan apply via CLI transitions to APPLIED state."""
plan_applied = _make_plan(
phase=PlanPhase.APPLY,
processing_state=ProcessingState.APPLIED,
)
mock_service = MagicMock()
mock_service.apply_plan.return_value = plan_applied
# For auto-discovery
plan_exec_complete = _make_plan(
phase=PlanPhase.EXECUTE,
processing_state=ProcessingState.COMPLETE,
)
mock_service.list_plans.return_value = [plan_exec_complete]
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = cli_runner.invoke(
plan_app,
["lifecycle-apply", _ROOT_PLAN_ID],
)
if result.exit_code != 0:
_fail(
f"lifecycle-apply exit_code={result.exit_code}: "
f"{result.output}\n{result.exception}"
)
mock_service.apply_plan.assert_called_once()
print("m6-plan-apply-lifecycle-ok")
# -------------------------------------------------------------------
# Subcommand: failure-handler-logic
# -------------------------------------------------------------------
def failure_handler_logic() -> None:
"""Verify SubplanFailureHandler retry and stop-others logic.
Tests:
- fail_fast=True stops others
- SEQUENTIAL mode stops others
- Retriable errors are retried
- Non-retriable errors are not retried
- Max retries exceeded stops retry
"""
handler = SubplanFailureHandler()
# fail_fast stops others
config_ff = SubplanConfig(
execution_mode=ExecutionMode.PARALLEL,
fail_fast=True,
)
errored = SubplanStatus(
subplan_id=str(ULID()),
action_name="local/sub",
status=ProcessingState.ERRORED,
error="TimeoutError: timed out",
)
if not handler.should_stop_others(config_ff, errored):
_fail("fail_fast=True should stop others")
# SEQUENTIAL stops others regardless of fail_fast
config_seq = SubplanConfig(
execution_mode=ExecutionMode.SEQUENTIAL,
fail_fast=False,
)
if not handler.should_stop_others(config_seq, errored):
_fail("SEQUENTIAL should stop others")
# Retriable error should retry
config_retry = SubplanConfig(retry_failed=True, max_retries=2)
retriable = SubplanStatus(
subplan_id=str(ULID()),
action_name="local/sub",
status=ProcessingState.ERRORED,
error="TimeoutError: timed out",
attempt_number=1,
)
if not handler.should_retry(config_retry, retriable):
_fail("retriable error should be retried")
# Non-retriable error should NOT retry
non_retriable = SubplanStatus(
subplan_id=str(ULID()),
action_name="local/sub",
status=ProcessingState.ERRORED,
error="ConfigurationError: bad config",
attempt_number=1,
)
if handler.should_retry(config_retry, non_retriable):
_fail("non-retriable error should not be retried")
# Max retries exceeded
exhausted = SubplanStatus(
subplan_id=str(ULID()),
action_name="local/sub",
status=ProcessingState.ERRORED,
error="TimeoutError: timed out",
attempt_number=3,
)
if handler.should_retry(config_retry, exhausted):
_fail("max retries exceeded should not retry")
print("m6-failure-handler-logic-ok")
# -------------------------------------------------------------------
# Subcommand: subplan-config-modes
# -------------------------------------------------------------------
def subplan_config_modes() -> None:
"""Verify SubplanConfig supports all execution modes and merge strategies."""
# All execution modes
for mode in ExecutionMode:
cfg = SubplanConfig(execution_mode=mode)
if cfg.execution_mode != mode:
_fail(f"execution_mode mismatch: {cfg.execution_mode} != {mode}")
# All merge strategies
for strat in SubplanMergeStrategy:
cfg = SubplanConfig(merge_strategy=strat)
if cfg.merge_strategy != strat:
_fail(f"merge_strategy mismatch: {cfg.merge_strategy} != {strat}")
# max_parallel bounds
cfg_min = SubplanConfig(max_parallel=1)
if cfg_min.max_parallel != 1:
_fail(f"max_parallel=1 mismatch: {cfg_min.max_parallel}")
cfg_max = SubplanConfig(max_parallel=50)
if cfg_max.max_parallel != 50:
_fail(f"max_parallel=50 mismatch: {cfg_max.max_parallel}")
# Timeout
cfg_timeout = SubplanConfig(timeout_per_subplan_seconds=600)
if cfg_timeout.timeout_per_subplan_seconds != 600:
_fail(f"timeout mismatch: {cfg_timeout.timeout_per_subplan_seconds}")
# Default values
default = SubplanConfig()
if default.execution_mode != ExecutionMode.SEQUENTIAL:
_fail(f"default mode: {default.execution_mode}")
if default.merge_strategy != SubplanMergeStrategy.GIT_THREE_WAY:
_fail(f"default merge: {default.merge_strategy}")
if default.max_parallel != 5:
_fail(f"default max_parallel: {default.max_parallel}")
print("m6-subplan-config-modes-ok")
# -------------------------------------------------------------------
# Subcommand: decision-tree-porting
# -------------------------------------------------------------------
def decision_tree_porting() -> None:
"""Verify decision tree structure for a porting task.
Creates a decision tree with PROMPT_DEFINITION root,
SUBPLAN_PARALLEL_SPAWN decisions for module decomposition,
and verifies parent-child relationships.
"""
root = Decision(
decision_id=_ROOT_DEC_ID,
plan_id=_ROOT_PLAN_ID,
parent_decision_id=None,
sequence_number=0,
decision_type=DecisionType.PROMPT_DEFINITION,
question="Port Python project to TypeScript",
chosen_option="Hierarchical module-by-module porting",
rationale="Break down into parallel workstreams",
downstream_decision_ids=_CHILD_DEC_IDS,
)
# PROMPT_DEFINITION must be root (no parent)
if root.parent_decision_id is not None:
_fail("PROMPT_DEFINITION should have no parent")
# Spawn decisions for each module
spawn_decisions = []
for i, cid in enumerate(_CHILD_DEC_IDS):
d = Decision(
decision_id=cid,
plan_id=_ROOT_PLAN_ID,
parent_decision_id=_ROOT_DEC_ID,
sequence_number=i + 1,
decision_type=DecisionType.SUBPLAN_PARALLEL_SPAWN,
question=f"How to port module_{i:02d}?",
chosen_option=f"Create subplan for module_{i:02d}",
rationale=f"Module {i} is independent",
downstream_plan_ids=[_CHILD_PLAN_IDS[i]],
)
spawn_decisions.append(d)
# Verify parent-child links
for d in spawn_decisions:
if d.parent_decision_id != _ROOT_DEC_ID:
_fail(f"child {d.decision_id} parent mismatch")
if d.decision_type != DecisionType.SUBPLAN_PARALLEL_SPAWN:
_fail(f"child type should be SUBPLAN_PARALLEL_SPAWN: {d.decision_type}")
if len(d.downstream_plan_ids) != 1:
_fail(f"child should spawn 1 plan: {len(d.downstream_plan_ids)}")
# Root should reference all children
if len(root.downstream_decision_ids) != len(_CHILD_DEC_IDS):
_fail(
f"root downstream count: "
f"{len(root.downstream_decision_ids)} != {len(_CHILD_DEC_IDS)}"
)
# Verify superseded_by for correction flow
corrected_root = root.with_superseded_by(str(ULID()))
if corrected_root.superseded_by is None:
_fail("superseded_by should be set")
# Original should be untouched (frozen model)
if root.superseded_by is not None:
_fail("original root superseded_by should still be None")
print("m6-decision-tree-porting-ok")
# -------------------------------------------------------------------
# Dispatcher
# -------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"action-create-porting": action_create_porting,
"plan-use-execute": plan_use_execute,
"hierarchical-decomposition": hierarchical_decomposition,
"correction-affected-subtree": correction_affected_subtree,
"parallel-execution-scale": parallel_execution_scale,
"porting-task-autonomous": porting_task_autonomous,
"plan-apply-lifecycle": plan_apply_lifecycle,
"failure-handler-logic": failure_handler_logic,
"subplan-config-modes": subplan_config_modes,
"decision-tree-porting": decision_tree_porting,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(
f"Usage: helper_m6_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())
robot/m6_e2e_verification.robot
+123
View File
@@ -0,0 +1,123 @@
*** Settings ***
Documentation End-to-end verification of M6 success criteria:
... Firefox-scale autonomous porting via action creation,
... plan use/execute/apply, hierarchical decomposition
... (4+ levels of subplans), decision correction on
... affected subtree only, parallel execution scaling
... to 10+ concurrent subplans, and full lifecycle
... completion of a realistic porting task.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_m6_e2e_verification.py
*** Test Cases ***
Porting Action Created From YAML Config
[Documentation] Create a porting action from YAML config via the
... ``agents action create --config`` CLI command.
... Verifies the lifecycle service is called with the
... expected action attributes.
${result}= Run Process ${PYTHON} ${HELPER} action-create-porting cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-action-create-porting-ok
Plan Use And Execute On Large Project
[Documentation] Use a porting action on a large project and execute
... the plan via CLI. Verifies both ``plan use`` and
... ``plan execute`` commands invoke the lifecycle service.
${result}= Run Process ${PYTHON} ${HELPER} plan-use-execute cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-plan-use-execute-ok
Hierarchical Decomposition Creates Four Plus Levels
[Documentation] Verify hierarchical decomposition creates 4+ levels
... of subplans: root -> L1 (4 children) -> L2 (8) ->
... L3 (4) -> L4 (4 leaves). Verifies identity flags
... and root_plan_id propagation through all levels.
${result}= Run Process ${PYTHON} ${HELPER} hierarchical-decomposition cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-hierarchical-decomposition-ok
Decision Correction Recomputes Only Affected Subtree
[Documentation] Verify that decision correction targets only the
... affected subtree. Creates a decision tree, targets
... one child for correction, and verifies the
... CorrectionImpact excludes the root and sibling
... decisions. Tests rollback_tier and dry-run report.
${result}= Run Process ${PYTHON} ${HELPER} correction-affected-subtree cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-correction-affected-subtree-ok
Parallel Execution Scales To Ten Plus Concurrent Subplans
[Documentation] Verify parallel execution with SubplanConfig
... in PARALLEL mode and max_parallel=15. Creates 15
... concurrent SubplanStatuses and verifies the plan
... model handles them correctly.
${result}= Run Process ${PYTHON} ${HELPER} parallel-execution-scale cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-parallel-execution-scale-ok
Realistic Porting Task Completes Autonomously
[Documentation] Simulate the full autonomous porting lifecycle:
... ACTION -> STRATEGIZE -> EXECUTE -> APPLY with
... 10 subplans, verifying phase transitions,
... subplan completion, and total files changed.
${result}= Run Process ${PYTHON} ${HELPER} porting-task-autonomous cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-porting-task-autonomous-ok
Plan Apply Via Lifecycle CLI
[Documentation] Verify ``agents plan lifecycle-apply`` transitions
... the plan to APPLIED state via the lifecycle service.
${result}= Run Process ${PYTHON} ${HELPER} plan-apply-lifecycle cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-plan-apply-lifecycle-ok
Subplan Failure Handler Retry And Stop Logic
[Documentation] Verify SubplanFailureHandler: fail_fast stops others,
... SEQUENTIAL mode stops others, retriable errors are
... retried, non-retriable errors are skipped, and max
... retries exceeded prevents retry.
${result}= Run Process ${PYTHON} ${HELPER} failure-handler-logic cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-failure-handler-logic-ok
Subplan Config Supports All Execution Modes
[Documentation] Verify SubplanConfig accepts all execution modes
... (SEQUENTIAL, PARALLEL, DEPENDENCY_ORDERED) and all
... merge strategies. Tests bounds for max_parallel and
... timeout configuration.
${result}= Run Process ${PYTHON} ${HELPER} subplan-config-modes cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-subplan-config-modes-ok
Decision Tree Structure For Porting Task
[Documentation] Verify decision tree with PROMPT_DEFINITION root and
... SUBPLAN_PARALLEL_SPAWN children. Tests parent-child
... links, downstream_plan_ids, and superseded_by for
... the correction flow.
${result}= Run Process ${PYTHON} ${HELPER} decision-tree-porting cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} m6-decision-tree-porting-ok