fix(cli): add missing resource command flags per specification #1192

Merged
freemo merged 2 commits from feature/m4-resource-flags into master 2026-04-02 08:31:27 +00:00
9 changed files with 802 additions and 12 deletions
+108
View File
@@ -0,0 +1,108 @@
Feature: Resource and LSP CLI missing flags (issue #904)
As a CleverAgents user
I want the CLI to support all specification-defined flags
So that I can fully manage resources and LSP servers per the spec
# ---- resource add --update ----
Scenario: Add a resource with --update re-registers it
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
And I add a resource flags resource "git-checkout" named "local/upd-test" with path "/tmp/u1"
When I run resource flags add "git-checkout" "local/upd-test" with path "/tmp/u2" and update
Then the resource flags output should contain "Added resource"
And the resource flags output should contain "local/upd-test"
Scenario: Add with --update when resource does not yet exist
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
When I run resource flags add "git-checkout" "local/new-upd" with path "/tmp/nu" and update
Then the resource flags output should contain "Added resource"
# ---- resource add --mount (container types) ----
Scenario: Mount flag rejected for non-container type
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
When I run resource flags add "git-checkout" "local/bad-mount" with path "/tmp/bm" and mount "local/repo:/workspace"
Then the resource flags command should fail
And the resource flags output should contain "container types"
# ---- resource add --clone-into ----
Scenario: Clone-into flag rejected for non-container type
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
When I run resource flags add "git-checkout" "local/bad-clone" with path "/tmp/bc" and clone-into "https://example.com/repo.git:/workspace"
Then the resource flags command should fail
And the resource flags output should contain "container types"
Scenario: Clone-into flag with invalid format
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
When I run resource flags add "container-instance" "local/bad-fmt" with clone-into "no-colon-here"
Then the resource flags command should fail
And the resource flags output should contain "REPO_URL"
# ---- resource list --all ----
Scenario: Resource list default excludes auto-discovered resources
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
And a user-added resource flags resource "git-checkout" named "local/user-res" with path "/tmp/ur"
And an auto-discovered resource flags resource named "local/auto-res"
When I run resource flags list
Then the resource flags output should contain "local/user-res"
And the resource flags output should not contain "local/auto-res"
Scenario: Resource list --all includes auto-discovered resources
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
And a user-added resource flags resource "git-checkout" named "local/user-res2" with path "/tmp/ur2"
And an auto-discovered resource flags resource named "local/auto-res2"
When I run resource flags list with --all
Then the resource flags output should contain "local/user-res2"
And the resource flags output should contain "local/auto-res2"
# ---- resource tree --depth default ----
Scenario: Tree depth defaults to 3
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
And a resource flags resource "git-checkout" named "local/depth-root" with path "/tmp/dr"
When I run resource flags tree for "local/depth-root" with default depth
Then the resource flags tree depth used should be 3
# ---- resource type list [REGEX] ----
Scenario: Resource type list with regex filter
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
When I run resource flags type list with regex "git"
Then the resource flags output should contain "git-checkout"
And the resource flags output should not contain "fs-directory"
Scenario: Resource type list without regex shows all
Given a fresh resource flags test registry
And resource flags built-in types are bootstrapped
When I run resource flags type list without regex
Then the resource flags output should contain "git-checkout"
And the resource flags output should contain "fs-directory"
# ---- lsp list [REGEX] ----
Scenario: LSP list with regex filter
Given a clean LSP flags test environment
And a registered LSP flags server "local/pyright-test" for language "python"
And a registered LSP flags server "local/clangd-test" for language "cpp"
When I run lsp flags list with regex "pyright"
Then the lsp flags output should contain "local/pyright-test"
And the lsp flags output should not contain "local/clangd-test"
Scenario: LSP list without regex shows all
Given a clean LSP flags test environment
And a registered LSP flags server "local/pyright-test2" for language "python"
And a registered LSP flags server "local/clangd-test2" for language "cpp"
When I run lsp flags list without regex
Then the lsp flags output should contain "local/pyright-test2"
And the lsp flags output should contain "local/clangd-test2"
@@ -398,6 +398,8 @@ def step_run_resource_add_branch_only(
branch=branch,
description=None,
read_only=False,
clone_into=None,
update=False,
fmt="rich",
)
context.resource_cli_output = output
@@ -0,0 +1,550 @@
"""Step definitions for ``resource_cli_flags_904.feature`` (issue #904).
Tests the new CLI flags:
- resource add --update, --mount (container types), --clone-into
- resource list --all
- resource tree --depth default changed to 3
- resource type list [REGEX]
- lsp list [REGEX]
"""
from __future__ import annotations
from datetime import UTC, datetime
from io import StringIO
from typing import Any
from behave import given, then, when
from behave.runner import Context
from rich.console import Console
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.cli.commands.lsp import _reset_registry
from cleveragents.infrastructure.database.models import Base, ResourceModel
from cleveragents.lsp.models import LspCapability, LspServerConfig
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_flags_service(context: Context) -> ResourceRegistryService:
"""Create or return a cached ResourceRegistryService for flags tests."""
if not hasattr(context, "flags_cli_service"):
engine = create_engine("sqlite:///:memory:", echo=False)
@event.listens_for(engine, "connect")
def _fk(conn: Any, _rec: Any) -> None:
conn.cursor().execute("PRAGMA foreign_keys=ON")
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, expire_on_commit=False)
context.flags_cli_service = ResourceRegistryService(session_factory=factory)
return context.flags_cli_service
def _capture_flags_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
"""Run a CLI function capturing its console output and success status."""
import contextlib
buf = StringIO()
console = Console(
file=buf, width=200, no_color=True, highlight=False, force_terminal=False
)
import cleveragents.cli.commands.resource as resource_mod
orig_console = resource_mod.console
resource_mod.console = console
failed = False
try:
with contextlib.redirect_stdout(buf):
func(*args, **kwargs)
except SystemExit:
failed = True
except Exception:
failed = True
finally:
resource_mod.console = orig_console
return buf.getvalue(), failed
def _patch_flags_service(context: Context) -> Any:
"""Monkey-patch the container to return our in-memory service."""
import cleveragents.cli.commands.resource as resource_mod
orig_fn = resource_mod._get_registry_service
def _mock_get() -> ResourceRegistryService:
return _make_flags_service(context)
resource_mod._get_registry_service = _mock_get
return orig_fn
def _unpatch_flags_service(orig_fn: Any) -> None:
"""Restore original service getter."""
import cleveragents.cli.commands.resource as resource_mod
resource_mod._get_registry_service = orig_fn
# ---------------------------------------------------------------------------
# Background / Given
# ---------------------------------------------------------------------------
@given("a fresh resource flags test registry")
def step_fresh_flags_registry(context: Context) -> None:
engine = create_engine("sqlite:///:memory:", echo=False)
@event.listens_for(engine, "connect")
def _fk(conn: Any, _rec: Any) -> None:
conn.cursor().execute("PRAGMA foreign_keys=ON")
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, expire_on_commit=False)
context.flags_cli_service = ResourceRegistryService(session_factory=factory)
context.flags_cli_output = ""
context.flags_cli_failed = False
context.flags_tree_depth_used = None
@given("resource flags built-in types are bootstrapped")
def step_flags_bootstrap(context: Context) -> None:
service = _make_flags_service(context)
service.bootstrap_builtin_types()
@given(
'I add a resource flags resource "{type_name}" named "{name}" with path "{path}"'
)
def step_flags_add_resource_given(
context: Context, type_name: str, name: str, path: str
) -> None:
from cleveragents.cli.commands.resource import resource_add
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_add,
type_name=type_name,
name=name,
path=path,
branch=None,
description=None,
read_only=False,
clone_into=None,
update=False,
fmt="rich",
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
@given('a resource flags resource "{type_name}" named "{name}" with path "{path}"')
def step_flags_add_resource_given_alt(
context: Context, type_name: str, name: str, path: str
) -> None:
service = _make_flags_service(context)
service.register_resource(
type_name=type_name,
name=name,
location=path,
description=None,
read_only=False,
properties={"path": path},
)
@given(
'a user-added resource flags resource "{type_name}" named "{name}" with path "{path}"'
)
def step_flags_add_user_resource(
context: Context, type_name: str, name: str, path: str
) -> None:
service = _make_flags_service(context)
service.register_resource(
type_name=type_name,
name=name,
location=path,
description=None,
read_only=False,
properties={"path": path},
)
@given('an auto-discovered resource flags resource named "{name}"')
def step_flags_add_auto_discovered(context: Context, name: str) -> None:
"""Directly insert an auto-discovered resource into the DB."""
from ulid import ULID
service = _make_flags_service(context)
session = service._session()
try:
now_iso = datetime.now(tz=UTC).isoformat()
db_resource = ResourceModel(
resource_id=str(ULID()),
namespaced_name=name,
namespace=name.split("/", 1)[0] if "/" in name else None,
type_name="git-checkout",
resource_kind="physical",
location="/tmp/auto",
description="Auto-discovered",
read_only=False,
auto_discovered=True,
sandbox_strategy=None,
content_hash=None,
properties_json=None,
metadata_json=None,
created_at=now_iso,
updated_at=now_iso,
)
session.add(db_resource)
session.commit()
finally:
session.close()
# ---------------------------------------------------------------------------
# LSP Given
# ---------------------------------------------------------------------------
@given("a clean LSP flags test environment")
def step_clean_lsp_flags(context: Context) -> None:
_reset_registry()
context.lsp_flags_output = ""
context.lsp_flags_failed = False
@given('a registered LSP flags server "{name}" for language "{lang}"')
def step_register_lsp_flags_server(context: Context, name: str, lang: str) -> None:
from cleveragents.cli.commands.lsp import _get_registry
registry = _get_registry()
config = LspServerConfig(
name=name,
command="test-server",
args=["--stdio"],
languages=[lang],
capabilities=[LspCapability.DIAGNOSTICS],
)
registry.register(config)
# ---------------------------------------------------------------------------
# When steps — resource add
# ---------------------------------------------------------------------------
@when('I run resource flags add "{type_name}" "{name}" with path "{path}" and update')
def step_flags_add_update(
context: Context, type_name: str, name: str, path: str
) -> None:
from cleveragents.cli.commands.resource import resource_add
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_add,
type_name=type_name,
name=name,
path=path,
branch=None,
description=None,
read_only=False,
clone_into=None,
update=True,
fmt="rich",
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
@when(
'I run resource flags add "{type_name}" "{name}" with path "{path}" and mount "{mount_spec}"'
)
def step_flags_add_mount(
context: Context, type_name: str, name: str, path: str, mount_spec: str
) -> None:
from cleveragents.cli.commands.resource import resource_add
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_add,
type_name=type_name,
name=name,
path=path,
branch=None,
description=None,
read_only=False,
mount=[mount_spec],
clone_into=None,
update=False,
fmt="rich",
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
@when(
'I run resource flags add "{type_name}" "{name}" with path "{path}" and clone-into "{clone_spec}"'
)
def step_flags_add_clone_into(
context: Context, type_name: str, name: str, path: str, clone_spec: str
) -> None:
from cleveragents.cli.commands.resource import resource_add
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_add,
type_name=type_name,
name=name,
path=path,
branch=None,
description=None,
read_only=False,
clone_into=clone_spec,
update=False,
fmt="rich",
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
@when('I run resource flags add "{type_name}" "{name}" with clone-into "{clone_spec}"')
def step_flags_add_clone_into_no_path(
context: Context, type_name: str, name: str, clone_spec: str
) -> None:
from cleveragents.cli.commands.resource import resource_add
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_add,
type_name=type_name,
name=name,
path=None,
branch=None,
description=None,
read_only=False,
clone_into=clone_spec,
update=False,
fmt="rich",
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
# ---------------------------------------------------------------------------
# When steps — resource list
# ---------------------------------------------------------------------------
@when("I run resource flags list")
def step_flags_list(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_list
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_list, type_filter=None, show_all=False, fmt="rich"
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
@when("I run resource flags list with --all")
def step_flags_list_all(context: Context) -> None:
from cleveragents.cli.commands.resource import resource_list
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(
resource_list, type_filter=None, show_all=True, fmt="rich"
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
# ---------------------------------------------------------------------------
# When steps — resource tree
# ---------------------------------------------------------------------------
@when('I run resource flags tree for "{name}" with default depth')
def step_flags_tree_default(context: Context, name: str) -> None:
"""Run resource tree and capture the depth that was actually used."""
from cleveragents.cli.commands.resource import resource_tree
orig = _patch_flags_service(context)
try:
# We call resource_tree *without* passing depth so the default is used.
# We capture what depth the function uses by inspecting its default.
import inspect
sig = inspect.signature(resource_tree)
default_depth = sig.parameters["depth"].default
context.flags_tree_depth_used = default_depth
output, failed = _capture_flags_output(
resource_tree, resource=name, type_filter=None, fmt="rich"
)
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
# ---------------------------------------------------------------------------
# When steps — resource type list [REGEX]
# ---------------------------------------------------------------------------
@when('I run resource flags type list with regex "{regex}"')
def step_flags_type_list_regex(context: Context, regex: str) -> None:
from cleveragents.cli.commands.resource import type_list
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(type_list, regex=regex, fmt="rich")
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
@when("I run resource flags type list without regex")
def step_flags_type_list_no_regex(context: Context) -> None:
from cleveragents.cli.commands.resource import type_list
orig = _patch_flags_service(context)
try:
output, failed = _capture_flags_output(type_list, regex=None, fmt="rich")
context.flags_cli_output = output
context.flags_cli_failed = failed
finally:
_unpatch_flags_service(orig)
# ---------------------------------------------------------------------------
# When steps — lsp list [REGEX]
# ---------------------------------------------------------------------------
def _capture_lsp_output(func: Any, *args: Any, **kwargs: Any) -> tuple[str, bool]:
"""Run an LSP CLI function capturing its console output."""
import contextlib
buf = StringIO()
console = Console(
file=buf, width=200, no_color=True, highlight=False, force_terminal=False
)
import cleveragents.cli.commands.lsp as lsp_mod
orig_console = lsp_mod.console
lsp_mod.console = console
failed = False
try:
with contextlib.redirect_stdout(buf):
func(*args, **kwargs)
except SystemExit:
failed = True
except Exception:
failed = True
finally:
lsp_mod.console = orig_console
return buf.getvalue(), failed
@when('I run lsp flags list with regex "{regex}"')
def step_lsp_flags_list_regex(context: Context, regex: str) -> None:
from cleveragents.cli.commands.lsp import list_servers
output, failed = _capture_lsp_output(list_servers, regex=regex, fmt="rich")
context.lsp_flags_output = output
context.lsp_flags_failed = failed
@when("I run lsp flags list without regex")
def step_lsp_flags_list_no_regex(context: Context) -> None:
from cleveragents.cli.commands.lsp import list_servers
output, failed = _capture_lsp_output(list_servers, regex=None, fmt="rich")
context.lsp_flags_output = output
context.lsp_flags_failed = failed
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then('the resource flags output should contain "{text}"')
def step_flags_output_contains(context: Context, text: str) -> None:
output = context.flags_cli_output
assert text.lower() in output.lower(), (
f"Expected '{text}' in output, got:\n{output}"
)
@then('the resource flags output should not contain "{text}"')
def step_flags_output_not_contains(context: Context, text: str) -> None:
output = context.flags_cli_output
assert text.lower() not in output.lower(), (
f"Did not expect '{text}' in output, got:\n{output}"
)
@then("the resource flags command should fail")
def step_flags_command_failed(context: Context) -> None:
assert context.flags_cli_failed, "Expected command to fail but it succeeded"
@then("the resource flags tree depth used should be {expected_depth:d}")
def step_flags_tree_depth(context: Context, expected_depth: int) -> None:
assert context.flags_tree_depth_used == expected_depth, (
f"Expected default depth {expected_depth}, got {context.flags_tree_depth_used}"
)
@then('the lsp flags output should contain "{text}"')
def step_lsp_flags_output_contains(context: Context, text: str) -> None:
output = context.lsp_flags_output
assert text.lower() in output.lower(), (
f"Expected '{text}' in output, got:\n{output}"
)
@then('the lsp flags output should not contain "{text}"')
def step_lsp_flags_output_not_contains(context: Context, text: str) -> None:
output = context.lsp_flags_output
assert text.lower() not in output.lower(), (
f"Did not expect '{text}' in output, got:\n{output}"
)
+8
View File
@@ -268,6 +268,8 @@ def _do_resource_add(context: Context, type_name: str, name: str, path: str) ->
branch=None,
description=None,
read_only=False,
clone_into=None,
update=False,
fmt="rich",
)
context.resource_cli_output = output
@@ -519,6 +521,8 @@ def step_run_resource_add_fmt(
branch=None,
description=None,
read_only=False,
clone_into=None,
update=False,
fmt=fmt,
)
context.resource_cli_output = output
@@ -552,6 +556,8 @@ def step_run_resource_add_branch_desc(
branch=branch if branch else None,
description=desc,
read_only=False,
clone_into=None,
update=False,
fmt="rich",
)
context.resource_cli_output = output
@@ -593,6 +599,8 @@ def step_add_resource_long_desc(context: Context) -> None:
branch=None,
description=long_desc,
read_only=False,
clone_into=None,
update=False,
fmt="rich",
)
context.resource_cli_output = output
@@ -141,6 +141,8 @@ def step_run_resource_add_fs(
branch=None,
description=None,
image=None,
clone_into=None,
update=False,
read_only=False,
fmt="rich",
)
@@ -140,6 +140,8 @@ def step_run_resource_add_git(
branch=branch,
description=None,
image=None,
clone_into=None,
update=False,
read_only=False,
fmt="rich",
)
@@ -179,6 +179,7 @@ class ResourceInstanceMixin:
type_name: str | None = None,
*,
exact: bool = False,
include_auto_discovered: bool = True,
) -> list[Resource]:
"""List registered resources.
@@ -188,6 +189,8 @@ class ResourceInstanceMixin:
include resources whose type is a **subtype** of
*type_name* (ADR-042 polymorphic listing). When
``True``, match only the exact type name.
include_auto_discovered: When ``False``, exclude resources
with ``auto_discovered=True`` (default shows all).
Returns:
List of ``Resource`` domain objects.
@@ -204,6 +207,8 @@ class ResourceInstanceMixin:
subtypes = find_subtypes(type_name, registry)
all_types = [type_name, *subtypes]
query = query.filter(ResourceModel.type_name.in_(all_types))
if not include_auto_discovered:
query = query.filter(ResourceModel.auto_discovered == False) # noqa: E712
query = query.order_by(ResourceModel.created_at)
rows = query.all()
return [db_resource_to_domain(row) for row in rows]
+19
View File
@@ -48,6 +48,7 @@ from __future__ import annotations
import json
import logging
import os
import re as _re
from pathlib import Path
from typing import Annotated, Any
@@ -237,6 +238,12 @@ def remove(
@app.command("list")
def list_servers(
regex: Annotated[
str | None,
typer.Argument(
help="Optional regex to filter server names",
),
] = None,
namespace: Annotated[
str | None,
typer.Option("--namespace", "-n", help="Filter by namespace"),
@@ -252,8 +259,11 @@ def list_servers(
) -> None:
"""List registered LSP servers.
An optional positional REGEX argument filters server names.
Examples:
agents lsp list
agents lsp list "pyright.*"
agents lsp list --namespace local
agents lsp list --language python
agents lsp list --format json
@@ -261,6 +271,15 @@ def list_servers(
registry = _get_registry()
servers = registry.list_servers(namespace=namespace, language=language)
# Apply optional regex filter on server names.
if regex is not None:
try:
pattern = _re.compile(regex)
except _re.error as re_exc:
console.print(f"[red]Invalid regex:[/red] {re_exc}")
raise typer.Abort() from re_exc
servers = [s for s in servers if pattern.search(s.name)]
if not servers:
console.print("[yellow]No LSP servers found.[/yellow]")
console.print("Register one with 'agents lsp add --config <file>'")
+106 -12
View File
@@ -55,6 +55,7 @@ from __future__ import annotations
import json
import logging
import re
from pathlib import Path
from typing import Annotated, Any
@@ -100,6 +101,9 @@ console = _get_console()
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# Resource types that support --mount and --clone-into flags.
_CONTAINER_TYPES = frozenset({"container-instance", "devcontainer-instance"})
def _get_registry_service() -> ResourceRegistryService:
"""Get the ResourceRegistryService from the DI container."""
@@ -291,6 +295,12 @@ def type_remove(
@type_app.command("list")
def type_list(
regex: Annotated[
str | None,
typer.Argument(
help="Optional regex to filter type names",
),
] = None,
fmt: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
@@ -298,14 +308,26 @@ def type_list(
) -> None:
"""List all registered resource types (built-in and custom).
An optional positional REGEX argument filters type names.
Examples:
agents resource type list
agents resource type list "git.*"
agents resource type list --format json
"""
try:
service = _get_registry_service()
types = service.list_types()
# Apply optional regex filter on type names.
if regex is not None:
try:
pattern = re.compile(regex)
except re.error as re_exc:
console.print(f"[red]Invalid regex:[/red] {re_exc}")
raise typer.Abort() from re_exc
types = [t for t in types if pattern.search(t.name)]
if not types:
console.print("[yellow]No resource types registered.[/yellow]")
return
@@ -548,7 +570,7 @@ def resource_add(
typer.Option(
"--mount",
help=(
"Mount specification for container-instance resources. "
"Mount specification for container types. "
"Formats: <resource>:<path> (resource-ref, rw) or "
"<host-path>:<container-path>:<mode> (host-path, mode=rw|ro). "
"Can be specified multiple times."
@@ -560,12 +582,19 @@ def resource_add(
typer.Option(
"--clone-into",
help=(
"Clone a remote repository into the container at the specified "
"path. Format: REPO_URL:CONTAINER_PATH (e.g., "
"https://github.com/acme/api.git:/workspace)"
"Clone a repository into the container at a specified path. "
"Format: <REPO_URL>:<CONTAINER_PATH>. "
"Only valid for container types."
),
),
] = None,
update: Annotated[
bool,
typer.Option(
"--update",
help="Re-register an existing resource (remove and re-add).",
),
] = False,
read_only: Annotated[
bool,
typer.Option("--read-only", help="Mark resource as read-only"),
@@ -577,15 +606,19 @@ def resource_add(
) -> None:
"""Add a new resource instance of the specified type.
Use ``--update`` to re-register an existing resource by name.
Examples:
agents resource add git-checkout local/my-repo --path /home/user/repo
agents resource add fs-directory local/data --path /data --read-only
agents resource add devcontainer-instance local/dc --path /project
agents resource add container-instance local/ctr --image ubuntu:latest
agents resource add container-instance local/ctr --image node:18 \
--mount local/api-repo:/workspace \
agents resource add container-instance local/ctr --image node:18 \\
--mount local/api-repo:/workspace \\
--mount /var/shared/config:/config:ro
agents resource add container-instance cloud/ci --clone-into https://github.com/acme/api.git:/workspace
agents resource add container-instance local/ctr --image node:18 \\
--clone-into https://github.com/org/repo.git:/workspace/repo
agents resource add git-checkout local/my-repo --path /tmp --update
"""
try:
# Validate --clone-into is only used with container resource types.
@@ -612,6 +645,36 @@ def resource_add(
service = _get_registry_service()
# --update: remove existing resource before re-registering
if update:
try:
existing = service.show_resource(name)
# Remove the existing resource so we can re-register
session = service._session()
try:
from cleveragents.infrastructure.database.models import (
ResourceEdgeModel,
ResourceModel,
)
# Remove edges first
session.query(ResourceEdgeModel).filter(
(ResourceEdgeModel.parent_id == existing.resource_id)
| (ResourceEdgeModel.child_id == existing.resource_id)
).delete(synchronize_session="fetch")
session.query(ResourceModel).filter_by(
resource_id=existing.resource_id
).delete(synchronize_session="fetch")
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
except NotFoundError:
# Resource doesn't exist yet — proceed with normal add
pass
# Build properties from type-specific flags
properties: dict[str, Any] = {}
if path is not None:
@@ -621,15 +684,32 @@ def resource_add(
if image is not None:
properties["image"] = image
if mount:
if type_name != "container-instance":
if type_name not in _CONTAINER_TYPES:
console.print(
"[red]Error:[/red] --mount is only valid for "
"container-instance resources."
"container types "
f"({', '.join(sorted(_CONTAINER_TYPES))})."
)
raise typer.Abort()
properties["mounts"] = json.dumps([_parse_mount_spec(m) for m in mount])
if clone_into is not None:
properties["clone_into"] = clone_into
if type_name not in _CONTAINER_TYPES:
console.print(
"[red]Error:[/red] --clone-into is only valid for "
"container types "
f"({', '.join(sorted(_CONTAINER_TYPES))})."
)
raise typer.Abort()
parts = clone_into.rsplit(":", 1)
if len(parts) != 2 or not parts[0] or not parts[1]:
console.print(
"[red]Error:[/red] --clone-into must be in "
"<REPO_URL>:<CONTAINER_PATH> format."
)
raise typer.Abort()
properties["clone_into"] = json.dumps(
{"repo_url": parts[0], "container_path": parts[1]}
)
resource = service.register_resource(
type_name=type_name,
@@ -672,6 +752,13 @@ def resource_list(
str | None,
typer.Option("--type", "-t", help="Filter by resource type"),
] = None,
show_all: Annotated[
bool,
typer.Option(
"--all",
help="Include auto-discovered child resources in the listing.",
),
] = False,
fmt: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
@@ -679,14 +766,21 @@ def resource_list(
) -> None:
"""List registered resources.
By default only user-added (named) resources are shown. Use
``--all`` to include auto-discovered children.
Examples:
agents resource list
agents resource list --all
agents resource list --type git-checkout
agents resource list --format json
"""
try:
service = _get_registry_service()
resources = service.list_resources(type_name=type_filter)
resources = service.list_resources(
type_name=type_filter,
include_auto_discovered=show_all,
)
if not resources:
console.print("[yellow]No resources found.[/yellow]")
@@ -806,7 +900,7 @@ def resource_tree(
depth: Annotated[
int,
typer.Option("--depth", "-d", help="Maximum tree depth (-1 for unlimited)"),
] = -1,
] = 3,
type_filter: Annotated[
str | None,
typer.Option("--type", "-t", help="Filter children by resource type"),