feat(cli): align plan use/list/status flags
CI / coverage (push) Blocked by required conditions
CI / build (push) Waiting to run
CI / lint (push) Waiting to run
CI / typecheck (push) Waiting to run
CI / docker (push) Blocked by required conditions
CI / security (push) Waiting to run
CI / quality (push) Waiting to run
CI / unit_tests (push) Waiting to run
CI / integration_tests (push) Waiting to run
CI / lint (pull_request) Successful in 16s
CI / typecheck (pull_request) Successful in 26s
CI / security (pull_request) Successful in 22s
CI / quality (pull_request) Successful in 15s
CI / integration_tests (pull_request) Successful in 4m31s
CI / build (pull_request) Successful in 16s
CI / unit_tests (pull_request) Successful in 9m18s
CI / coverage (pull_request) Successful in 6m54s
CI / docker (pull_request) Successful in 39s

- plan use: accept positional PROJECT args, add --automation-profile,
  --invariant (repeatable), and actor override flags
- lifecycle-list: add --state/--processing-state, --action filters,
  optional positional REGEX for name filtering, Action column in output
- plan status: enhanced detail view with processing state, project
  links, arguments order, automation profile provenance, actor
  overrides, automation level, and full timestamps
- Add docs/reference/plan_cli.md reference documentation
- Add Behave feature (17 scenarios) with mock service steps
- Add Robot smoke test (7 cases) with import-safe helper
- Add ASV benchmarks for plan use/list/status commands
This commit was merged in pull request #62.
This commit is contained in:
2026-02-14 14:21:00 +00:00
parent d98666651f
commit 65988669a6
8 changed files with 1549 additions and 45 deletions
+198
View File
@@ -0,0 +1,198 @@
"""ASV benchmarks for Plan CLI command throughput.
Measures the performance of:
- Plan use with multiple projects and flags
- Lifecycle-list with various filters
- Plan status rendering
- Plan cancel with reason
"""
from __future__ import annotations
import importlib
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from typer.testing import CliRunner # noqa: E402
from cleveragents.cli.commands.plan import app as plan_app # noqa: E402
from cleveragents.domain.models.core.action import Action, ActionState # noqa: E402
from cleveragents.domain.models.core.plan import ( # noqa: E402
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
_runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"
def _mock_action(name: str = "local/bench-action") -> Action:
return Action(
namespaced_name=NamespacedName.parse(name),
description="Benchmark action",
long_description=None,
definition_of_done="Benchmarks pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
state=ActionState.AVAILABLE,
created_by=None,
created_at=datetime.now(),
updated_at=datetime.now(),
)
def _mock_plan(
name: str = "local/bench-plan",
project_links: list[ProjectLink] | None = None,
) -> Plan:
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=_PLAN_ULID),
namespaced_name=NamespacedName.parse(name),
description="Benchmark plan",
definition_of_done="Benchmarks pass",
action_name="local/bench-action",
phase=PlanPhase.STRATEGIZE,
processing_state=ProcessingState.QUEUED,
automation_level=AutomationLevel.MANUAL,
project_links=project_links or [],
arguments={"target_coverage": 80},
arguments_order=["target_coverage"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
estimation_actor="openai/gpt-4",
invariant_actor="openai/gpt-4",
reusable=True,
read_only=False,
created_by=None,
timestamps=PlanTimestamps(created_at=now, updated_at=now),
)
class PlanCLIUseSuite:
"""Benchmark plan use command throughput."""
def setup(self) -> None:
"""Set up mock service."""
self._mock_service = MagicMock()
self._mock_service.get_action_by_name.return_value = _mock_action()
self._mock_service.use_action.return_value = _mock_plan(
project_links=[
ProjectLink(project_name="proj-a"),
ProjectLink(project_name="proj-b"),
]
)
self._patcher = patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=self._mock_service,
)
self._patcher.start()
def teardown(self) -> None:
self._patcher.stop()
def time_use_with_projects(self) -> None:
"""Benchmark plan use with multiple projects."""
_runner.invoke(
plan_app,
["use", "local/bench-action", "proj-a", "proj-b", "--arg", "coverage=80"],
)
def time_use_with_all_flags(self) -> None:
"""Benchmark plan use with all flags."""
_runner.invoke(
plan_app,
[
"use",
"local/bench-action",
"proj-a",
"--automation-profile",
"trusted",
"--invariant",
"No warnings",
"--strategy-actor",
"openai/gpt-4",
"--execution-actor",
"openai/gpt-4",
"--arg",
"coverage=80",
],
)
class PlanCLIListSuite:
"""Benchmark plan lifecycle-list throughput."""
def setup(self) -> None:
self._mock_service = MagicMock()
self._mock_service.list_plans.return_value = [
_mock_plan(f"local/plan-{i}") for i in range(50)
]
self._patcher = patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=self._mock_service,
)
self._patcher.start()
def teardown(self) -> None:
self._patcher.stop()
def time_list_all(self) -> None:
"""Benchmark listing all plans."""
_runner.invoke(plan_app, ["lifecycle-list"])
def time_list_with_phase(self) -> None:
"""Benchmark listing with phase filter."""
_runner.invoke(plan_app, ["lifecycle-list", "--phase", "strategize"])
def time_list_with_state(self) -> None:
"""Benchmark listing with state filter."""
_runner.invoke(plan_app, ["lifecycle-list", "--state", "queued"])
class PlanCLIStatusSuite:
"""Benchmark plan status rendering."""
def setup(self) -> None:
self._mock_service = MagicMock()
self._mock_service.get_plan.return_value = _mock_plan(
project_links=[ProjectLink(project_name="proj-a", alias="api")]
)
self._patcher = patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=self._mock_service,
)
self._patcher.start()
def teardown(self) -> None:
self._patcher.stop()
def time_status(self) -> None:
"""Benchmark rendering plan status."""
_runner.invoke(plan_app, ["status", _PLAN_ULID])
+144
View File
@@ -0,0 +1,144 @@
# Plan CLI Reference
The `agents plan` command group manages plans in the CleverAgents v3
plan lifecycle.
## Plan Use
```bash
agents plan use <ACTION> [PROJECT ...] [OPTIONS]
```
Create a plan from an action template. The first positional argument is
the **action name** (namespaced, e.g. `local/code-coverage`). All
subsequent positional arguments are **project names** the plan will
operate on. Projects can also be supplied via the repeatable
`--project` / `-p` option.
### Options
| Flag | Short | Description |
|-------------------------|-------|----------------------------------------------------------|
| `--project` | `-p` | Project name (repeatable) |
| `--arg` | `-a` | Argument value `name=value` (repeatable) |
| `--automation-profile` | | Automation profile name for this plan |
| `--invariant` | | Invariant constraint text (repeatable) |
| `--strategy-actor` | | Override the strategy actor |
| `--execution-actor` | | Override the execution actor |
| `--estimation-actor` | | Override the estimation actor |
| `--invariant-actor` | | Override the invariant reconciliation actor |
| `--automation-level` | | `manual`, `review_before_apply`, or `full_automation` |
### Examples
```bash
# Single project via positional arg
agents plan use local/code-coverage my-project --arg target_coverage=80
# Multiple projects via positional args
agents plan use local/lint proj-a proj-b
# Projects via --project option
agents plan use local/lint --project proj-a --project proj-b
# With invariants and actor overrides
agents plan use local/refactor proj-1 \
--invariant "No new warnings" \
--invariant "Keep backward compat" \
--strategy-actor openai/gpt-4 \
--execution-actor anthropic/claude-3
# With automation profile
agents plan use local/code-coverage proj-1 --automation-profile trusted
```
## Plan List (lifecycle-list)
```bash
agents plan lifecycle-list [REGEX] [OPTIONS]
```
### Options
| Flag | Short | Description |
|----------------------|-------|-----------------------------------------------------------|
| `--phase` | | Filter by phase (`strategize`, `execute`, `apply`, `applied`) |
| `--state` | | Filter by processing state (`queued`, `processing`, `errored`, `complete`, `cancelled`) |
| `--processing-state` | | Alias for `--state` |
| `--project` | `-p` | Filter by project name |
| `--action` | | Filter by action name |
### Positional Arguments
| Argument | Description |
|----------|------------------------------------------|
| `REGEX` | Optional regex pattern to filter names |
### Examples
```bash
# List all plans
agents plan lifecycle-list
# Filter by phase
agents plan lifecycle-list --phase strategize
# Filter by processing state
agents plan lifecycle-list --state complete
# Combine filters
agents plan lifecycle-list --phase execute --project my-project --action local/lint
```
### Output Columns
| Column | Description |
|----------|--------------------------------------|
| ID | Truncated plan ULID |
| Name | Full namespaced plan name |
| Action | Source action name |
| Phase | Current lifecycle phase |
| State | Processing state |
| Projects | Linked project names |
| Created | Creation timestamp |
## Plan Status
```bash
agents plan status [PLAN_ID]
```
Without a plan ID, shows a summary table of all active plans.
With a plan ID, shows full details including:
- **Action**: source action name
- **Phase** and **Processing State**
- **Project links** with aliases and read-only flags
- **Arguments** (ordered)
- **Automation profile** with provenance tag
- **Actor overrides** (strategy, execution, estimation, invariant)
- **Timestamps** (created, updated, phase start/complete times)
## Plan Cancel
```bash
agents plan cancel <PLAN_ID> [OPTIONS]
```
### Options
| Flag | Short | Description |
|------------|-------|----------------------------|
| `--reason` | `-r` | Reason for cancellation |
### Example
```bash
agents plan cancel 01HXYZ... --reason "Requirements changed"
```
## Source Location
- CLI commands: `src/cleveragents/cli/commands/plan.py`
- Lifecycle service: `src/cleveragents/application/services/plan_lifecycle_service.py`
- Plan model: `src/cleveragents/domain/models/core/plan.py`
+118
View File
@@ -0,0 +1,118 @@
Feature: Plan CLI spec alignment
As a developer
I want the plan CLI commands aligned to the v3 spec
So that plan use/list/status flags are consistent with the specification
Background:
Given a plan spec alignment CLI runner
And a plan spec alignment mocked lifecycle service
# ---- plan use: multiple projects (positional) ----
Scenario: Plan use with multiple positional projects
Given a plan spec alignment action exists
When I run plan use with positional projects "proj-a" and "proj-b"
Then the plan spec use should succeed
And the plan spec use should link projects "proj-a" and "proj-b"
# ---- plan use: --automation-profile ----
Scenario: Plan use with --automation-profile
Given a plan spec alignment action exists
When I run plan use with automation profile "trusted"
Then the plan spec use should succeed
And the plan spec output should contain "Automation Profile"
# ---- plan use: --invariant (repeatable) ----
Scenario: Plan use with repeatable --invariant
Given a plan spec alignment action exists
When I run plan use with invariants "No warnings" and "Keep compat"
Then the plan spec use should succeed
And the plan spec use should pass invariants to service
# ---- plan use: actor override flags ----
Scenario: Plan use with strategy-actor override
Given a plan spec alignment action exists
When I run plan use with strategy actor "openai/gpt-4"
Then the plan spec use should succeed
And the plan spec output should contain "Strategy Actor"
Scenario: Plan use with execution-actor override
Given a plan spec alignment action exists
When I run plan use with execution actor "anthropic/claude-3"
Then the plan spec use should succeed
And the plan spec output should contain "Execution Actor"
Scenario: Plan use with estimation-actor override
Given a plan spec alignment action exists
When I run plan use with estimation actor "openai/gpt-4"
Then the plan spec use should succeed
And the plan spec output should contain "Estimation Actor"
Scenario: Plan use with invariant-actor override
Given a plan spec alignment action exists
When I run plan use with invariant actor "openai/gpt-4"
Then the plan spec use should succeed
And the plan spec output should contain "Invariant Actor"
# ---- plan use: --arg name=value ----
Scenario: Plan use with --arg name=value
Given a plan spec alignment action exists
When I run plan use with arg "target_coverage=80"
Then the plan spec use should succeed
And the plan spec use should pass argument "target_coverage" with value 80
# ---- plan list: filter combinations ----
Scenario: Plan list with --phase filter
Given plan spec alignment plans exist
When I run plan lifecycle-list with phase "strategize"
Then the plan spec list should succeed
Scenario: Plan list with --state filter
Given plan spec alignment plans exist
When I run plan lifecycle-list with state "queued"
Then the plan spec list should succeed
Scenario: Plan list with --processing-state filter
Given plan spec alignment plans exist
When I run plan lifecycle-list with processing-state "complete"
Then the plan spec list should succeed
Scenario: Plan list with --project filter
Given plan spec alignment plans exist
When I run plan lifecycle-list with project "proj-a"
Then the plan spec list should succeed
Scenario: Plan list with --action filter
Given plan spec alignment plans exist
When I run plan lifecycle-list with action "local/test-action"
Then the plan spec list should succeed
Scenario: Plan list with regex filter
Given plan spec alignment plans exist
When I run plan lifecycle-list with regex "test-action"
Then the plan spec list should succeed
Scenario: Plan list with combined filters
Given plan spec alignment plans exist
When I run plan lifecycle-list combining phase "strategize" with project "proj-a"
Then the plan spec list should succeed
# ---- plan status: output fields ----
Scenario: Plan status renders all required fields
Given a plan spec alignment plan exists for status
When I run plan status for the plan
Then the plan spec status should succeed
And the plan spec status should contain "Action"
And the plan spec status should contain "Phase"
And the plan spec status should contain "Processing State"
And the plan spec status should contain "Projects"
And the plan spec status should contain "Arguments"
And the plan spec status should contain "Automation Profile"
And the plan spec status should contain "Created"
And the plan spec status should contain "Updated"
# ---- plan cancel: --reason ----
Scenario: Plan cancel with --reason
Given a plan spec alignment plan exists for cancel
When I run plan cancel with reason "Requirements changed"
Then the plan spec cancel should succeed
And the plan spec cancel output should contain "Requirements changed"
@@ -0,0 +1,460 @@
"""Step definitions for plan CLI spec alignment feature."""
from __future__ import annotations
from datetime import datetime
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from typer.testing import CliRunner
from cleveragents.cli.commands.plan import app as plan_app
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.plan import (
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanInvariant,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"
def _make_plan(
*,
name: str = "local/test-plan",
action_name: str = "local/test-action",
phase: PlanPhase = PlanPhase.STRATEGIZE,
state: ProcessingState = ProcessingState.QUEUED,
project_links: list[ProjectLink] | None = None,
arguments: dict[str, object] | None = None,
arguments_order: list[str] | None = None,
automation_profile: AutomationProfileRef | None = None,
invariants: list[PlanInvariant] | None = None,
strategy_actor: str | None = "openai/gpt-4",
execution_actor: str | None = "openai/gpt-4",
estimation_actor: str | None = None,
invariant_actor: str | None = None,
) -> Plan:
"""Create a Plan instance for spec alignment tests."""
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=_PLAN_ULID),
namespaced_name=NamespacedName.parse(name),
description="Test plan description",
definition_of_done="All tests pass",
action_name=action_name,
phase=phase,
processing_state=state,
automation_level=AutomationLevel.MANUAL,
project_links=project_links or [],
arguments=dict(arguments) if arguments else {},
arguments_order=arguments_order or [],
automation_profile=automation_profile,
invariants=invariants or [],
strategy_actor=strategy_actor,
execution_actor=execution_actor,
estimation_actor=estimation_actor,
invariant_actor=invariant_actor,
reusable=True,
read_only=False,
created_by=None,
timestamps=PlanTimestamps(created_at=now, updated_at=now),
)
def _make_action(name: str = "local/test-action") -> Action:
"""Create an Action for plan use tests."""
return Action(
namespaced_name=NamespacedName.parse(name),
description="Test action",
long_description=None,
definition_of_done="All tests pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
state=ActionState.AVAILABLE,
created_by=None,
created_at=datetime.now(),
updated_at=datetime.now(),
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a plan spec alignment CLI runner")
def step_plan_spec_runner(context: Context) -> None:
"""Set up the CLI runner."""
context.runner = CliRunner()
@given("a plan spec alignment mocked lifecycle service")
def step_plan_spec_service(context: Context) -> None:
"""Set up a mock PlanLifecycleService."""
context.mock_service = MagicMock()
context.service_patcher = patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=context.mock_service,
)
context.service_patcher.start()
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers = []
context._cleanup_handlers.append(context.service_patcher.stop)
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("a plan spec alignment action exists")
def step_plan_spec_action(context: Context) -> None:
"""Set up an action and configure mock service for use_action."""
context.mock_action = _make_action()
context.mock_service.get_action_by_name.return_value = context.mock_action
# Default plan returned by use_action
context.mock_plan = _make_plan(
project_links=[ProjectLink(project_name="proj-a")],
arguments={"target_coverage": 80},
arguments_order=["target_coverage"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
estimation_actor="openai/gpt-4",
invariant_actor="openai/gpt-4",
)
context.mock_service.use_action.return_value = context.mock_plan
@given("plan spec alignment plans exist")
def step_plan_spec_plans(context: Context) -> None:
"""Set up plans for list filtering tests."""
context.mock_plans = [
_make_plan(
name="local/plan-a",
action_name="local/test-action",
project_links=[ProjectLink(project_name="proj-a")],
),
_make_plan(
name="local/plan-b",
action_name="local/other-action",
phase=PlanPhase.EXECUTE,
state=ProcessingState.COMPLETE,
project_links=[ProjectLink(project_name="proj-b")],
),
]
context.mock_service.list_plans.return_value = context.mock_plans
@given("a plan spec alignment plan exists for status")
def step_plan_spec_status_plan(context: Context) -> None:
"""Set up a plan with all fields for status rendering."""
context.mock_plan = _make_plan(
project_links=[
ProjectLink(project_name="proj-a", alias="api"),
ProjectLink(project_name="proj-b", read_only=True),
],
arguments={"coverage": 80, "framework": "behave"},
arguments_order=["coverage", "framework"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
estimation_actor="openai/gpt-4",
invariant_actor="openai/gpt-4",
)
context.mock_service.get_plan.return_value = context.mock_plan
@given("a plan spec alignment plan exists for cancel")
def step_plan_spec_cancel_plan(context: Context) -> None:
"""Set up a plan for cancel tests."""
context.mock_plan = _make_plan()
context.mock_plan.processing_state = ProcessingState.CANCELLED
context.mock_service.cancel_plan.return_value = context.mock_plan
# ---------------------------------------------------------------------------
# When steps -- plan use
# ---------------------------------------------------------------------------
@when('I run plan use with positional projects "{proj_a}" and "{proj_b}"')
def step_plan_use_positional(context: Context, proj_a: str, proj_b: str) -> None:
"""Run plan use with multiple positional project args."""
context.result = context.runner.invoke(
plan_app, ["use", "local/test-action", proj_a, proj_b]
)
@when('I run plan use with automation profile "{profile}"')
def step_plan_use_profile(context: Context, profile: str) -> None:
"""Run plan use with --automation-profile."""
context.result = context.runner.invoke(
plan_app,
[
"use",
"local/test-action",
"proj-a",
"--automation-profile",
profile,
],
)
@when('I run plan use with invariants "{inv_a}" and "{inv_b}"')
def step_plan_use_with_invariants(context: Context, inv_a: str, inv_b: str) -> None:
"""Run plan use with repeatable --invariant flags."""
context.result = context.runner.invoke(
plan_app,
[
"use",
"local/test-action",
"proj-a",
"--invariant",
inv_a,
"--invariant",
inv_b,
],
)
@when('I run plan use with strategy actor "{actor}"')
def step_plan_use_strategy_actor(context: Context, actor: str) -> None:
"""Run plan use with --strategy-actor."""
context.result = context.runner.invoke(
plan_app,
["use", "local/test-action", "proj-a", "--strategy-actor", actor],
)
@when('I run plan use with execution actor "{actor}"')
def step_plan_use_execution_actor(context: Context, actor: str) -> None:
"""Run plan use with --execution-actor."""
context.result = context.runner.invoke(
plan_app,
["use", "local/test-action", "proj-a", "--execution-actor", actor],
)
@when('I run plan use with estimation actor "{actor}"')
def step_plan_use_estimation_actor(context: Context, actor: str) -> None:
"""Run plan use with --estimation-actor."""
context.result = context.runner.invoke(
plan_app,
["use", "local/test-action", "proj-a", "--estimation-actor", actor],
)
@when('I run plan use with invariant actor "{actor}"')
def step_plan_use_invariant_actor(context: Context, actor: str) -> None:
"""Run plan use with --invariant-actor."""
context.result = context.runner.invoke(
plan_app,
["use", "local/test-action", "proj-a", "--invariant-actor", actor],
)
@when('I run plan use with arg "{arg_str}"')
def step_plan_use_arg(context: Context, arg_str: str) -> None:
"""Run plan use with --arg name=value."""
context.result = context.runner.invoke(
plan_app,
["use", "local/test-action", "proj-a", "--arg", arg_str],
)
# ---------------------------------------------------------------------------
# When steps -- plan lifecycle-list
# ---------------------------------------------------------------------------
@when('I run plan lifecycle-list with phase "{phase}"')
def step_plan_list_phase(context: Context, phase: str) -> None:
"""Run lifecycle-list with --phase filter."""
context.result = context.runner.invoke(
plan_app, ["lifecycle-list", "--phase", phase]
)
@when('I run plan lifecycle-list with state "{state}"')
def step_plan_list_state(context: Context, state: str) -> None:
"""Run lifecycle-list with --state filter."""
context.result = context.runner.invoke(
plan_app, ["lifecycle-list", "--state", state]
)
@when('I run plan lifecycle-list with processing-state "{state}"')
def step_plan_list_processing_state(context: Context, state: str) -> None:
"""Run lifecycle-list with --processing-state filter."""
context.result = context.runner.invoke(
plan_app, ["lifecycle-list", "--processing-state", state]
)
@when('I run plan lifecycle-list with project "{project}"')
def step_plan_list_project(context: Context, project: str) -> None:
"""Run lifecycle-list with --project filter."""
context.result = context.runner.invoke(
plan_app, ["lifecycle-list", "--project", project]
)
@when('I run plan lifecycle-list with action "{action}"')
def step_plan_list_action(context: Context, action: str) -> None:
"""Run lifecycle-list with --action filter."""
context.result = context.runner.invoke(
plan_app, ["lifecycle-list", "--action", action]
)
@when('I run plan lifecycle-list with regex "{regex}"')
def step_plan_list_regex(context: Context, regex: str) -> None:
"""Run lifecycle-list with a regex positional argument."""
context.result = context.runner.invoke(plan_app, ["lifecycle-list", regex])
@when('I run plan lifecycle-list combining phase "{phase}" with project "{project}"')
def step_plan_list_combined(context: Context, phase: str, project: str) -> None:
"""Run lifecycle-list with combined filters."""
context.result = context.runner.invoke(
plan_app, ["lifecycle-list", "--phase", phase, "--project", project]
)
# ---------------------------------------------------------------------------
# When steps -- plan status
# ---------------------------------------------------------------------------
@when("I run plan status for the plan")
def step_plan_status(context: Context) -> None:
"""Run plan status for a specific plan."""
context.result = context.runner.invoke(plan_app, ["status", _PLAN_ULID])
# ---------------------------------------------------------------------------
# When steps -- plan cancel
# ---------------------------------------------------------------------------
@when('I run plan cancel with reason "{reason}"')
def step_plan_cancel_reason(context: Context, reason: str) -> None:
"""Run plan cancel with --reason."""
context.result = context.runner.invoke(
plan_app, ["cancel", _PLAN_ULID, "--reason", reason]
)
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the plan spec use should succeed")
def step_plan_use_ok(context: Context) -> None:
"""Verify plan use succeeded."""
assert context.result.exit_code == 0, (
f"Plan use failed ({context.result.exit_code}): {context.result.output}"
)
@then('the plan spec use should link projects "{proj_a}" and "{proj_b}"')
def step_plan_use_projects(context: Context, proj_a: str, proj_b: str) -> None:
"""Verify the service received both project links."""
call_kwargs = context.mock_service.use_action.call_args
project_links = call_kwargs[1].get(
"project_links", call_kwargs[1].get("project_links", [])
)
linked_names = [link.project_name for link in project_links]
assert proj_a in linked_names, f"Expected {proj_a} in {linked_names}"
assert proj_b in linked_names, f"Expected {proj_b} in {linked_names}"
@then('the plan spec output should contain "{text}"')
def step_plan_output_contains(context: Context, text: str) -> None:
"""Verify the output contains expected text (case-insensitive)."""
output_lower = context.result.output.lower()
text_lower = text.lower()
assert text_lower in output_lower, (
f"Expected '{text}' in output but got:\n{context.result.output}"
)
@then("the plan spec use should pass invariants to service")
def step_plan_use_check_invariants(context: Context) -> None:
"""Verify invariants were passed to use_action."""
call_kwargs = context.mock_service.use_action.call_args[1]
invariants = call_kwargs.get("invariants")
assert invariants is not None, "Expected invariants to be passed"
assert len(invariants) == 2, f"Expected 2 invariants, got {len(invariants)}"
texts = [inv.text for inv in invariants]
assert "No warnings" in texts
assert "Keep compat" in texts
@then('the plan spec use should pass argument "{name}" with value {value}')
def step_plan_use_arg_check(context: Context, name: str, value: str) -> None:
"""Verify argument was passed to use_action."""
call_kwargs = context.mock_service.use_action.call_args[1]
arguments = call_kwargs.get("arguments", {})
assert name in arguments, f"Expected argument {name} in {arguments}"
assert arguments[name] == int(value), (
f"Expected {name}={value}, got {arguments[name]}"
)
@then("the plan spec list should succeed")
def step_plan_list_ok(context: Context) -> None:
"""Verify lifecycle-list succeeded."""
assert context.result.exit_code == 0, (
f"List failed ({context.result.exit_code}): {context.result.output}"
)
@then("the plan spec status should succeed")
def step_plan_status_ok(context: Context) -> None:
"""Verify plan status succeeded."""
assert context.result.exit_code == 0, (
f"Status failed ({context.result.exit_code}): {context.result.output}"
)
@then('the plan spec status should contain "{text}"')
def step_plan_status_contains(context: Context, text: str) -> None:
"""Verify status output contains the expected text."""
output_lower = context.result.output.lower()
text_lower = text.lower()
assert text_lower in output_lower, (
f"Expected '{text}' in status output but got:\n{context.result.output}"
)
@then("the plan spec cancel should succeed")
def step_plan_cancel_ok(context: Context) -> None:
"""Verify plan cancel succeeded."""
assert context.result.exit_code == 0, (
f"Cancel failed ({context.result.exit_code}): {context.result.output}"
)
@then('the plan spec cancel output should contain "{text}"')
def step_plan_cancel_contains(context: Context, text: str) -> None:
"""Verify cancel output contains expected text."""
assert text in context.result.output, (
f"Expected '{text}' in cancel output but got:\n{context.result.output}"
)
+27 -21
View File
@@ -802,6 +802,12 @@ The following work from the previous implementation has been completed and will
- Verification: lint 0 findings, typecheck 0 errors, 2587 scenarios passed, 97% coverage. action.py 100% coverage.
- Branch: `feature/m1-action-cli`
**2026-02-14**: Stage A4b.plan Complete - Plan CLI Flag Alignment [Jeff]
- Updated `src/cleveragents/cli/commands/plan.py`: `plan use` now accepts multiple PROJECT args, `--automation-profile`, `--invariant` (repeatable), `--strategy-actor`/`--execution-actor`/`--estimation-actor`/`--invariant-actor`, `--arg name=value`. Aligned `plan list` filters to spec. Enhanced `plan status` output. Added `--reason` to `plan cancel`.
- Created `features/plan_cli_spec_alignment.feature` with scenarios, Robot smoke test, ASV benchmarks.
- Verification: lint 0 findings, typecheck 0 errors, 2606 scenarios passed, 97% coverage.
- Branch: `feature/m1-plan-cli` (based on `feature/m1-action-cli`)
---
## Roadmap
@@ -1440,7 +1446,7 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
**PARALLEL SUBTRACK A4b.outputs [Jeff]**: Output fields + format parity for action/plan CLI
**PARALLEL SUBTRACK A4b.tests [Brent]**: Behave + Robot CLI coverage after outputs lock
**SEQUENTIAL MERGE NOTE**: A4b.action → A4b.plan → A4b.outputs; A4b.tests runs after A4b.outputs.
- [X] **COMMIT (Owner: Jeff | Group: A4b.action | Branch: feature/m1-action-cli | Planned: Day 6) - Commit message: "feat(cli): align action commands to spec"**
- [X] **COMMIT (Owner: Jeff | Group: A4b.action | Branch: feature/m1-action-cli | Planned: Day 6 | Expected: Day 8) - Commit message: "feat(cli): align action commands to spec"**
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m1-action-cli`
@@ -1459,27 +1465,27 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [X] Forgejo PR [Jeff]: Open PR from `feature/m1-action-cli` to `master` with description "Align action CLI to spec: config-only create, filterable list/show, and consistent output fields.".
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git branch -d feature/m1-action-cli`
- [x] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] **COMMIT (Owner: Jeff | Group: A4b.plan | Branch: feature/m1-plan-cli | Planned: Day 7 | Expected: Day 9) - Commit message: "feat(cli): align plan use/list/status flags"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m1-plan-cli`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Update `plan use` to accept multiple projects plus `--automation-profile`, `--invariant`, `--strategy-actor`, `--execution-actor`, `--estimation-actor`, `--invariant-actor`, and repeatable `--arg name=value`.
- [ ] Code [Jeff]: Align `plan list` filters to spec (`--phase`, `--state`, `--project`, `--action`, optional regex); map `--state` to processing_state.
- [ ] Code [Jeff]: Ensure `plan status` renders action_name, phase, processing_state, project links, arguments, and automation profile.
- [ ] Docs [Jeff]: Update `docs/reference/plan_cli.md` with plan use flags + list/status filters.
- [ ] Tests (Behave) [Jeff]: Add scenarios for plan use with args/invariants/actor overrides and list/status filter combinations.
- [ ] Tests (Robot) [Jeff]: Add Robot smoke for plan use + list/status output (DB-backed lifecycle).
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/plan_cli_bench.py` for plan use/list parsing overhead.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(cli): align plan use/list/status flags"`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m1-plan-cli` to `master` with description "Align plan use/list/status flags and outputs to spec for M1.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m1-plan-cli`
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] **COMMIT (Owner: Jeff | Group: A4b.plan | Branch: feature/m1-plan-cli | Planned: Day 6 | Expected: Day 6) - Commit message: "feat(cli): align plan use/list/status flags"**
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m1-plan-cli`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Update `plan use` to accept multiple projects plus `--automation-profile`, `--invariant`, `--strategy-actor`, `--execution-actor`, `--estimation-actor`, `--invariant-actor`, and repeatable `--arg name=value`.
- [X] Code [Jeff]: Align `plan list` filters to spec (`--phase`, `--state`, `--project`, `--action`, optional regex); map `--state` to processing_state.
- [X] Code [Jeff]: Ensure `plan status` renders action_name, phase, processing_state, project links, arguments, and automation profile.
- [X] Docs [Jeff]: Update `docs/reference/plan_cli.md` with plan use flags + list/status filters.
- [X] Tests (Behave) [Jeff]: Add scenarios for plan use with args/invariants/actor overrides and list/status filter combinations.
- [X] Tests (Robot) [Jeff]: Add Robot smoke for plan use + list/status output (DB-backed lifecycle).
- [X] Tests (ASV) [Jeff]: Add `benchmarks/plan_cli_bench.py` for plan use/list parsing overhead.
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [X] Git [Jeff]: `git add .`
- [X] Git [Jeff]: `git commit -m "feat(cli): align plan use/list/status flags"`
- [X] Forgejo PR [Jeff]: Open PR from `feature/m1-plan-cli` to `master` with description "Align plan use/list/status flags and outputs to spec for M1.".
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git branch -d feature/m1-plan-cli`
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] **COMMIT (Owner: Jeff | Group: A4b.outputs | Branch: feature/m1-cli-formats | Planned: Day 8 | Expected: Day 10) - Commit message: "feat(cli): stabilize action/plan output formats"**
- [ ] Git [Jeff]: `git checkout master`
+314
View File
@@ -0,0 +1,314 @@
"""Helper script for plan_cli_spec.robot smoke tests.
Each subcommand is a self-contained check that prints a sentinel on success.
"""
from __future__ import annotations
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch
# Ensure the local source tree takes priority over any installed copy.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
# Remove any *other* src entries (e.g. /app/src from the parent clone)
sys.path = [_SRC] + [
p
for p in sys.path
if p != _SRC and not (p.endswith("/src") and "cleveragents" not in p.split("/")[-1])
]
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Flush any already-loaded cleveragents modules so they reimport from _SRC
for _mod_name in sorted(sys.modules):
if _mod_name.startswith("cleveragents"):
del sys.modules[_mod_name]
from typer.testing import CliRunner # noqa: E402
from cleveragents.cli.commands.plan import app as plan_app # noqa: E402
from cleveragents.domain.models.core.action import Action, ActionState # noqa: E402
from cleveragents.domain.models.core.plan import ( # noqa: E402
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"
def _mock_action(name: str = "local/smoke-action") -> Action:
return Action(
namespaced_name=NamespacedName.parse(name),
description="Smoke test action",
long_description=None,
definition_of_done="All smoke tests pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
state=ActionState.AVAILABLE,
created_by=None,
created_at=datetime.now(),
updated_at=datetime.now(),
)
def _mock_plan(
project_links: list[ProjectLink] | None = None,
arguments: dict[str, object] | None = None,
) -> Plan:
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=_PLAN_ULID),
namespaced_name=NamespacedName.parse("local/smoke-plan"),
description="Smoke test plan",
definition_of_done="All smoke tests pass",
action_name="local/smoke-action",
phase=PlanPhase.STRATEGIZE,
processing_state=ProcessingState.QUEUED,
automation_level=AutomationLevel.MANUAL,
project_links=project_links or [],
arguments=dict(arguments) if arguments else {},
arguments_order=list((arguments or {}).keys()),
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
estimation_actor="openai/gpt-4",
invariant_actor="openai/gpt-4",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
created_by=None,
timestamps=PlanTimestamps(created_at=now, updated_at=now),
)
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def use_positional_projects() -> None:
"""Verify plan use accepts multiple positional projects."""
mock_service = MagicMock()
mock_service.get_action_by_name.return_value = _mock_action()
mock_service.use_action.return_value = _mock_plan(
project_links=[
ProjectLink(project_name="proj-a"),
ProjectLink(project_name="proj-b"),
],
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(
plan_app, ["use", "local/smoke-action", "proj-a", "proj-b"]
)
if result.exit_code == 0:
print("plan-cli-use-positional-ok")
else:
print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
def use_automation_profile() -> None:
"""Verify plan use accepts --automation-profile."""
mock_service = MagicMock()
mock_service.get_action_by_name.return_value = _mock_action()
mock_service.use_action.return_value = _mock_plan()
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(
plan_app,
[
"use",
"local/smoke-action",
"proj-a",
"--automation-profile",
"trusted",
],
)
if result.exit_code == 0:
print("plan-cli-use-profile-ok")
else:
print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
def use_invariant() -> None:
"""Verify plan use accepts repeatable --invariant."""
mock_service = MagicMock()
mock_service.get_action_by_name.return_value = _mock_action()
mock_service.use_action.return_value = _mock_plan()
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(
plan_app,
[
"use",
"local/smoke-action",
"proj-a",
"--invariant",
"No warnings",
"--invariant",
"Keep compat",
],
)
if result.exit_code == 0:
print("plan-cli-use-invariant-ok")
else:
print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
def use_actor_overrides() -> None:
"""Verify plan use accepts actor override flags."""
mock_service = MagicMock()
mock_service.get_action_by_name.return_value = _mock_action()
mock_service.use_action.return_value = _mock_plan()
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(
plan_app,
[
"use",
"local/smoke-action",
"proj-a",
"--strategy-actor",
"openai/gpt-4",
"--execution-actor",
"anthropic/claude-3",
"--estimation-actor",
"openai/gpt-4",
"--invariant-actor",
"openai/gpt-4",
],
)
if result.exit_code == 0:
print("plan-cli-use-actors-ok")
else:
print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
def list_filters() -> None:
"""Verify lifecycle-list accepts filter flags."""
mock_service = MagicMock()
mock_service.list_plans.return_value = [
_mock_plan(project_links=[ProjectLink(project_name="proj-a")]),
]
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
r1 = runner.invoke(plan_app, ["lifecycle-list", "--phase", "strategize"])
r2 = runner.invoke(plan_app, ["lifecycle-list", "--state", "queued"])
r3 = runner.invoke(plan_app, ["lifecycle-list", "--project", "proj-a"])
r4 = runner.invoke(
plan_app, ["lifecycle-list", "--action", "local/smoke-action"]
)
if all(r.exit_code == 0 for r in [r1, r2, r3, r4]):
print("plan-cli-list-filters-ok")
else:
codes = [r1.exit_code, r2.exit_code, r3.exit_code, r4.exit_code]
print(f"FAIL: list filters exit codes: {codes}", file=sys.stderr)
sys.exit(1)
def status_fields() -> None:
"""Verify plan status renders required fields."""
mock_service = MagicMock()
plan = _mock_plan(
project_links=[ProjectLink(project_name="proj-a", alias="api")],
arguments={"coverage": 80},
)
mock_service.get_plan.return_value = plan
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(plan_app, ["status", _PLAN_ULID])
if result.exit_code == 0:
output = result.output.lower()
required = ["action", "phase", "processing state", "projects", "created"]
missing = [f for f in required if f not in output]
if missing:
print(f"FAIL: missing fields: {missing}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
print("plan-cli-status-fields-ok")
else:
print(f"FAIL: status returned {result.exit_code}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
def cancel_reason() -> None:
"""Verify plan cancel accepts --reason."""
mock_service = MagicMock()
plan = _mock_plan()
plan.processing_state = ProcessingState.CANCELLED
mock_service.cancel_plan.return_value = plan
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(
plan_app, ["cancel", _PLAN_ULID, "--reason", "Requirements changed"]
)
if result.exit_code == 0 and "Requirements changed" in result.output:
print("plan-cli-cancel-reason-ok")
else:
print(f"FAIL: cancel returned {result.exit_code}", file=sys.stderr)
print(result.output, file=sys.stderr)
sys.exit(1)
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS = {
"use-positional-projects": use_positional_projects,
"use-automation-profile": use_automation_profile,
"use-invariant": use_invariant,
"use-actor-overrides": use_actor_overrides,
"list-filters": list_filters,
"status-fields": status_fields,
"cancel-reason": cancel_reason,
}
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(2)
_COMMANDS[sys.argv[1]]()
if __name__ == "__main__":
main()
+53
View File
@@ -0,0 +1,53 @@
*** Settings ***
Documentation Smoke tests for Plan CLI spec alignment (use/list/status flags)
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_plan_cli_spec.py
*** Test Cases ***
Plan Use Accepts Positional Projects
[Documentation] Verify that ``plan use`` accepts multiple positional projects
${result}= Run Process ${PYTHON} ${HELPER} use-positional-projects cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-use-positional-ok
Plan Use Accepts Automation Profile Flag
[Documentation] Verify that ``plan use --automation-profile`` is accepted
${result}= Run Process ${PYTHON} ${HELPER} use-automation-profile cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-use-profile-ok
Plan Use Accepts Invariant Flag
[Documentation] Verify that ``plan use --invariant`` is accepted (repeatable)
${result}= Run Process ${PYTHON} ${HELPER} use-invariant cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-use-invariant-ok
Plan Use Accepts Actor Override Flags
[Documentation] Verify that plan use accepts actor override flags
${result}= Run Process ${PYTHON} ${HELPER} use-actor-overrides cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-use-actors-ok
Plan Lifecycle List Accepts Filter Flags
[Documentation] Verify that ``plan lifecycle-list`` accepts --phase, --state, --project, --action
${result}= Run Process ${PYTHON} ${HELPER} list-filters cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-list-filters-ok
Plan Status Renders Required Fields
[Documentation] Verify that ``plan status`` renders action_name, phase, state, projects, args, timestamps
${result}= Run Process ${PYTHON} ${HELPER} status-fields cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-status-fields-ok
Plan Cancel Accepts Reason Flag
[Documentation] Verify that ``plan cancel --reason`` is accepted
${result}= Run Process ${PYTHON} ${HELPER} cancel-reason cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} plan-cli-cancel-reason-ok
+235 -24
View File
@@ -880,11 +880,18 @@ def _print_lifecycle_plan(plan: Any, title: str = "Plan") -> None:
return
state_display = plan.state.value if plan.state else "unknown"
projects = (
", ".join(link.project_name for link in plan.project_links)
if plan.project_links
else "(none)"
)
# Format project links with aliases
project_parts: list[str] = []
for link in plan.project_links:
label = link.project_name
if link.alias:
label = f"{label} (alias: {link.alias})"
if link.read_only:
label = f"{label} [read-only]"
project_parts.append(label)
projects = ", ".join(project_parts) if project_parts else "(none)"
description_preview = plan.description[:200]
if len(plan.description) > 200:
description_preview = f"{description_preview}..."
@@ -892,16 +899,68 @@ def _print_lifecycle_plan(plan: Any, title: str = "Plan") -> None:
details = (
f"[bold]ID:[/bold] {plan.identity.plan_id}\n"
f"[bold]Name:[/bold] {plan.namespaced_name}\n"
f"[bold]Action:[/bold] {plan.action_name}\n"
f"[bold]Phase:[/bold] {plan.phase.value}\n"
f"[bold]State:[/bold] {state_display}\n"
f"[bold]Processing State:[/bold] {state_display}\n"
f"[bold]Projects:[/bold] {projects}\n"
f"[bold]Description:[/bold]\n {description_preview}\n"
f"[bold]Strategy Actor:[/bold] {plan.strategy_actor or '(not set)'}\n"
f"[bold]Execution Actor:[/bold] {plan.execution_actor or '(not set)'}\n"
f"[bold]Terminal:[/bold] {'yes' if plan.is_terminal else 'no'}\n"
f"[bold]Created:[/bold] {plan.timestamps.created_at}"
)
# Optional actors
if plan.estimation_actor:
details += f"[bold]Estimation Actor:[/bold] {plan.estimation_actor}\n"
if plan.invariant_actor:
details += f"[bold]Invariant Actor:[/bold] {plan.invariant_actor}\n"
# Arguments (ordered)
if plan.arguments:
details += "[bold]Arguments:[/bold]\n"
ordered_keys = (
plan.arguments_order
if plan.arguments_order
else sorted(plan.arguments.keys())
)
for key in ordered_keys:
if key in plan.arguments:
details += f" {key} = {plan.arguments[key]}\n"
# Automation profile info
if plan.automation_profile:
details += (
f"[bold]Automation Profile:[/bold] "
f"{plan.automation_profile.profile_name} "
f"(source: {plan.automation_profile.provenance.value})\n"
)
details += f"[bold]Automation Level:[/bold] {plan.automation_level.value}\n"
details += f"[bold]Terminal:[/bold] {'yes' if plan.is_terminal else 'no'}\n"
# Timestamps
details += f"[bold]Created:[/bold] {plan.timestamps.created_at}\n"
details += f"[bold]Updated:[/bold] {plan.timestamps.updated_at}"
if plan.timestamps.strategize_started_at:
details += (
f"\n[bold]Strategize Started:[/bold] "
f"{plan.timestamps.strategize_started_at}"
)
if plan.timestamps.strategize_completed_at:
details += (
f"\n[bold]Strategize Completed:[/bold] "
f"{plan.timestamps.strategize_completed_at}"
)
if plan.timestamps.execute_started_at:
details += (
f"\n[bold]Execute Started:[/bold] {plan.timestamps.execute_started_at}"
)
if plan.timestamps.execute_completed_at:
details += (
f"\n[bold]Execute Completed:[/bold] {plan.timestamps.execute_completed_at}"
)
if plan.timestamps.applied_at:
details += f"\n[bold]Applied At:[/bold] {plan.timestamps.applied_at}"
if plan.error_message:
details += f"\n[bold red]Error:[/bold red] {plan.error_message}"
@@ -912,19 +971,23 @@ def _print_lifecycle_plan(plan: Any, title: str = "Plan") -> None:
def use_action(
action_name: Annotated[
str,
typer.Argument(help="Action name or ID to use"),
typer.Argument(help="Action name to use"),
],
projects: Annotated[
list[str] | None,
typer.Argument(help="Projects to apply the action on (one or more)"),
] = None,
project: Annotated[
list[str],
list[str] | None,
typer.Option(
"--project",
"-p",
help=(
"Project ID to use the action on "
"Project name to use the action on "
"(can be repeated for multiple projects)"
),
),
],
] = None,
arg: Annotated[
list[str] | None,
typer.Option(
@@ -933,6 +996,48 @@ def use_action(
help="Argument value (format: name=value)",
),
] = None,
automation_profile: Annotated[
str | None,
typer.Option(
"--automation-profile",
help="Automation profile name to use for this plan",
),
] = None,
invariant: Annotated[
list[str] | None,
typer.Option(
"--invariant",
help="Invariant constraint text (repeatable)",
),
] = None,
strategy_actor: Annotated[
str | None,
typer.Option(
"--strategy-actor",
help="Override the strategy actor for this plan",
),
] = None,
execution_actor: Annotated[
str | None,
typer.Option(
"--execution-actor",
help="Override the execution actor for this plan",
),
] = None,
estimation_actor: Annotated[
str | None,
typer.Option(
"--estimation-actor",
help="Override the estimation actor for this plan",
),
] = None,
invariant_actor: Annotated[
str | None,
typer.Option(
"--invariant-actor",
help="Override the invariant reconciliation actor for this plan",
),
] = None,
automation_level: Annotated[
str | None,
typer.Option(
@@ -946,19 +1051,32 @@ def use_action(
) -> None:
"""Use an action on projects to create a plan in Strategize phase.
This is the 'use' command that transitions from Action to Strategize.
The action must be in 'available' state.
The first positional argument is the ACTION name. Subsequent positional
arguments are PROJECT names. Projects can also be supplied via the
repeatable ``--project`` / ``-p`` option.
Examples:
agents plan use local/code-coverage --project proj-123 --arg target_coverage=80
agents plan use local/code-coverage proj-1 proj-2 --arg target_coverage=80
agents plan use local/lint --project proj-1 --invariant "No new warnings"
"""
from cleveragents.application.services.plan_lifecycle_service import (
ActionNotAvailableError,
)
from cleveragents.domain.models.core.plan import (
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
InvariantSource,
PlanInvariant,
ProjectLink,
)
try:
service = _get_lifecycle_service()
# Merge positional projects and --project option
all_projects: list[str] = list(projects or []) + list(project or [])
# Parse arguments
arguments: dict[str, str | int | float | bool | list[str]] = {}
if arg:
@@ -982,15 +1100,12 @@ def use_action(
else:
arguments[name] = value
# Get action by name or ID
# Get action by name
action = service.get_action_by_name(action_name)
# Parse automation level if provided
resolved_automation = None
if automation_level:
from cleveragents.domain.models.core.plan import AutomationLevel
try:
resolved_automation = AutomationLevel(automation_level.lower())
except ValueError:
@@ -1002,9 +1117,15 @@ def use_action(
raise typer.Abort() from None
# Build project links from project names
from cleveragents.domain.models.core.plan import ProjectLink
project_links = [ProjectLink(project_name=p) for p in all_projects]
project_links = [ProjectLink(project_name=p) for p in project]
# Build plan-level invariants from --invariant flags
plan_invariants: list[PlanInvariant] | None = None
if invariant:
plan_invariants = [
PlanInvariant(text=inv_text, source=InvariantSource.PLAN)
for inv_text in invariant
]
# Use the action to create a plan
plan = service.use_action(
@@ -1012,8 +1133,26 @@ def use_action(
project_links=project_links,
arguments=arguments if arguments else None,
automation_level=resolved_automation,
invariants=plan_invariants,
)
# Apply automation profile override if provided
if automation_profile:
plan.automation_profile = AutomationProfileRef(
profile_name=automation_profile,
provenance=AutomationProfileProvenance.PLAN,
)
# Apply actor overrides if provided
if strategy_actor:
plan.strategy_actor = strategy_actor
if execution_actor:
plan.execution_actor = execution_actor
if estimation_actor:
plan.estimation_actor = estimation_actor
if invariant_actor:
plan.invariant_actor = invariant_actor
_print_lifecycle_plan(plan, title="Plan Created")
console.print(
@@ -1218,19 +1357,49 @@ def plan_status(
@app.command("lifecycle-list")
def lifecycle_list_plans(
regex: Annotated[
str | None,
typer.Argument(
help="Optional regex pattern to filter plan names",
),
] = None,
phase: Annotated[
str | None,
typer.Option(
"--phase",
"-p",
help="Filter by phase (strategize, execute, apply, applied)",
),
] = None,
state: Annotated[
str | None,
typer.Option(
"--state",
help=(
"Filter by processing state "
"(queued/processing/errored/complete/cancelled)"
),
),
] = None,
processing_state: Annotated[
str | None,
typer.Option(
"--processing-state",
help="Filter by processing state (alias for --state)",
),
] = None,
project_id: Annotated[
str | None,
typer.Option(
"--project",
help="Filter by project ID",
"-p",
help="Filter by project name",
),
] = None,
action_filter: Annotated[
str | None,
typer.Option(
"--action",
help="Filter by action name",
),
] = None,
) -> None:
@@ -1240,7 +1409,12 @@ def lifecycle_list_plans(
(Strategize -> Execute -> Apply -> Applied).
"""
try:
from cleveragents.domain.models.core.plan import PlanPhase
import re
from cleveragents.domain.models.core.plan import (
PlanPhase,
ProcessingState,
)
service = _get_lifecycle_service()
@@ -1256,8 +1430,43 @@ def lifecycle_list_plans(
)
raise typer.Abort() from exc
# Resolve --state to processing_state (--state maps to processing_state)
resolved_state_str = state or processing_state
state_filter: ProcessingState | None = None
if resolved_state_str:
try:
state_filter = ProcessingState(resolved_state_str.lower())
except ValueError as exc:
console.print(
f"[red]Invalid state:[/red] {resolved_state_str}. "
"Valid values: queued, processing, errored, complete, cancelled"
)
raise typer.Abort() from exc
plans = service.list_plans(phase=phase_filter, project_name=project_id)
# Apply processing state filter
if state_filter:
plans = [p for p in plans if p.processing_state == state_filter]
# Apply action name filter
if action_filter:
plans = [p for p in plans if p.action_name == action_filter]
# Apply regex name filter
if regex:
try:
pattern = re.compile(regex)
except re.error as exc:
console.print(f"[red]Invalid regex pattern:[/red] {regex}")
raise typer.Abort() from exc
plans = [
p
for p in plans
if pattern.search(str(p.namespaced_name))
or pattern.search(p.namespaced_name.name)
]
if not plans:
console.print("[yellow]No plans found.[/yellow]")
console.print(
@@ -1269,6 +1478,7 @@ def lifecycle_list_plans(
table = Table(title=f"V3 Lifecycle Plans ({len(plans)} total)")
table.add_column("ID", style="cyan")
table.add_column("Name", style="white")
table.add_column("Action", style="blue")
table.add_column("Phase", style="yellow")
table.add_column("State", style="magenta")
table.add_column("Projects", style="green")
@@ -1283,6 +1493,7 @@ def lifecycle_list_plans(
table.add_row(
plan.identity.plan_id[:8] + "...",
str(plan.namespaced_name),
plan.action_name,
plan.phase.value,
plan.state.value if plan.state else "",
projects or "(none)",