fix invariant: use _resolve_scope consistently in list_invariants and respect is_global param
- Fix _resolve_scope() to properly use the is_global parameter instead of ignoring it - Replace standalone if/elif chain in list_invariants with a call to _resolve_scope for consistent scope resolution
This commit is contained in:
@@ -0,0 +1,274 @@
|
||||
"""Invariant management commands for CleverAgents CLI.
|
||||
|
||||
The ``agents invariant`` command group manages natural-language constraints
|
||||
that flow into plan execution decisions.
|
||||
|
||||
## Commands
|
||||
|
||||
| Command | Description |
|
||||
|----------------------------|------------------------------------------|
|
||||
| ``agents invariant add`` | Add a new invariant constraint |
|
||||
| ``agents invariant list`` | List invariants with optional filters |
|
||||
| ``agents invariant remove``| Soft-delete an invariant by ID |
|
||||
|
||||
## Scope Flags
|
||||
|
||||
Each invariant belongs to exactly one scope. Pass the matching flag:
|
||||
|
||||
- ``--global``: System-wide invariant
|
||||
- ``--project PROJECT``: Project-scoped invariant
|
||||
- ``--action ACTION``: Action-template invariant
|
||||
- ``--plan PLAN_ID``: Plan-specific invariant
|
||||
|
||||
If no scope flag is given, ``--global`` is assumed.
|
||||
|
||||
## Examples
|
||||
|
||||
```bash
|
||||
agents invariant add --global "Never delete production data"
|
||||
agents invariant add --project myapp "All API changes need tests"
|
||||
agents invariant list --effective --project myapp
|
||||
agents invariant remove INV_ULID
|
||||
```
|
||||
|
||||
Based on ``docs/specification.md`` and implementation plan Stage M3.5.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Annotated
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
from cleveragents.application.services.invariant_service import InvariantService
|
||||
from cleveragents.cli.formatting import OutputFormat, format_output
|
||||
from cleveragents.core.exceptions import CleverAgentsError, NotFoundError
|
||||
from cleveragents.domain.models.core.invariant import Invariant, InvariantScope
|
||||
|
||||
app = typer.Typer(
|
||||
help="Manage invariant constraints for plan execution.",
|
||||
)
|
||||
console = Console()
|
||||
|
||||
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
|
||||
|
||||
# Module-level service instance (in-memory, same lifetime as CLI process)
|
||||
_service: InvariantService | None = None
|
||||
|
||||
|
||||
def _get_service() -> InvariantService:
|
||||
"""Return (or lazily create) the module-level InvariantService."""
|
||||
global _service
|
||||
if _service is None:
|
||||
_service = InvariantService()
|
||||
return _service
|
||||
|
||||
|
||||
def _resolve_scope(
|
||||
is_global: bool,
|
||||
project: str | None,
|
||||
plan: str | None,
|
||||
action: str | None,
|
||||
) -> tuple[InvariantScope, str]:
|
||||
"""Resolve scope flag combination to (scope, source_name).
|
||||
|
||||
Raises:
|
||||
typer.BadParameter: On conflicting or missing flags.
|
||||
"""
|
||||
flags_set = sum(
|
||||
[is_global, project is not None, plan is not None, action is not None]
|
||||
)
|
||||
if flags_set > 1:
|
||||
raise typer.BadParameter(
|
||||
"Specify at most one scope flag: --global, --project, --plan, or --action"
|
||||
)
|
||||
|
||||
# Explicit global check (handles the case where --global is set)
|
||||
if is_global:
|
||||
return InvariantScope.GLOBAL, "system"
|
||||
|
||||
if project is not None:
|
||||
return InvariantScope.PROJECT, project
|
||||
if plan is not None:
|
||||
return InvariantScope.PLAN, plan
|
||||
if action is not None:
|
||||
return InvariantScope.ACTION, action
|
||||
|
||||
# Default to global when no scope flag is provided
|
||||
return InvariantScope.GLOBAL, "system"
|
||||
|
||||
|
||||
def _invariant_dict(inv: Invariant) -> dict[str, object]:
|
||||
"""Serialise an Invariant for non-rich output formats."""
|
||||
return {
|
||||
"id": inv.id,
|
||||
"text": inv.text,
|
||||
"scope": inv.scope.value,
|
||||
"source_name": inv.source_name,
|
||||
"active": inv.active,
|
||||
"created_at": inv.created_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
@app.command()
|
||||
def add(
|
||||
text: Annotated[str, typer.Argument(help="The invariant constraint text")],
|
||||
is_global: Annotated[
|
||||
bool, typer.Option("--global", help="System-wide invariant")
|
||||
] = False,
|
||||
project: Annotated[
|
||||
str | None, typer.Option("--project", help="Project name")
|
||||
] = None,
|
||||
plan: Annotated[str | None, typer.Option("--plan", help="Plan ID (ULID)")] = None,
|
||||
action: Annotated[str | None, typer.Option("--action", help="Action name")] = None,
|
||||
fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
|
||||
) -> None:
|
||||
"""Add a new invariant constraint.
|
||||
|
||||
Examples:
|
||||
agents invariant add --global "Never delete production data"
|
||||
agents invariant add --project myapp "All API changes need tests"
|
||||
"""
|
||||
try:
|
||||
scope, source_name = _resolve_scope(is_global, project, plan, action)
|
||||
service = _get_service()
|
||||
inv = service.add_invariant(text=text, scope=scope, source_name=source_name)
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
console.print(format_output(_invariant_dict(inv), fmt))
|
||||
return
|
||||
|
||||
console.print(f"[green]Invariant added:[/green] {inv.id}")
|
||||
console.print(f" Text: {inv.text}")
|
||||
console.print(f" Scope: {inv.scope.value}")
|
||||
console.print(f" Source: {inv.source_name}")
|
||||
|
||||
except CleverAgentsError as e:
|
||||
console.print(f"[red]Error:[/red] {e.message}")
|
||||
raise typer.Abort() from e
|
||||
|
||||
|
||||
@app.command("list")
|
||||
def list_invariants(
|
||||
is_global: Annotated[
|
||||
bool, typer.Option("--global", help="Filter global invariants")
|
||||
] = False,
|
||||
project: Annotated[
|
||||
str | None, typer.Option("--project", help="Filter by project")
|
||||
] = None,
|
||||
plan: Annotated[
|
||||
str | None, typer.Option("--plan", help="Filter by plan ID")
|
||||
] = None,
|
||||
action: Annotated[
|
||||
str | None, typer.Option("--action", help="Filter by action")
|
||||
] = None,
|
||||
effective: Annotated[
|
||||
bool, typer.Option("--effective", help="Show merged effective set")
|
||||
] = False,
|
||||
regex: Annotated[
|
||||
str | None,
|
||||
typer.Argument(help="Optional regex to filter invariant text"),
|
||||
] = None,
|
||||
fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
|
||||
) -> None:
|
||||
"""List invariants with optional filters.
|
||||
|
||||
Examples:
|
||||
agents invariant list
|
||||
agents invariant list --global
|
||||
agents invariant list --effective --project myapp
|
||||
agents invariant list "data.*safe"
|
||||
"""
|
||||
try:
|
||||
service = _get_service()
|
||||
|
||||
# Use shared scope resolver for consistent mutual-exclusion validation
|
||||
scope, source_name = _resolve_scope(is_global, project, plan, action)
|
||||
|
||||
invariants = service.list_invariants(
|
||||
scope=scope,
|
||||
source_name=source_name,
|
||||
effective=effective,
|
||||
)
|
||||
|
||||
# Apply regex filter
|
||||
if regex:
|
||||
try:
|
||||
pattern = re.compile(regex)
|
||||
except re.error as exc:
|
||||
console.print(f"[red]Invalid regex:[/red] {regex}")
|
||||
raise typer.Abort() from exc
|
||||
invariants = [inv for inv in invariants if pattern.search(inv.text)]
|
||||
|
||||
if not invariants:
|
||||
console.print("[yellow]No invariants found.[/yellow]")
|
||||
return
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
data = [_invariant_dict(inv) for inv in invariants]
|
||||
console.print(format_output(data, fmt))
|
||||
return
|
||||
|
||||
table = Table(title=f"Invariants ({len(invariants)} total)")
|
||||
table.add_column("ID", style="cyan", max_width=26)
|
||||
table.add_column("Scope", style="yellow")
|
||||
table.add_column("Source", style="magenta")
|
||||
table.add_column("Text", style="white")
|
||||
table.add_column("Active", justify="center")
|
||||
|
||||
for inv in invariants:
|
||||
table.add_row(
|
||||
inv.id,
|
||||
inv.scope.value,
|
||||
inv.source_name,
|
||||
inv.text,
|
||||
"yes" if inv.active else "no",
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
|
||||
except CleverAgentsError as e:
|
||||
console.print(f"[red]Error:[/red] {e.message}")
|
||||
raise typer.Abort() from e
|
||||
|
||||
|
||||
@app.command()
|
||||
def remove(
|
||||
invariant_id: Annotated[str, typer.Argument(help="Invariant ULID to remove")],
|
||||
yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
|
||||
fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
|
||||
) -> None:
|
||||
"""Remove (soft-delete) an invariant by ID.
|
||||
|
||||
Examples:
|
||||
agents invariant remove 01HXYZ...
|
||||
agents invariant remove --yes 01HXYZ...
|
||||
"""
|
||||
try:
|
||||
if not yes:
|
||||
confirmed = typer.confirm(
|
||||
f"Remove invariant {invariant_id}?", default=False
|
||||
)
|
||||
if not confirmed:
|
||||
console.print("[yellow]Cancelled.[/yellow]")
|
||||
raise typer.Abort()
|
||||
|
||||
service = _get_service()
|
||||
inv = service.remove_invariant(invariant_id)
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
data = _invariant_dict(inv)
|
||||
console.print(format_output(data, fmt))
|
||||
return
|
||||
|
||||
console.print(f"[green]Invariant removed:[/green] {inv.id}")
|
||||
|
||||
except NotFoundError as e:
|
||||
console.print(f"[red]Invariant not found:[/red] {invariant_id}")
|
||||
raise typer.Abort() from e
|
||||
except CleverAgentsError as e:
|
||||
console.print(f"[red]Error:[/red] {e.message}")
|
||||
raise typer.Abort() from e
|
||||
Reference in New Issue
Block a user