fix(cli): add Invariants and Validations panels to project show rich output #9460

Open
HAL9000 wants to merge 2 commits from fix/project-show-missing-panels into master
12 changed files with 989 additions and 650 deletions
+7
View File
@@ -230,6 +230,13 @@ ensuring data is stored with proper parameter values.
- **Plan Rollback Command** (#8557): Implemented `agents plan rollback <plan-id> [<checkpoint-id>]` for checkpoint-based plan state restoration in Epic #8493. The command restores a plan's sandbox to the state captured at a given checkpoint, discarding all decisions made after that checkpoint. The checkpoint can be specified as an optional positional second argument or via the `--to-checkpoint` named option. Supports `--yes/-y` flag to skip confirmation prompts and `--format/-f` for output format selection (rich/plain/json/yaml). Included with comprehensive BDD test coverage (>= 97%) and spec-aligned output formatting showing rollback summary, changes reverted, impact analysis, and post-rollback state panels.
### Fixed
- **Project Show Missing Invariants and Validations Panels** (#9333): The
`agents project show` command now displays **Invariants** and **Validations**
panels in its rich output. The `NamespacedProject` domain model gains
`invariants: list[str]` and `invariant_actor: str | None` fields loaded from
the database. The `_project_spec_dict()` helper now includes `invariants`,
`invariant_actor`, and `validations` keys, ensuring all non-rich output
formats (JSON, YAML, plain, table) also expose these fields.
- **ACMS execute-phase assembler respects project-level hot_max_tokens** (#11035): Fixed
``_resolve_hot_max_tokens()`` to read ``hot_max_tokens`` from
``context_policy_json["acms_config"]["hot_max_tokens"]`` — the correct sub-key written
+1
View File
@@ -76,3 +76,4 @@ Below are some of the specific details of various contributions.
* HAL 9000 has contributed the ACMS Context Tier Hydration documentation (PR #9208 / issue #6175): documented the `context_tier_hydrator` module in the ACMS Architecture section of the specification, covering its public interface, file listing strategy, budget limits, and fragment structure.
* HAL 9000 has contributed the agent task memory leak fix (#9044): replaced `list.remove` with `set.discard` as the done_callback for asyncio tasks in `Agent._tasks`, preventing unbounded memory growth in long-lived agents and ensuring safe concurrent task removal.
* HAL 9000 has contributed the ACMS context show/clear CLI commands (PR #9675 / issue #9586): implemented `context show <view>` displaying assembled context with per-tier budget utilization summary (hot/warm/cold), and `context clear` with --path, --tag, --tier filtering plus confirmation prompt with --yes bypass. Includes 12 Behave BDD scenarios, 9 Robot Framework integration tests, ASV benchmarks, full type annotations, and _TierServiceProtocol for type safety.
* HAL 9000 has contributed the Project Show Invariants and Validations panels fix (PR #9460 / issue #9333): added `invariants` and `invariant_actor` fields to the `NamespacedProject` domain model, updated `NamespacedProjectModel.to_domain()` and `from_domain()` for roundtrip preservation, extracted `project_show.py` with Invariants/Validations rich panels, and restructured project CLI commands into modular files under 500 lines.
+29
View File
@@ -181,3 +181,32 @@ Feature: Project CLI commands (B0.cli.projects)
And the spec dict should have key "linked_resources"
And the spec dict should have key "created_at"
And the spec dict should have key "updated_at"
Scenario: Project spec dict contains invariants and validations keys
Given a project "local/spec-inv" already exists
When I generate the project spec dict for "local/spec-inv"
Then the spec dict should have key "invariants"
And the spec dict should have key "invariant_actor"
And the spec dict should have key "validations"
# ── Invariants and Validations panels ───────────────────────
Scenario: Show project with no invariants displays none
Given a project "local/no-inv-proj" already exists
When I show project "local/no-inv-proj"
Then the show output should contain "invariants"
Scenario: Show project displays validations count
Given a project "local/val-proj" already exists
When I show project "local/val-proj"
Then the show output should contain "validations"
Scenario: Show project with invariants displays invariant list
Given a project "local/inv-show-proj" already exists with invariants
When I show project "local/inv-show-proj"
Then the show output should contain "invariants"
Scenario: Show project with invariant_actor displays actor
Given a project "local/actor-proj" already exists with invariant actor
When I show project "local/actor-proj"
Then the show output should contain "invariant_actor"
+34
View File
@@ -290,6 +290,40 @@ def step_pcli_project_exists_with_desc(context: Any, name: str, desc: str) -> No
_pcli_create_project(context, name, description=desc)
@given('a project "{name}" already exists with invariants')
def step_pcli_project_exists_with_invariants(context: Any, name: str) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
invariants=["do not break the API", "keep tests green"],
)
context.pcli_project_repo.create(proj)
@given('a project "{name}" already exists with invariant actor')
def step_pcli_project_exists_with_invariant_actor(context: Any, name: str) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
invariant_actor="test-actor",
)
context.pcli_project_repo.create(proj)
@given('a registered resource "{res_name}" exists')
def step_pcli_registered_resource_exists(context: Any, res_name: str) -> None:
resource_id = _pcli_register_resource(context, res_name)
+9 -2
View File
@@ -157,11 +157,18 @@ def test_spec_dict() -> None:
"name",
"description",
"linked_resources",
"invariants",
"invariant_actor",
"validations",
"created_at",
"updated_at",
}
if set(data.keys()) != expected_keys:
raise AssertionError(f"Key mismatch: {set(data.keys())} != {expected_keys}")
missing = expected_keys - set(data.keys())
extra = set(data.keys()) - expected_keys
if missing or extra:
raise AssertionError(
f"Key mismatch: missing={missing}, extra={extra}, got={set(data.keys())}"
)
print("spec-dict-ok")
+19 -645
View File
@@ -19,31 +19,29 @@ Based on ADR-009 (CLI Framework) and implementation_plan.md task B0.cli.projects
from __future__ import annotations
import re
from datetime import UTC, datetime
from pathlib import Path
from typing import Annotated, Any
import typer
from rich.panel import Panel
from rich.table import Table
from cleveragents.application.services.context_service import DEFAULT_IGNORE_PATTERNS
from cleveragents.cli.commands.project_context import app as context_app
from cleveragents.cli.commands.project_legacy import (
file_filter_app,
register_legacy_commands,
)
from cleveragents.cli.commands.project_resource_commands import (
register_resource_commands,
)
from cleveragents.cli.commands.project_show import register_show_command
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.cli.renderers import _get_console, _get_err_console
from cleveragents.core.exceptions import (
CleverAgentsError,
ConfigurationError,
DatabaseError,
NotFoundError,
ValidationError,
)
# Create sub-app for project commands
app = typer.Typer(help="Project management commands")
file_filter_app = typer.Typer(
help="Manage project include/exclude filters", name="file-filter"
)
console = _get_console()
err_console = _get_err_console()
@@ -155,7 +153,8 @@ def _project_spec_dict(project: Any) -> dict[str, object]:
"""Return project data as a dict using spec field names.
Keys: namespaced_name, namespace, name, description,
linked_resources, created_at, updated_at.
linked_resources, invariants, invariant_actor, validations,
created_at, updated_at.
"""
linked: list[dict[str, object]] = []
for lr in project.linked_resources:
@@ -176,6 +175,9 @@ def _project_spec_dict(project: Any) -> dict[str, object]:
"name": project.name,
"description": project.description,
"linked_resources": linked,
"invariants": list(getattr(project, "invariants", [])),
"invariant_actor": getattr(project, "invariant_actor", None),
"validations": 0, # per spec: project-scoped validations count
"created_at": project.created_at.isoformat()
if hasattr(project.created_at, "isoformat")
else str(project.created_at),
@@ -186,361 +188,14 @@ def _project_spec_dict(project: Any) -> dict[str, object]:
# ---------------------------------------------------------------------------
# Legacy init helpers (preserved)
# Register sub-apps and commands
# ---------------------------------------------------------------------------
def init_command(
name: str | None = None,
path: Path | None = None,
force: bool = False,
create_ignore_file: bool = False,
default_filters: bool = False,
yes: bool = False,
) -> None:
"""Programmatic interface for initializing a new CleverAgents project."""
from cleveragents.application.container import get_container
from cleveragents.application.services.project_service import ProjectService
if path is None:
path = Path.cwd()
project_name = name or path.name
container = get_container()
# When --yes is passed, disable the migration confirmation prompt
# entirely so the command never blocks waiting for interactive
# input (bug #783). ``require_confirmation=False`` tells the
# migration runner to skip the prompt; the ``prompt_for_migration``
# callback is set as a belt-and-suspenders fallback that
# auto-approves in case any code path still reaches the prompt.
if yes:
container.unit_of_work.add_kwargs(
require_confirmation=False,
prompt_for_migration=lambda _: True,
)
project_service: ProjectService = container.project_service()
project = project_service.initialize_project(
name=project_name,
path=path,
force=force,
create_ignore_file=create_ignore_file,
apply_default_filters=default_filters,
)
if yes:
# Non-interactive mode: produce spec-aligned output
# (specification.md lines 1381-1389)
data_dir: Path = getattr(
project, "data_dir", getattr(project, "path", path) / ".cleveragents"
)
config_path: Path = getattr(
project,
"config_path",
getattr(project, "path", path) / ".cleveragents" / "config.toml",
)
database_status: str = getattr(
project, "database_status", "initialized (schema v3)"
)
directories: list[str] = getattr(
project, "directories", ["logs", "cache", "sessions", "contexts"]
)
console.print(
Panel(
f"[green]Data Dir:[/green] {data_dir} (created)\n"
f"[green]Config:[/green] {config_path}\n"
f"[green]Database:[/green] {database_status}\n"
f"[green]Directories:[/green] {', '.join(directories)}",
title="Initialized",
expand=False,
)
)
console.print("[green]✓ OK[/green] Initialized (non-interactive)")
else:
console.print(
Panel(
f"[green]✓[/green] Project '{project.name}' "
f"initialized successfully!\n\n"
f"Location: {project.path / '.cleveragents'}\n"
f"Database: SQLite\n"
f"Status: Ready",
title="Project Initialized",
expand=False,
)
)
# ---------------------------------------------------------------------------
# File-filter sub-app (legacy, preserved as-is)
# ---------------------------------------------------------------------------
@file_filter_app.command("show")
def file_filter_show() -> None:
"""Show current include/exclude filters for this project."""
project_service, project = _get_project_service_and_current_project()
includes, excludes = project_service.get_project_filters(project)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Project File Filters",
expand=False,
)
)
@file_filter_app.command("add")
def file_filter_add(
include: Annotated[
list[str] | None,
typer.Option(
help="Include globs to add (repeatable)",
),
] = None,
exclude: Annotated[
list[str] | None,
typer.Option(
help="Exclude globs to add (repeatable)",
),
] = None,
defaults: Annotated[
bool,
typer.Option(
"--defaults",
help="Add the recommended default exclude globs",
),
] = False,
) -> None:
"""Add include/exclude globs to the project settings."""
project_service, project = _get_project_service_and_current_project()
include_list = [p for p in (include or []) if p]
exclude_list = [p for p in (exclude or []) if p]
if defaults:
exclude_list.extend(DEFAULT_IGNORE_PATTERNS)
updated = project_service.update_file_filters(
project,
include_add=include_list,
exclude_add=exclude_list if exclude_list else None,
)
includes, excludes = project_service.get_project_filters(updated)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Updated Project File Filters",
expand=False,
)
)
@file_filter_app.command("clear")
def file_filter_clear(
clear_include: Annotated[
bool,
typer.Option(help="Clear include globs"),
] = False,
clear_exclude: Annotated[
bool,
typer.Option(help="Clear exclude globs"),
] = False,
) -> None:
"""Clear include/exclude globs for the project."""
project_service, project = _get_project_service_and_current_project()
if not clear_include and not clear_exclude:
clear_include = True
clear_exclude = True
updated = project_service.update_file_filters(
project,
clear_include=clear_include,
clear_exclude=clear_exclude,
)
includes, excludes = project_service.get_project_filters(updated)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Cleared Project File Filters",
expand=False,
)
)
@file_filter_app.command("remove")
def file_filter_remove(
include: Annotated[
list[str] | None,
typer.Option(
help="Include glob(s) to remove (repeatable)",
),
] = None,
exclude: Annotated[
list[str] | None,
typer.Option(
help="Exclude glob(s) to remove (repeatable)",
),
] = None,
) -> None:
"""Remove include/exclude globs from the project settings."""
project_service, project = _get_project_service_and_current_project()
include_list = [p for p in (include or []) if p]
exclude_list = [p for p in (exclude or []) if p]
updated = project_service.update_file_filters(
project,
include_remove=include_list if include_list else None,
exclude_remove=exclude_list if exclude_list else None,
)
includes, excludes = project_service.get_project_filters(updated)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Updated Project File Filters",
expand=False,
)
)
app.add_typer(file_filter_app, name="file-filter")
app.add_typer(context_app, name="context")
# ---------------------------------------------------------------------------
# Legacy commands (init, status, clean - preserved for backward compat)
# ---------------------------------------------------------------------------
@app.command(name="init")
def init(
name: Annotated[
str | None,
typer.Argument(help="Project name (defaults to current directory name)"),
] = None,
path: Annotated[
Path | None,
typer.Option(
"--path",
"-p",
help="Project path (defaults to current directory)",
),
] = None,
force: Annotated[
bool, typer.Option("--force", "-f", help="Force reinitialization")
] = False,
create_ignore_file: Annotated[
bool,
typer.Option(
"--create-ignore-file",
help="Write a default .agentsignore with recommended patterns",
),
] = False,
default_filters: Annotated[
bool,
typer.Option(
"--default-filters",
help="Seed project exclude filters with recommended defaults",
),
] = False,
yes: Annotated[
bool,
typer.Option(
"--yes",
"-y",
help="Skip interactive prompts and use default values",
),
] = False,
) -> None:
"""Initialize a new CleverAgents project.
Creates the .cleveragents directory structure with database and configuration.
"""
try:
init_command(
name=name,
path=path,
force=force,
create_ignore_file=create_ignore_file,
default_filters=default_filters,
yes=yes,
)
except ValidationError as e:
from cleveragents.shared.redaction import redact_dict, redact_value
err_console.print(f"[red]Validation Error:[/red] {redact_value(e.message)}")
if e.details:
safe = redact_dict(e.details)
for key, value in safe.items():
err_console.print(f" {key}: {value}")
raise typer.Abort() from e
except ConfigurationError as e:
console.print(f"[red]Configuration Error:[/red] {e.message}")
raise typer.Abort() from e
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
except Exception as e:
console.print(f"[red]Unexpected error:[/red] {e}")
raise typer.Abort() from e
@app.command(name="status")
def status() -> None:
"""Show current project status and information."""
from cleveragents.application.container import get_container
from cleveragents.application.services.project_service import ProjectService
try:
container = get_container()
project_service: ProjectService = container.project_service()
# Get current project
project = project_service.get_current_project()
if not project:
err_console.print(
"[red]Error: No project found in current directory.[/red]"
)
console.print("Run 'cleveragents init' to initialize a project.")
raise typer.Exit(1)
# Get project stats
stats = project_service.get_project_stats(project)
# Display project information
info_text = f"""
[bold]Project:[/bold] {project.name}
[bold]Path:[/bold] {project.path}
[bold]Created:[/bold] {project.created_at}
[bold]Statistics:[/bold]
Plans: {stats.get("plans", 0)}
Context Files: {stats.get("context_files", 0)}
Total Changes: {stats.get("changes", 0)}
Current Plan: {stats.get("current_plan", "None")}
"""
console.print(Panel(info_text.strip(), title="Project Status", expand=False))
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
@app.command(name="clean")
def clean(
confirm: Annotated[
bool, typer.Option("--yes", "-y", help="Skip confirmation")
] = False,
) -> None:
"""Clean project cache and temporary files."""
console.print("[yellow]Project cleaning not yet implemented.[/yellow]")
raise typer.Abort()
register_legacy_commands(app)
register_show_command(app)
register_resource_commands(app)
# ---------------------------------------------------------------------------
@@ -603,7 +258,6 @@ def create(
err_console.print(f"[red]Error:[/red] {exc.message}")
raise typer.Exit(1) from exc
# Store invariants and invariant_actor if provided
if invariant or invariant_actor:
_store_project_extras(
project.namespaced_name,
@@ -611,7 +265,6 @@ def create(
inv_actor=invariant_actor,
)
# Link resources if specified
if resource:
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
@@ -628,7 +281,6 @@ def create(
f"'{res_name}': {exc}[/yellow]"
)
# Re-fetch project for display (includes linked resources)
try:
created = svc.get_project(project.namespaced_name)
except Exception:
@@ -636,6 +288,8 @@ def create(
data = _project_spec_dict(created)
if output_format.lower() == OutputFormat.RICH:
from rich.panel import Panel
console.print(
Panel(
f"[green]✓[/green] Project '{created.namespaced_name}' created.\n"
@@ -650,165 +304,6 @@ def create(
console.print(format_output(data, output_format))
@app.command(name="link-resource")
def link_resource(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
resource_name: Annotated[
str,
typer.Argument(help="Resource name or ULID to link"),
],
read_only: Annotated[
bool,
typer.Option("--read-only", help="Link as read-only"),
] = False,
alias: Annotated[
str | None,
typer.Option("--alias", help="Alias for the resource within the project"),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Link a resource to a project."""
svc = _get_namespaced_project_service()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
# Validate project exists
try:
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
# Resolve resource
try:
res = registry.show_resource(resource_name)
except NotFoundError as exc:
err_console.print(f"[red]Resource not found:[/red] {resource_name}")
raise typer.Exit(1) from exc
# Create link
try:
link_repo.create_link(
project_name=proj.namespaced_name,
resource_id=res.resource_id,
alias=alias,
read_only=read_only,
)
except DatabaseError as exc:
err_console.print(f"[red]Error linking resource:[/red] {exc.message}")
raise typer.Exit(1) from exc
if output_format.lower() == OutputFormat.RICH:
ro_label = " (read-only)" if read_only else ""
console.print(
f"[green]✓[/green] Linked resource '{resource_name}'{ro_label} "
f"to project '{project}'."
)
else:
console.print(
format_output(
{
"project": proj.namespaced_name,
"resource_id": res.resource_id,
"resource_name": res.name or resource_name,
"read_only": read_only,
"alias": alias,
},
output_format,
)
)
@app.command(name="unlink-resource")
def unlink_resource(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
resource_name: Annotated[
str,
typer.Argument(help="Resource name or ULID to unlink"),
],
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Unlink a resource from a project."""
svc = _get_namespaced_project_service()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
# Validate project exists
try:
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
# Resolve resource
try:
res = registry.show_resource(resource_name)
except NotFoundError as exc:
err_console.print(f"[red]Resource not found:[/red] {resource_name}")
raise typer.Exit(1) from exc
# Find the matching link
links = link_repo.list_links(proj.namespaced_name)
target_link: Any = None
for link in links:
if str(link.resource_id) == res.resource_id:
target_link = link
break
if target_link is None:
err_console.print(
f"[red]Resource '{resource_name}' is not linked to "
f"project '{project}'.[/red]"
)
raise typer.Exit(1)
if not yes:
confirm = typer.confirm(
f"Unlink resource '{resource_name}' from project '{project}'?"
)
if not confirm:
raise typer.Abort()
try:
link_repo.remove_link(str(target_link.link_id))
except DatabaseError as exc:
err_console.print(f"[red]Error unlinking resource:[/red] {exc.message}")
raise typer.Exit(1) from exc
if output_format.lower() == OutputFormat.RICH:
console.print(
f"[green]✓[/green] Unlinked resource '{resource_name}' "
f"from project '{project}'."
)
else:
console.print(
format_output(
{
"project": proj.namespaced_name,
"resource_id": res.resource_id,
"unlinked": True,
},
output_format,
)
)
@app.command(name="list")
def list_projects(
namespace: Annotated[
@@ -833,7 +328,6 @@ def list_projects(
err_console.print(f"[red]Error listing projects:[/red] {exc.message}")
raise typer.Exit(1) from exc
# Apply regex filter if provided
if regex:
try:
pattern = re.compile(regex)
@@ -872,123 +366,3 @@ def list_projects(
else:
data = [_project_spec_dict(p) for p in projects]
console.print(format_output(data, output_format))
@app.command(name="show")
def show(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Show details of a project."""
svc = _get_namespaced_project_service()
try:
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
data = _project_spec_dict(proj)
if output_format.lower() == OutputFormat.RICH:
lines: list[str] = [
f"[bold]Name:[/bold] {proj.namespaced_name}",
f"[bold]Namespace:[/bold] {proj.namespace}",
f"[bold]Description:[/bold] {proj.description or '(none)'}",
f"[bold]Created:[/bold] {proj.created_at}",
f"[bold]Updated:[/bold] {proj.updated_at}",
]
if proj.linked_resources:
lines.append(
f"\n[bold]Linked Resources ({len(proj.linked_resources)}):[/bold]"
)
for lr in proj.linked_resources:
ro_marker = " [dim](read-only)[/dim]" if lr.project_read_only else ""
alias_marker = f" alias={lr.alias}" if lr.alias else ""
lines.append(f" - {lr.resource_id}{ro_marker}{alias_marker}")
else:
lines.append("\n[bold]Linked Resources:[/bold] (none)")
console.print(
Panel(
"\n".join(lines),
title=f"Project: {proj.namespaced_name}",
expand=False,
)
)
else:
console.print(format_output(data, output_format))
@app.command(name="delete")
def delete(
name: Annotated[
str,
typer.Argument(help="Project namespaced name to delete"),
],
force: Annotated[
bool,
typer.Option("--force", "-f", help="Force delete even if resources are linked"),
] = False,
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
output_format: Annotated[
str,
typer.Option("--format", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Delete a project from the registry."""
svc = _get_namespaced_project_service()
# Validate project exists
try:
proj = svc.get_project(name)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {name}")
raise typer.Exit(1) from exc
# Check for linked resources (unless --force)
if proj.linked_resources and not force:
err_console.print(
f"[red]Project '{name}' has {len(proj.linked_resources)} "
f"linked resource(s). Use --force to delete anyway.[/red]"
)
raise typer.Exit(1)
if not yes:
confirm = typer.confirm(f"Delete project '{name}'?")
if not confirm:
raise typer.Abort()
try:
deleted = svc.delete_project(name)
except DatabaseError as exc:
err_console.print(f"[red]Error deleting project:[/red] {exc.message}")
raise typer.Exit(1) from exc
if not deleted:
err_console.print(f"[red]Project '{name}' could not be deleted.[/red]")
raise typer.Exit(1)
if output_format.lower() == OutputFormat.RICH:
console.print(f"[green]✓[/green] Project '{name}' deleted.")
else:
console.print(
format_output(
{
"deleted": name,
"success": True,
"deleted_at": datetime.now(tz=UTC),
},
output_format,
)
)
@@ -0,0 +1,401 @@
"""Legacy project commands and file-filter sub-app.
Extracted from ``project.py`` to keep that module under 500 lines.
These commands are preserved for backward compatibility.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any
import typer
Outdated
Review

BLOCKING — Import ordering violation (ruff I001)

The if TYPE_CHECKING: block appears between stdlib imports and third-party imports. This violates ruff isort rules and is almost certainly the cause of the lint CI failure.

Current ordering:

from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any

if TYPE_CHECKING:  # ← placed here, BEFORE third-party imports
    from cleveragents.application.services.project_service import ProjectService

import typer           # ← third-party
from rich.panel import Panel  # ← third-party

Correct ordering (move if TYPE_CHECKING: after all stdlib/third-party imports):

from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any

import typer
from rich.panel import Panel

from cleveragents.application.services.context_service import DEFAULT_IGNORE_PATTERNS
from cleveragents.cli.renderers import _get_console, _get_err_console
from cleveragents.core.exceptions import (
    CleverAgentsError,
    ConfigurationError,
    ValidationError,
)

if TYPE_CHECKING:
    from cleveragents.application.services.project_service import ProjectService

Fix: run ruff check --fix src/cleveragents/cli/commands/project_legacy.py && ruff format src/cleveragents/cli/commands/project_legacy.py


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING — Import ordering violation (ruff I001)** The `if TYPE_CHECKING:` block appears between stdlib imports and third-party imports. This violates ruff isort rules and is almost certainly the cause of the lint CI failure. Current ordering: ```python from pathlib import Path from typing import TYPE_CHECKING, Annotated, Any if TYPE_CHECKING: # ← placed here, BEFORE third-party imports from cleveragents.application.services.project_service import ProjectService import typer # ← third-party from rich.panel import Panel # ← third-party ``` Correct ordering (move `if TYPE_CHECKING:` after all stdlib/third-party imports): ```python from pathlib import Path from typing import TYPE_CHECKING, Annotated, Any import typer from rich.panel import Panel from cleveragents.application.services.context_service import DEFAULT_IGNORE_PATTERNS from cleveragents.cli.renderers import _get_console, _get_err_console from cleveragents.core.exceptions import ( CleverAgentsError, ConfigurationError, ValidationError, ) if TYPE_CHECKING: from cleveragents.application.services.project_service import ProjectService ``` Fix: run `ruff check --fix src/cleveragents/cli/commands/project_legacy.py && ruff format src/cleveragents/cli/commands/project_legacy.py` --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
Outdated
Review

BLOCKING — Import ordering violation (ruff I001) — STILL NOT FIXED

This is the same violation from Round 7 (review ID 7645). The if TYPE_CHECKING: block appears at line 12, between stdlib imports and third-party imports (import typer at line 15). This violates ruff isort rules and is the direct cause of the lint CI failure.

The new commit 8a2d884a did not fix this — it only modified database/models.py.

Fix: move if TYPE_CHECKING: to after ALL third-party and first-party imports.

Fix command: ruff check --fix src/cleveragents/cli/commands/project_legacy.py && ruff format src/cleveragents/cli/commands/project_legacy.py


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING — Import ordering violation (ruff I001) — STILL NOT FIXED** This is the same violation from Round 7 (review ID 7645). The `if TYPE_CHECKING:` block appears at line 12, between stdlib imports and third-party imports (`import typer` at line 15). This violates ruff isort rules and is the direct cause of the lint CI failure. The new commit `8a2d884a` did not fix this — it only modified `database/models.py`. Fix: move `if TYPE_CHECKING:` to after ALL third-party and first-party imports. Fix command: `ruff check --fix src/cleveragents/cli/commands/project_legacy.py && ruff format src/cleveragents/cli/commands/project_legacy.py` --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
Outdated
Review

BLOCKING — Import ordering violation (ruff I001) — 3rd consecutive round unfixed

The if TYPE_CHECKING: block (lines 12–13) appears BEFORE the third-party imports import typer (line 15) and from rich.panel import Panel (line 16). This violates ruff isort rule I001.

Per isort rules, if TYPE_CHECKING: blocks must appear AFTER all unconditional stdlib, third-party, and first-party imports.

Current (wrong):

from typing import TYPE_CHECKING, Annotated, Any

if TYPE_CHECKING:          # ← before third-party imports
    from ...project_service import ProjectService

import typer
from rich.panel import Panel

Required fix:

from typing import TYPE_CHECKING, Annotated, Any

import typer
from rich.panel import Panel

from cleveragents...  # first-party imports

if TYPE_CHECKING:          # ← after ALL unconditional imports
    from ...project_service import ProjectService

Run: ruff check --select I001 --fix src/cleveragents/cli/commands/project_legacy.py

**BLOCKING — Import ordering violation (ruff I001) — 3rd consecutive round unfixed** The `if TYPE_CHECKING:` block (lines 12–13) appears BEFORE the third-party imports `import typer` (line 15) and `from rich.panel import Panel` (line 16). This violates ruff isort rule I001. Per isort rules, `if TYPE_CHECKING:` blocks must appear AFTER all unconditional stdlib, third-party, and first-party imports. Current (wrong): ```python from typing import TYPE_CHECKING, Annotated, Any if TYPE_CHECKING: # ← before third-party imports from ...project_service import ProjectService import typer from rich.panel import Panel ``` Required fix: ```python from typing import TYPE_CHECKING, Annotated, Any import typer from rich.panel import Panel from cleveragents... # first-party imports if TYPE_CHECKING: # ← after ALL unconditional imports from ...project_service import ProjectService ``` Run: `ruff check --select I001 --fix src/cleveragents/cli/commands/project_legacy.py`
from rich.panel import Panel
Outdated
Review

BLOCKING — Import ordering violation (ruff I001) — UNADDRESSED from Round 7

This violation was identified in Round 7 review (ID 7645) and remains unfixed in the current head commit. The if TYPE_CHECKING: block appears BEFORE third-party imports (typer, rich.panel.Panel), violating ruff isort rule I001.

Current (wrong):

from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any

if TYPE_CHECKING:   # ← BEFORE third-party imports — WRONG
    from cleveragents.application.services.project_service import ProjectService

import typer
from rich.panel import Panel

Fix by moving if TYPE_CHECKING: to after ALL stdlib and third-party imports, then run:

ruff check --fix src/cleveragents/cli/commands/project_legacy.py
ruff format src/cleveragents/cli/commands/project_legacy.py

This is the root cause of the lint CI failure.


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING — Import ordering violation (ruff I001) — UNADDRESSED from Round 7** This violation was identified in Round 7 review (ID 7645) and remains unfixed in the current head commit. The `if TYPE_CHECKING:` block appears BEFORE third-party imports (`typer`, `rich.panel.Panel`), violating ruff isort rule I001. Current (wrong): ```python from pathlib import Path from typing import TYPE_CHECKING, Annotated, Any if TYPE_CHECKING: # ← BEFORE third-party imports — WRONG from cleveragents.application.services.project_service import ProjectService import typer from rich.panel import Panel ``` Fix by moving `if TYPE_CHECKING:` to after ALL stdlib and third-party imports, then run: ``` ruff check --fix src/cleveragents/cli/commands/project_legacy.py ruff format src/cleveragents/cli/commands/project_legacy.py ``` This is the root cause of the lint CI failure. --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
from cleveragents.application.services.context_service import DEFAULT_IGNORE_PATTERNS
from cleveragents.cli.renderers import _get_console, _get_err_console
from cleveragents.core.exceptions import (
CleverAgentsError,
ConfigurationError,
ValidationError,
)
if TYPE_CHECKING:
from cleveragents.application.services.project_service import ProjectService
console = _get_console()
err_console = _get_err_console()
file_filter_app = typer.Typer(
help="Manage project include/exclude filters", name="file-filter"
)
# ---------------------------------------------------------------------------
# Legacy init helpers
# ---------------------------------------------------------------------------
def init_command(
name: str | None = None,
path: Path | None = None,
force: bool = False,
create_ignore_file: bool = False,
default_filters: bool = False,
yes: bool = False,
) -> None:
"""Programmatic interface for initializing a new CleverAgents project."""
from cleveragents.application.container import get_container
if path is None:
path = Path.cwd()
project_name = name or path.name
container = get_container()
# When --yes is passed, disable the migration confirmation prompt
# entirely so the command never blocks waiting for interactive
# input (bug #783). ``require_confirmation=False`` tells the
# migration runner to skip the prompt; the ``prompt_for_migration``
# callback is set as a belt-and-suspenders fallback that
# auto-approves in case any code path still reaches the prompt.
if yes:
container.unit_of_work.add_kwargs(
require_confirmation=False,
prompt_for_migration=lambda _: True,
)
project_service = container.project_service()
project = project_service.initialize_project(
name=project_name,
path=path,
force=force,
create_ignore_file=create_ignore_file,
apply_default_filters=default_filters,
)
if yes:
# Non-interactive mode: produce spec-aligned output
# (specification.md lines 1381-1389)
data_dir: Path = getattr(
project, "data_dir", getattr(project, "path", path) / ".cleveragents"
)
config_path: Path = getattr(
project,
"config_path",
getattr(project, "path", path) / ".cleveragents" / "config.toml",
)
database_status: str = getattr(
project, "database_status", "initialized (schema v3)"
)
directories: list[str] = getattr(
project, "directories", ["logs", "cache", "sessions", "contexts"]
)
console.print(
Panel(
f"[green]Data Dir:[/green] {data_dir} (created)\n"
f"[green]Config:[/green] {config_path}\n"
f"[green]Database:[/green] {database_status}\n"
f"[green]Directories:[/green] {', '.join(directories)}",
title="Initialized",
expand=False,
)
)
console.print("[green]✓ OK[/green] Initialized (non-interactive)")
else:
console.print(
Panel(
f"[green]✓[/green] Project '{project.name}' "
f"initialized successfully!\n\n"
f"Location: {project.path / '.cleveragents'}\n"
f"Database: SQLite\n"
f"Status: Ready",
title="Project Initialized",
expand=False,
)
)
# ---------------------------------------------------------------------------
# File-filter sub-app (legacy, preserved as-is)
# ---------------------------------------------------------------------------
def _get_project_service_and_current_project_legacy() -> tuple[ProjectService, Any]:
from cleveragents.application.container import get_container
from cleveragents.application.services.project_service import (
Outdated
Review

BLOCKING — ruff format failure (formatting nit introduced by this PR)

ruff format --check reports this file would be reformatted. There are two formatting issues that need to be fixed:

Issue A — Parenthesised return type (lines 127–129):

# Current:
def _get_project_service_and_current_project_legacy() -> (
    tuple[ProjectService, Any]
):

# ruff format requires:
def _get_project_service_and_current_project_legacy() -> tuple[ProjectService, Any]:

Issue B — Long err_console.print() call (~line 336):

# Current:
            err_console.print(
                f"[red]Validation Error:[/red] {redact_value(e.message)}"
            )

# ruff format requires (fits on one line):
            err_console.print(f"[red]Validation Error:[/red] {redact_value(e.message)}")

Fix: Run ruff format src/cleveragents/cli/commands/project_legacy.py — one command, auto-applies both changes.

This is the only CI gate failing due to code introduced by this PR. Everything else passes.


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING — `ruff format` failure (formatting nit introduced by this PR)** `ruff format --check` reports this file would be reformatted. There are two formatting issues that need to be fixed: **Issue A — Parenthesised return type (lines 127–129):** ```python # Current: def _get_project_service_and_current_project_legacy() -> ( tuple[ProjectService, Any] ): # ruff format requires: def _get_project_service_and_current_project_legacy() -> tuple[ProjectService, Any]: ``` **Issue B — Long `err_console.print()` call (~line 336):** ```python # Current: err_console.print( f"[red]Validation Error:[/red] {redact_value(e.message)}" ) # ruff format requires (fits on one line): err_console.print(f"[red]Validation Error:[/red] {redact_value(e.message)}") ``` **Fix:** Run `ruff format src/cleveragents/cli/commands/project_legacy.py` — one command, auto-applies both changes. This is the **only CI gate failing due to code introduced by this PR**. Everything else passes. --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
ProjectService as _ProjectService,
)
container = get_container()
project_service: _ProjectService = container.project_service()
project = project_service.get_current_project()
if not project:
err_console.print("[red]Error: No project found in current directory.[/red]")
raise typer.Exit(1)
return project_service, project
@file_filter_app.command("show")
def file_filter_show() -> None:
"""Show current include/exclude filters for this project."""
project_service, project = _get_project_service_and_current_project_legacy()
includes, excludes = project_service.get_project_filters(project)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Project File Filters",
expand=False,
)
)
@file_filter_app.command("add")
def file_filter_add(
include: Annotated[
list[str] | None,
typer.Option(
help="Include globs to add (repeatable)",
),
] = None,
exclude: Annotated[
list[str] | None,
typer.Option(
help="Exclude globs to add (repeatable)",
),
] = None,
defaults: Annotated[
bool,
typer.Option(
"--defaults",
help="Add the recommended default exclude globs",
),
] = False,
) -> None:
"""Add include/exclude globs to the project settings."""
project_service, project = _get_project_service_and_current_project_legacy()
include_list = [p for p in (include or []) if p]
exclude_list = [p for p in (exclude or []) if p]
if defaults:
exclude_list.extend(DEFAULT_IGNORE_PATTERNS)
updated = project_service.update_file_filters(
project,
include_add=include_list,
exclude_add=exclude_list if exclude_list else None,
)
includes, excludes = project_service.get_project_filters(updated)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Updated Project File Filters",
expand=False,
)
)
@file_filter_app.command("clear")
def file_filter_clear(
clear_include: Annotated[
bool,
typer.Option(help="Clear include globs"),
] = False,
clear_exclude: Annotated[
bool,
typer.Option(help="Clear exclude globs"),
] = False,
) -> None:
"""Clear include/exclude globs for the project."""
project_service, project = _get_project_service_and_current_project_legacy()
if not clear_include and not clear_exclude:
clear_include = True
clear_exclude = True
updated = project_service.update_file_filters(
project,
clear_include=clear_include,
clear_exclude=clear_exclude,
)
includes, excludes = project_service.get_project_filters(updated)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Cleared Project File Filters",
expand=False,
)
)
@file_filter_app.command("remove")
def file_filter_remove(
include: Annotated[
list[str] | None,
typer.Option(
help="Include glob(s) to remove (repeatable)",
),
] = None,
exclude: Annotated[
list[str] | None,
typer.Option(
help="Exclude glob(s) to remove (repeatable)",
),
] = None,
) -> None:
"""Remove include/exclude globs from the project settings."""
project_service, project = _get_project_service_and_current_project_legacy()
include_list = [p for p in (include or []) if p]
exclude_list = [p for p in (exclude or []) if p]
updated = project_service.update_file_filters(
project,
include_remove=include_list if include_list else None,
exclude_remove=exclude_list if exclude_list else None,
)
includes, excludes = project_service.get_project_filters(updated)
console.print(
Panel(
f"Includes: {includes or '[]'}\nExcludes: {excludes or '[]'}",
title="Updated Project File Filters",
expand=False,
)
)
# ---------------------------------------------------------------------------
# Legacy commands (init, status, clean)
# ---------------------------------------------------------------------------
def register_legacy_commands(app: typer.Typer) -> None:
"""Register legacy commands on *app*."""
@app.command(name="init")
def init(
name: Annotated[
str | None,
typer.Argument(help="Project name (defaults to current directory name)"),
] = None,
path: Annotated[
Path | None,
typer.Option(
"--path",
"-p",
help="Project path (defaults to current directory)",
),
] = None,
force: Annotated[
bool, typer.Option("--force", "-f", help="Force reinitialization")
] = False,
create_ignore_file: Annotated[
bool,
typer.Option(
"--create-ignore-file",
help="Write a default .agentsignore with recommended patterns",
),
] = False,
default_filters: Annotated[
bool,
typer.Option(
"--default-filters",
help="Seed project exclude filters with recommended defaults",
),
] = False,
yes: Annotated[
bool,
typer.Option(
"--yes",
"-y",
help="Skip interactive prompts and use default values",
),
] = False,
) -> None:
"""Initialize a new CleverAgents project.
Creates the .cleveragents directory structure with database and configuration.
"""
try:
init_command(
name=name,
path=path,
force=force,
create_ignore_file=create_ignore_file,
default_filters=default_filters,
yes=yes,
)
except ValidationError as e:
from cleveragents.shared.redaction import redact_dict, redact_value
err_console.print(f"[red]Validation Error:[/red] {redact_value(e.message)}")
if e.details:
safe = redact_dict(e.details)
for key, value in safe.items():
err_console.print(f" {key}: {value}")
raise typer.Abort() from e
except ConfigurationError as e:
console.print(f"[red]Configuration Error:[/red] {e.message}")
raise typer.Abort() from e
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
except Exception as e:
console.print(f"[red]Unexpected error:[/red] {e}")
raise typer.Abort() from e
@app.command(name="status")
def status() -> None:
"""Show current project status and information."""
from cleveragents.application.container import get_container
try:
container = get_container()
project_service = container.project_service()
# Get current project
project = project_service.get_current_project()
if not project:
err_console.print(
"[red]Error: No project found in current directory.[/red]"
)
console.print("Run 'cleveragents init' to initialize a project.")
raise typer.Exit(1)
# Get project stats
stats = project_service.get_project_stats(project)
# Display project information
info_text = f"""
[bold]Project:[/bold] {project.name}
[bold]Path:[/bold] {project.path}
[bold]Created:[/bold] {project.created_at}
[bold]Statistics:[/bold]
Plans: {stats.get("plans", 0)}
Context Files: {stats.get("context_files", 0)}
Total Changes: {stats.get("changes", 0)}
Current Plan: {stats.get("current_plan", "None")}
"""
console.print(
Panel(info_text.strip(), title="Project Status", expand=False)
)
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
@app.command(name="clean")
def clean(
confirm: Annotated[
bool, typer.Option("--yes", "-y", help="Skip confirmation")
] = False,
) -> None:
"""Clean project cache and temporary files."""
console.print("[yellow]Project cleaning not yet implemented.[/yellow]")
raise typer.Abort()
@@ -0,0 +1,255 @@
"""Resource link/unlink and delete commands for the project CLI.
Extracted from ``project.py`` to keep that module under 500 lines.
"""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Annotated, Any
import typer
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.cli.renderers import _get_console, _get_err_console
from cleveragents.core.exceptions import DatabaseError, NotFoundError
console = _get_console()
err_console = _get_err_console()
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
def register_resource_commands(app: typer.Typer) -> None:
"""Register link-resource, unlink-resource, and delete commands on *app*."""
@app.command(name="link-resource")
def link_resource(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
resource_name: Annotated[
str,
typer.Argument(help="Resource name or ULID to link"),
],
read_only: Annotated[
bool,
typer.Option("--read-only", help="Link as read-only"),
] = False,
alias: Annotated[
str | None,
typer.Option("--alias", help="Alias for the resource within the project"),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Link a resource to a project."""
from cleveragents.cli.commands.project import (
_get_namespaced_project_service,
_get_resource_link_repo,
_get_resource_registry_service,
)
svc = _get_namespaced_project_service()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
try:
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
try:
res = registry.show_resource(resource_name)
except NotFoundError as exc:
err_console.print(f"[red]Resource not found:[/red] {resource_name}")
raise typer.Exit(1) from exc
try:
link_repo.create_link(
project_name=proj.namespaced_name,
resource_id=res.resource_id,
alias=alias,
read_only=read_only,
)
except DatabaseError as exc:
err_console.print(f"[red]Error linking resource:[/red] {exc.message}")
raise typer.Exit(1) from exc
if output_format.lower() == OutputFormat.RICH:
ro_label = " (read-only)" if read_only else ""
console.print(
f"[green]✓[/green] Linked resource '{resource_name}'{ro_label} "
f"to project '{project}'."
)
else:
console.print(
format_output(
{
"project": proj.namespaced_name,
"resource_id": res.resource_id,
"resource_name": res.name or resource_name,
"read_only": read_only,
"alias": alias,
},
output_format,
)
)
@app.command(name="unlink-resource")
def unlink_resource(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
resource_name: Annotated[
str,
typer.Argument(help="Resource name or ULID to unlink"),
],
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Unlink a resource from a project."""
from cleveragents.cli.commands.project import (
_get_namespaced_project_service,
_get_resource_link_repo,
_get_resource_registry_service,
)
svc = _get_namespaced_project_service()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
try:
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
try:
res = registry.show_resource(resource_name)
except NotFoundError as exc:
err_console.print(f"[red]Resource not found:[/red] {resource_name}")
raise typer.Exit(1) from exc
links = link_repo.list_links(proj.namespaced_name)
target_link: Any = None
for link in links:
if str(link.resource_id) == res.resource_id:
target_link = link
break
if target_link is None:
err_console.print(
f"[red]Resource '{resource_name}' is not linked to "
f"project '{project}'.[/red]"
)
raise typer.Exit(1)
if not yes:
confirm = typer.confirm(
f"Unlink resource '{resource_name}' from project '{project}'?"
)
if not confirm:
raise typer.Abort()
try:
link_repo.remove_link(str(target_link.link_id))
except DatabaseError as exc:
err_console.print(f"[red]Error unlinking resource:[/red] {exc.message}")
raise typer.Exit(1) from exc
if output_format.lower() == OutputFormat.RICH:
console.print(
f"[green]✓[/green] Unlinked resource '{resource_name}' "
f"from project '{project}'."
)
else:
console.print(
format_output(
{
"project": proj.namespaced_name,
"resource_id": res.resource_id,
"unlinked": True,
},
output_format,
)
)
@app.command(name="delete")
def delete(
name: Annotated[
str,
typer.Argument(help="Project namespaced name to delete"),
],
force: Annotated[
bool,
typer.Option(
"--force", "-f", help="Force delete even if resources are linked"
),
] = False,
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
output_format: Annotated[
str,
typer.Option("--format", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Delete a project from the registry."""
from cleveragents.cli.commands.project import _get_namespaced_project_service
svc = _get_namespaced_project_service()
try:
proj = svc.get_project(name)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {name}")
raise typer.Exit(1) from exc
if proj.linked_resources and not force:
err_console.print(
f"[red]Project '{name}' has {len(proj.linked_resources)} "
f"linked resource(s). Use --force to delete anyway.[/red]"
)
raise typer.Exit(1)
if not yes:
confirm = typer.confirm(f"Delete project '{name}'?")
if not confirm:
raise typer.Abort()
try:
deleted = svc.delete_project(name)
except DatabaseError as exc:
err_console.print(f"[red]Error deleting project:[/red] {exc.message}")
raise typer.Exit(1) from exc
if not deleted:
err_console.print(f"[red]Project '{name}' could not be deleted.[/red]")
raise typer.Exit(1)
if output_format.lower() == OutputFormat.RICH:
console.print(f"[green]✓[/green] Project '{name}' deleted.")
else:
console.print(
format_output(
{
"deleted": name,
"success": True,
"deleted_at": datetime.now(tz=UTC),
},
output_format,
)
)
@@ -0,0 +1,204 @@
"""Show command and display helpers for the project CLI.
Extracted from ``project.py`` to keep that module under 500 lines.
"""
from __future__ import annotations
from typing import Annotated, Any
import typer
from rich.panel import Panel
from rich.table import Table
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.cli.renderers import _get_console, _get_err_console
from cleveragents.domain.models.core.invariant import InvariantScope
console = _get_console()
err_console = _get_err_console()
def _get_project_invariants(project: Any) -> list[dict[str, Any]]:
"""Fetch project-scoped invariants for display."""
from cleveragents.application.container import get_container
from cleveragents.application.services.invariant_service import (
InvariantService,
)
container = get_container()
try:
svc: InvariantService = container.invariant_service()
invariants = svc.list_invariants(
scope=InvariantScope.PROJECT,
source_name=project.namespaced_name,
)
return [
{
"text": inv.text,
"source": inv.source_name,
"scope": inv.scope.value,
}
for inv in invariants
]
except Exception:
return []
def _get_project_validations(project: Any) -> list[dict[str, Any]]:
"""Fetch validation attachments for a project.
Collects all validation attachments scoped to this project by querying
each linked resource's validation attachments and filtering by project name.
"""
from cleveragents.application.container import get_container
from cleveragents.application.services.tool_registry_service import (
ToolRegistryService,
)
container = get_container()
try:
svc: ToolRegistryService = container.tool_registry_service()
attachments: list[dict[str, Any]] = []
for lr in getattr(project, "linked_resources", []):
resource_id = getattr(lr, "resource_id", None)
if not resource_id:
continue
resource_attachments = svc.list_validations_for_resource(
resource_id=resource_id,
project_name=project.namespaced_name,
)
for att in resource_attachments:
if isinstance(att, dict):
attachments.append(att)
else:
attachments.append(
{
"validation_name": getattr(att, "validation_name", ""),
"resource_id": getattr(att, "resource_id", ""),
"mode": getattr(att, "mode", "required"),
}
)
return attachments
except Exception:
return []
def register_show_command(app: typer.Typer) -> None:
"""Register the ``show`` command on *app*."""
@app.command(name="show")
def show(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
output_format: Annotated[
str,
typer.Option(
"--format",
"-f",
help="Output format: json, yaml, plain, table, or rich (default: rich)",
),
] = "rich",
) -> None:
"""Show details of a project."""
from cleveragents.cli.commands.project import (
_get_namespaced_project_service,
_project_spec_dict,
)
svc = _get_namespaced_project_service()
try:
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
data = _project_spec_dict(proj)
if output_format.lower() == OutputFormat.RICH:
lines: list[str] = [
f"[bold]Name:[/bold] {proj.namespaced_name}",
f"[bold]Namespace:[/bold] {proj.namespace}",
f"[bold]Description:[/bold] {proj.description or '(none)'}",
f"[bold]Created:[/bold] {proj.created_at}",
f"[bold]Updated:[/bold] {proj.updated_at}",
]
if proj.linked_resources:
lines.append(
f"\n[bold]Linked Resources ({len(proj.linked_resources)}):[/bold]"
)
for lr in proj.linked_resources:
ro_marker = (
" [dim](read-only)[/dim]" if lr.project_read_only else ""
)
alias_marker = f" alias={lr.alias}" if lr.alias else ""
lines.append(f" - {lr.resource_id}{ro_marker}{alias_marker}")
else:
lines.append("\n[bold]Linked Resources:[/bold] (none)")
console.print(
Panel(
"\n".join(lines),
title=f"Project: {proj.namespaced_name}",
expand=False,
)
)
# --- Invariants Panel ---
invs = _get_project_invariants(proj)
if invs:
inv_table = Table(title="Invariants", show_header=True, expand=True)
inv_table.add_column("#", style="dim", justify="right", width=4)
inv_table.add_column("Invariant", style="bold")
inv_table.add_column("Source")
for i, inv_data in enumerate(invs, start=1):
inv_table.add_row(
str(i),
inv_data.get("text", ""),
inv_data.get("source", ""),
)
console.print(inv_table)
else:
console.print(
Panel(
"[dim]No project-level invariants defined.[/dim]",
title="Invariants",
expand=False,
)
)
# --- Validations Panel ---
val_attachments = _get_project_validations(proj)
if val_attachments:
val_table = Table(title="Validations", show_header=True, expand=True)
val_table.add_column("#", style="dim", justify="right", width=4)
val_table.add_column("Validation")
val_table.add_column("Resource")
val_table.add_column("Mode")
for i, att in enumerate(val_attachments, start=1):
if isinstance(att, dict):
val_name = att.get("validation_name", "")
res_id = att.get("resource_id", "")
mode = att.get("mode", "required")
else:
val_name = getattr(att, "validation_name", "")
res_id = getattr(att, "resource_id", "")
mode = getattr(att, "mode", "required")
val_table.add_row(str(i), str(val_name), str(res_id), str(mode))
console.print(val_table)
else:
console.print(
Panel(
"[dim]No validation attachments for this project.[/dim]",
title="Validations",
expand=False,
)
)
else:
data.setdefault("invariants", _get_project_invariants(proj))
data.setdefault("validations", _get_project_validations(proj))
console.print(format_output(data, output_format))
+3 -1
View File
@@ -563,7 +563,9 @@ def init(
] = False,
) -> None:
"""Initialize a new CleverAgents project in the current directory."""
from cleveragents.cli.commands.project import init_command as project_init_command
from cleveragents.cli.commands.project_legacy import (
init_command as project_init_command,
)
try:
project_init_command(
@@ -373,6 +373,18 @@ class NamespacedProject(BaseModel):
description="Resources linked to this project from the Resource Registry",
)
# Invariants (spec section: project-level invariants)
invariants: list[str] = Field(
default_factory=list,
description="Project-level invariants that must be maintained",
)
# Invariant Actor (spec section: actor for invariant reconciliation)
invariant_actor: str | None = Field(
default=None,
description="Actor responsible for invariant reconciliation",
)
# Context configuration (spec section 3)
context_config: ContextConfig = Field(
default_factory=ContextConfig,
1
@@ -1404,11 +1404,24 @@ class NamespacedProjectModel(Base): # type: ignore[misc]
parts = ns_name.split("/", 1)
short_name = parts[1] if len(parts) > 1 else parts[0]
# Parse invariants from JSON
invariants: list[str] = []
raw_invariants = cast("str | None", self.invariants_json)
if raw_invariants:
try:
inv_list = json.loads(raw_invariants)
if isinstance(inv_list, list):
invariants = [str(i) for i in inv_list if i]
except (ValueError, TypeError):
pass
return NamespacedProject(
name=short_name,
namespace=cast(str, self.namespace),
description=cast("str | None", self.description),
linked_resources=linked_resources,
invariants=invariants,
invariant_actor=cast("str | None", self.invariant_actor),
context_config=context_config,
created_at=datetime.fromisoformat(cast(str, self.created_at)),
updated_at=datetime.fromisoformat(cast(str, self.updated_at)),
@@ -1436,9 +1449,9 @@ class NamespacedProjectModel(Base): # type: ignore[misc]
namespaced_name=project.namespaced_name,
namespace=project.namespace,
description=project.description,
invariants_json=json.dumps([]),
invariants_json=json.dumps(getattr(project, "invariants", []) or []),
automation_profile=None,
invariant_actor=None,
invariant_actor=getattr(project, "invariant_actor", None),
context_policy_json=context_policy_json,
tags_json=tags_json_str,
created_by=None,