robandClaude Opus 4.5
fe5b2047a9
- save_tool() now preserves registry_hash, registry_status, registry_owner, and registry_feedback from existing config when writing Tool.to_dict(), preventing edits from erasing registry metadata. - CLI publish (registry_commands.py) and collection publish (collections_commands.py) now save registry_owner from the publish response, matching the GUI publish dialog behavior. Co-Authored-By: Claude Opus 4.5 <[email protected]>
925 lines
33 KiB
Python
925 lines
33 KiB
Python
"""Collections commands for CmdForge CLI."""
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
|
|
from ..resolver import install_from_registry
|
|
|
|
|
|
def cmd_collections(args):
|
|
"""Handle collections subcommands."""
|
|
cmd = args.collections_cmd
|
|
|
|
# Existing commands (with --local flag support added)
|
|
if cmd == "list":
|
|
return _cmd_collections_list(args)
|
|
elif cmd == "info":
|
|
return _cmd_collections_info(args)
|
|
elif cmd == "install":
|
|
return _cmd_collections_install(args)
|
|
|
|
# Local management commands
|
|
elif cmd == "create":
|
|
return cmd_collections_create(args)
|
|
elif cmd == "show":
|
|
return cmd_collections_show(args)
|
|
elif cmd == "add":
|
|
return cmd_collections_add(args)
|
|
elif cmd == "remove":
|
|
return cmd_collections_remove(args)
|
|
elif cmd == "delete":
|
|
return cmd_collections_delete(args)
|
|
|
|
# Publishing commands
|
|
elif cmd == "publish":
|
|
return cmd_collections_publish(args)
|
|
elif cmd == "status":
|
|
return cmd_collections_status(args)
|
|
|
|
else:
|
|
# Default: show help
|
|
print("Collections commands:")
|
|
print(" list List available collections (registry)")
|
|
print(" info <name> Show registry collection details")
|
|
print(" install <name> Install all tools in a registry collection")
|
|
print()
|
|
print("Local collection management:")
|
|
print(" create <name> Create a new local collection")
|
|
print(" show <name> Show local collection details")
|
|
print(" add <coll> <tool> Add a tool to a collection")
|
|
print(" remove <coll> <tool> Remove a tool from a collection")
|
|
print(" delete <name> Delete a local collection")
|
|
print()
|
|
print("Publishing:")
|
|
print(" publish <name> Publish a collection to the registry")
|
|
print(" status <name> Check status of a pending collection")
|
|
return 0
|
|
|
|
|
|
def _cmd_collections_list(args):
|
|
"""List available collections (registry or local)."""
|
|
import json as json_mod
|
|
from ..collection import list_collections, get_collection
|
|
|
|
# Handle --local flag
|
|
if getattr(args, 'local', False):
|
|
local_colls = list_collections()
|
|
|
|
if getattr(args, 'json', False):
|
|
data = []
|
|
for name in local_colls:
|
|
coll = get_collection(name)
|
|
if coll:
|
|
data.append(coll.to_dict())
|
|
print(json_mod.dumps({"collections": data}, indent=2))
|
|
return 0
|
|
|
|
if not local_colls:
|
|
print("No local collections.")
|
|
print("Create one with: cmdforge collections create <name>")
|
|
return 0
|
|
|
|
print(f"Local collections ({len(local_colls)}):\n")
|
|
for name in local_colls:
|
|
coll = get_collection(name)
|
|
if coll:
|
|
status = "pending" if coll.pending_approval else ("published" if coll.published else "local")
|
|
print(f" {coll.display_name}")
|
|
print(f" {coll.name} ({len(coll.tools)} tools) [{status}]")
|
|
print()
|
|
return 0
|
|
|
|
# Registry listing
|
|
from ..registry_client import RegistryError, get_client
|
|
|
|
try:
|
|
client = get_client()
|
|
collections = client.get_collections()
|
|
|
|
# JSON output
|
|
if getattr(args, 'json', False):
|
|
print(json.dumps({"collections": collections}, indent=2))
|
|
return 0
|
|
|
|
if not collections:
|
|
print("No collections available.")
|
|
return 0
|
|
|
|
print(f"Available collections ({len(collections)}):\n")
|
|
for coll in collections:
|
|
name = coll.get("name", "")
|
|
display_name = coll.get("display_name", name)
|
|
description = coll.get("description", "")
|
|
# Use tool_count if available, fall back to counting tools list
|
|
tools = coll.get("tools", [])
|
|
tool_count = coll.get("tool_count", len(tools) if isinstance(tools, list) else 0)
|
|
|
|
print(f" {name}")
|
|
print(f" {display_name}")
|
|
if description:
|
|
desc_short = description[:60] + ('...' if len(description) > 60 else '')
|
|
print(f" {desc_short}")
|
|
print(f" Tools: {tool_count}")
|
|
print()
|
|
|
|
print("Install a collection with: cmdforge collections install <name>")
|
|
|
|
except RegistryError as e:
|
|
if e.code == "CONNECTION_ERROR":
|
|
print("Could not connect to the registry.", file=sys.stderr)
|
|
else:
|
|
print(f"Error: {e.message}", file=sys.stderr)
|
|
return 1
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
def _cmd_collections_info(args):
|
|
"""Show collection details."""
|
|
from ..registry_client import RegistryError, get_client
|
|
|
|
name = args.name
|
|
|
|
try:
|
|
client = get_client()
|
|
coll = client.get_collection(name)
|
|
|
|
# JSON output
|
|
if getattr(args, 'json', False):
|
|
print(json.dumps(coll, indent=2))
|
|
return 0
|
|
|
|
display_name = coll.get("display_name", name)
|
|
description = coll.get("description", "")
|
|
maintainer = coll.get("maintainer", "")
|
|
tools = coll.get("tools", [])
|
|
pinned = coll.get("pinned", {})
|
|
tags = coll.get("tags", [])
|
|
|
|
print(f"{display_name}")
|
|
print("=" * 50)
|
|
if description:
|
|
print(f"\n{description}\n")
|
|
if maintainer:
|
|
print(f"Maintainer: {maintainer}")
|
|
if tags:
|
|
print(f"Tags: {', '.join(tags)}")
|
|
|
|
print(f"\nTools ({len(tools)}):")
|
|
for tool in tools:
|
|
# Handle both object format (from API) and string format
|
|
if isinstance(tool, dict):
|
|
tool_ref = f"{tool['owner']}/{tool['name']}"
|
|
version_constraint = tool.get('pinned_version', '')
|
|
else:
|
|
tool_ref = tool
|
|
version_constraint = pinned.get(tool_ref, '')
|
|
|
|
if version_constraint:
|
|
print(f" - {tool_ref} @ {version_constraint}")
|
|
else:
|
|
print(f" - {tool_ref}")
|
|
|
|
print(f"\nInstall all: cmdforge collections install {name}")
|
|
|
|
except RegistryError as e:
|
|
if e.code == "COLLECTION_NOT_FOUND":
|
|
print(f"Collection '{name}' not found.", file=sys.stderr)
|
|
print("List available collections: cmdforge collections list", file=sys.stderr)
|
|
elif e.code == "CONNECTION_ERROR":
|
|
print("Could not connect to the registry.", file=sys.stderr)
|
|
else:
|
|
print(f"Error: {e.message}", file=sys.stderr)
|
|
return 1
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
def _cmd_collections_install(args):
|
|
"""Install all tools in a collection."""
|
|
from ..registry_client import RegistryError, get_client
|
|
|
|
name = args.name
|
|
use_pinned = getattr(args, 'pinned', False)
|
|
|
|
try:
|
|
client = get_client()
|
|
coll = client.get_collection(name)
|
|
|
|
tools = coll.get("tools", [])
|
|
pinned = coll.get("pinned", {}) if use_pinned else {}
|
|
display_name = coll.get("display_name", name)
|
|
|
|
if not tools:
|
|
print(f"Collection '{name}' has no tools.")
|
|
return 0
|
|
|
|
print(f"Installing collection: {display_name}")
|
|
print(f"Tools to install: {len(tools)}")
|
|
if use_pinned and pinned:
|
|
print(f"Using pinned versions: {len(pinned)} tools")
|
|
print()
|
|
|
|
installed = 0
|
|
failed = 0
|
|
|
|
for tool in tools:
|
|
# Handle both object format (from API) and string format
|
|
if isinstance(tool, dict):
|
|
tool_ref = f"{tool['owner']}/{tool['name']}"
|
|
version = tool.get('pinned_version') if use_pinned else None
|
|
else:
|
|
tool_ref = tool
|
|
version = pinned.get(tool_ref) if use_pinned else None
|
|
|
|
try:
|
|
print(f" Installing {tool_ref}...", end=" ", flush=True)
|
|
resolved = install_from_registry(tool_ref, version)
|
|
print(f"v{resolved.version}")
|
|
installed += 1
|
|
except RegistryError as e:
|
|
print(f"FAILED: {e.message}")
|
|
failed += 1
|
|
except Exception as e:
|
|
print(f"FAILED: {e}")
|
|
failed += 1
|
|
|
|
print()
|
|
print(f"Installed: {installed}/{len(tools)}")
|
|
if failed:
|
|
print(f"Failed: {failed}")
|
|
return 1
|
|
|
|
print(f"\nCollection '{name}' installed successfully!")
|
|
|
|
except RegistryError as e:
|
|
if e.code == "COLLECTION_NOT_FOUND":
|
|
print(f"Collection '{name}' not found.", file=sys.stderr)
|
|
print("List available collections: cmdforge collections list", file=sys.stderr)
|
|
elif e.code == "CONNECTION_ERROR":
|
|
print("Could not connect to the registry.", file=sys.stderr)
|
|
else:
|
|
print(f"Error: {e.message}", file=sys.stderr)
|
|
return 1
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Local Collection Management Commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_collections_create(args):
|
|
"""Create a new local collection."""
|
|
from ..collection import Collection, COLLECTIONS_DIR, list_collections
|
|
|
|
name = args.name.strip().lower()
|
|
|
|
# Validate name
|
|
if not name or not re.match(r'^[a-z0-9-]+$', name):
|
|
print("Error: Collection name must be lowercase alphanumeric with hyphens only")
|
|
return 1
|
|
|
|
# Check if exists
|
|
if name in list_collections():
|
|
print(f"Error: Collection '{name}' already exists")
|
|
print(f"Use: cmdforge collections show {name}")
|
|
print(f"Or delete it first: cmdforge collections delete {name}")
|
|
return 1
|
|
|
|
# Interactive prompts
|
|
display_name = input(f"Display name [{name.replace('-', ' ').title()}]: ").strip()
|
|
if not display_name:
|
|
display_name = name.replace("-", " ").title()
|
|
|
|
description = input("Description: ").strip()
|
|
tags_input = input("Tags (comma-separated): ").strip()
|
|
tags = [t.strip() for t in tags_input.split(",") if t.strip()]
|
|
|
|
collection = Collection(
|
|
name=name,
|
|
display_name=display_name,
|
|
description=description,
|
|
tags=tags,
|
|
)
|
|
collection.save()
|
|
|
|
print(f"\nCreated collection: {name}")
|
|
print(f" Location: {COLLECTIONS_DIR / f'{name}.yaml'}")
|
|
print(f"\nAdd tools with: cmdforge collections add {name} <tool>")
|
|
return 0
|
|
|
|
|
|
def cmd_collections_add(args):
|
|
"""Add a tool to a collection."""
|
|
from ..collection import get_collection
|
|
from ..tool import list_tools
|
|
|
|
collection = get_collection(args.collection)
|
|
if not collection:
|
|
print(f"Error: Collection '{args.collection}' not found")
|
|
return 1
|
|
|
|
tool_ref = args.tool
|
|
|
|
# Validate tool exists (local or suggest registry format)
|
|
if "/" not in tool_ref:
|
|
# Local tool reference
|
|
if tool_ref not in list_tools():
|
|
print(f"Warning: Local tool '{tool_ref}' not found")
|
|
confirm = input("Add anyway? [y/N]: ").strip().lower()
|
|
if confirm != "y":
|
|
return 1
|
|
|
|
if tool_ref in collection.tools:
|
|
print(f"Tool '{tool_ref}' already in collection")
|
|
return 0
|
|
|
|
collection.tools.append(tool_ref)
|
|
|
|
# Optional: set pinned version
|
|
if hasattr(args, 'version') and args.version:
|
|
collection.pinned[tool_ref] = args.version
|
|
|
|
collection.save()
|
|
print(f"Added '{tool_ref}' to collection '{collection.name}'")
|
|
return 0
|
|
|
|
|
|
def cmd_collections_remove(args):
|
|
"""Remove a tool from a collection."""
|
|
from ..collection import get_collection
|
|
|
|
collection = get_collection(args.collection)
|
|
if not collection:
|
|
print(f"Error: Collection '{args.collection}' not found")
|
|
return 1
|
|
|
|
tool_ref = args.tool
|
|
|
|
if tool_ref not in collection.tools:
|
|
print(f"Tool '{tool_ref}' not in collection")
|
|
return 1
|
|
|
|
collection.tools.remove(tool_ref)
|
|
collection.pinned.pop(tool_ref, None)
|
|
collection.save()
|
|
|
|
print(f"Removed '{tool_ref}' from collection '{collection.name}'")
|
|
return 0
|
|
|
|
|
|
def cmd_collections_show(args):
|
|
"""Show local collection details."""
|
|
from ..collection import get_collection, classify_tool_reference
|
|
from ..tool import list_tools
|
|
|
|
collection = get_collection(args.name)
|
|
if not collection:
|
|
print(f"Error: Collection '{args.name}' not found")
|
|
return 1
|
|
|
|
local_tools = set(list_tools())
|
|
|
|
print(f"\n{collection.display_name}")
|
|
print("=" * 50)
|
|
if collection.description:
|
|
print(f"\n{collection.description}\n")
|
|
|
|
if collection.tags:
|
|
print(f"Tags: {', '.join(collection.tags)}")
|
|
|
|
# Publishing status
|
|
if collection.pending_approval:
|
|
print(f"\nStatus: Pending (waiting for tool approvals)")
|
|
elif collection.published:
|
|
print(f"\nStatus: Published as '{collection.registry_name}'")
|
|
else:
|
|
print(f"\nStatus: Local only (not published)")
|
|
|
|
print(f"\nTools ({len(collection.tools)}):")
|
|
for tool_ref in collection.tools:
|
|
owner, name, is_local = classify_tool_reference(tool_ref)
|
|
pinned = collection.pinned.get(tool_ref, "")
|
|
version_str = f" @ {pinned}" if pinned else ""
|
|
|
|
if is_local:
|
|
exists = name in local_tools
|
|
status = "" if exists else " (not found)"
|
|
print(f" - {name}{version_str} [local]{status}")
|
|
else:
|
|
print(f" - {tool_ref}{version_str} [registry]")
|
|
|
|
if collection.pending_tools:
|
|
print(f"\nPending tool approvals:")
|
|
for tool in collection.pending_tools:
|
|
print(f" - {tool}")
|
|
|
|
return 0
|
|
|
|
|
|
def cmd_collections_delete(args):
|
|
"""Delete a local collection."""
|
|
from ..collection import get_collection
|
|
|
|
collection = get_collection(args.name)
|
|
if not collection:
|
|
print(f"Error: Collection '{args.name}' not found")
|
|
return 1
|
|
|
|
if not getattr(args, 'force', False):
|
|
confirm = input(f"Delete collection '{args.name}'? [y/N]: ").strip().lower()
|
|
if confirm != "y":
|
|
print("Cancelled")
|
|
return 0
|
|
|
|
collection.delete()
|
|
print(f"Deleted collection '{args.name}'")
|
|
return 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Publishing Commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_collections_publish(args):
|
|
"""Publish a collection to the registry."""
|
|
from ..collection import get_collection, resolve_tool_references
|
|
from ..registry_client import get_client, RegistryError
|
|
|
|
collection = get_collection(args.name)
|
|
if not collection:
|
|
print(f"Error: Collection '{args.name}' not found")
|
|
return 1
|
|
|
|
if not collection.tools:
|
|
print("Error: Collection has no tools")
|
|
return 1
|
|
|
|
# Get user info
|
|
client = get_client()
|
|
is_valid, error = client.validate_token()
|
|
if not is_valid:
|
|
print(f"Error: Not authenticated. Run: cmdforge config connect")
|
|
return 1
|
|
|
|
# Get user's registry slug
|
|
try:
|
|
user_info = client.get_me()
|
|
user_slug = user_info.get("slug", "")
|
|
except RegistryError as e:
|
|
print(f"Error getting user info: {e.message}")
|
|
return 1
|
|
|
|
if not user_slug:
|
|
print("Error: Could not determine your registry username")
|
|
return 1
|
|
|
|
# Check if resuming a pending publish
|
|
if getattr(args, 'resume', False) and collection.pending_approval:
|
|
return _resume_collection_publish(collection, client, user_slug)
|
|
|
|
# Resolve tool references (uses authenticated endpoint for user's own tools)
|
|
print(f"Analyzing collection '{collection.display_name}'...")
|
|
result = resolve_tool_references(
|
|
collection.tools, collection.pinned, user_slug, client
|
|
)
|
|
registry_refs = result.registry_refs
|
|
transformed_pinned = result.transformed_pinned
|
|
unpublished = result.local_unpublished
|
|
published = result.local_published # List of (name, status, has_approved) tuples
|
|
visibility_issues = result.visibility_issues
|
|
registry_tool_issues = result.registry_tool_issues
|
|
|
|
# Check for tools where latest version is pending
|
|
pending_latest = [(name, status) for name, status, has_approved in published if status != "approved"]
|
|
|
|
# Dry run mode
|
|
if getattr(args, 'dry_run', False):
|
|
print(f"\n[DRY RUN] Would publish collection: {collection.name}")
|
|
print(f" Display name: {collection.display_name}")
|
|
print(f" Tools: {len(collection.tools)}")
|
|
if visibility_issues:
|
|
print(f"\n BLOCKING: Non-public local tools ({len(visibility_issues)}):")
|
|
for name, visibility in visibility_issues:
|
|
print(f" - {name} (visibility: {visibility})")
|
|
if registry_tool_issues:
|
|
print(f"\n BLOCKING: Registry tool issues ({len(registry_tool_issues)}):")
|
|
for ref, reason in registry_tool_issues:
|
|
print(f" - {ref}: {reason}")
|
|
if unpublished:
|
|
print(f"\n Unpublished local tools ({len(unpublished)}):")
|
|
for tool in unpublished:
|
|
print(f" - {tool} -> {user_slug}/{tool}")
|
|
if pending_latest:
|
|
print(f"\n Pending approval ({len(pending_latest)}):")
|
|
for name, status in pending_latest:
|
|
print(f" - {user_slug}/{name} ({status})")
|
|
if transformed_pinned:
|
|
print(f"\n Version pins (transformed):")
|
|
for ref, version in transformed_pinned.items():
|
|
print(f" - {ref}: {version}")
|
|
print(f"\n Final tool references:")
|
|
for ref in registry_refs:
|
|
print(f" - {ref}")
|
|
return 0
|
|
|
|
# Handle visibility issues FIRST (local tools that aren't public)
|
|
if visibility_issues:
|
|
print(f"\n{len(visibility_issues)} local tool(s) are not public:")
|
|
for name, visibility in visibility_issues:
|
|
print(f" - {name} (visibility: {visibility})")
|
|
|
|
print(f"\nCollections can only include PUBLIC tools.")
|
|
print(f"Update these tools to be public before including in a collection:")
|
|
print(f" Edit the tool's config.yaml and set: visibility: public")
|
|
print(f" Then republish with: cmdforge registry publish <toolname>")
|
|
return 1
|
|
|
|
# Handle registry tool issues (external tools that aren't public/approved)
|
|
if registry_tool_issues:
|
|
print(f"\n{len(registry_tool_issues)} registry tool(s) cannot be included:")
|
|
for ref, reason in registry_tool_issues:
|
|
print(f" - {ref}: {reason}")
|
|
|
|
print(f"\nCollections can only include PUBLIC, APPROVED tools.")
|
|
print(f"Remove these tools from the collection or wait for them to be approved.")
|
|
return 1
|
|
|
|
# Handle tools where latest version is pending (warn but don't block)
|
|
if pending_latest:
|
|
print(f"\n{len(pending_latest)} tool(s) have pending latest versions:")
|
|
for name, status in pending_latest:
|
|
print(f" - {user_slug}/{name} ({status})")
|
|
print(f"\n (An older approved version may still be usable - will attempt publish)")
|
|
|
|
# Handle unpublished tools
|
|
if unpublished:
|
|
print(f"\nThis collection contains {len(unpublished)} unpublished local tool(s):")
|
|
for tool in unpublished:
|
|
print(f" - {tool}")
|
|
|
|
print(f"\nThese must be published as '{user_slug}/<name>' before the collection.")
|
|
print("\nNOTE: Only PUBLIC, APPROVED tools can be included in collections.")
|
|
print("\nOptions:")
|
|
print(" [1] Publish all tools now, then collection (recommended)")
|
|
print(" [2] Remove unpublished tools and publish remaining")
|
|
print(" [3] Cancel")
|
|
|
|
choice = input("\nChoice [1/2/3]: ").strip()
|
|
|
|
if choice == "1":
|
|
# Publish tools first
|
|
pending_tools = []
|
|
for tool_name in unpublished:
|
|
print(f"\nPublishing tool: {tool_name}...")
|
|
result = _publish_single_tool(tool_name, client)
|
|
if result.get("pending"):
|
|
pending_tools.append(tool_name)
|
|
print(f" -> Submitted for review (pending approval)")
|
|
elif result.get("success"):
|
|
print(f" -> Published as {user_slug}/{tool_name}")
|
|
else:
|
|
print(f" -> Failed: {result.get('error')}")
|
|
return 1
|
|
|
|
if pending_tools:
|
|
# Save pending state
|
|
collection.pending_approval = True
|
|
collection.pending_tools = pending_tools
|
|
collection.maintainer = user_slug
|
|
collection.save()
|
|
|
|
print(f"\nSome tools are pending approval:")
|
|
for tool in pending_tools:
|
|
print(f" - {user_slug}/{tool}")
|
|
print(f"\nCollection will be published once tools are approved.")
|
|
print(f"Check status with: cmdforge collections status {collection.name}")
|
|
print(f"Resume with: cmdforge collections publish {collection.name} --continue")
|
|
return 0
|
|
|
|
# All tools approved immediately (or auto-approved)
|
|
# Fall through to publish collection
|
|
|
|
elif choice == "2":
|
|
# Remove unpublished tools from refs and pinned
|
|
unpublished_full_refs = {f"{user_slug}/{u}" for u in unpublished}
|
|
registry_refs = [ref for ref in registry_refs if ref not in unpublished_full_refs]
|
|
# Also remove from transformed_pinned to avoid orphaned keys
|
|
for u in unpublished:
|
|
full_ref = f"{user_slug}/{u}"
|
|
transformed_pinned.pop(full_ref, None)
|
|
|
|
if not registry_refs:
|
|
print("Error: No tools remaining after removal")
|
|
return 1
|
|
|
|
# Update local collection to match what we're publishing
|
|
collection.tools = [t for t in collection.tools if t not in unpublished]
|
|
for u in unpublished:
|
|
collection.pinned.pop(u, None)
|
|
collection.save()
|
|
|
|
print(f"\nProceeding with {len(registry_refs)} tool(s)")
|
|
|
|
else:
|
|
print("Cancelled")
|
|
return 0
|
|
|
|
# Publish collection (using transformed pinned keys)
|
|
return _publish_collection_to_registry(collection, registry_refs, transformed_pinned, client, user_slug)
|
|
|
|
|
|
def _bump_patch_version(version: str) -> str:
|
|
"""Bump the patch version of a semver string."""
|
|
import re
|
|
match = re.match(r'^(\d+)\.(\d+)\.(\d+)(.*)?$', version)
|
|
if match:
|
|
major, minor, patch = int(match.group(1)), int(match.group(2)), int(match.group(3))
|
|
suffix = match.group(4) or ""
|
|
return f"{major}.{minor}.{patch + 1}{suffix}"
|
|
# Fallback: append .1 if not valid semver
|
|
return f"{version}.1"
|
|
|
|
|
|
def _publish_single_tool(tool_name: str, client) -> dict:
|
|
"""
|
|
Publish a single tool. Returns status dict.
|
|
|
|
Updates local tool config with registry_hash and registry_status
|
|
to maintain consistency with direct `cmdforge registry publish` flow.
|
|
|
|
If the tool was previously rejected, automatically bumps the version
|
|
to allow resubmission.
|
|
"""
|
|
from ..tool import load_tool, get_tools_dir
|
|
from ..registry_client import RegistryError
|
|
import yaml
|
|
|
|
# For publishing, prefer qualified (owned) path first, then fall back to unqualified.
|
|
# This ensures we publish the explicitly-owned version when both exist.
|
|
tool = None
|
|
try:
|
|
me = client.get_me()
|
|
owner = me.get("slug", "")
|
|
except Exception:
|
|
owner = ""
|
|
|
|
if owner:
|
|
tool = load_tool(f"{owner}/{tool_name}")
|
|
if not tool:
|
|
tool = load_tool(tool_name)
|
|
if not tool:
|
|
return {"success": False, "error": f"Tool '{tool_name}' not found"}
|
|
|
|
tool_dir = tool.path.parent if tool.path else (get_tools_dir() / tool_name)
|
|
config_path = tool_dir / "config.yaml"
|
|
|
|
# Check if tool was previously rejected - if so, bump version
|
|
version_bumped = False
|
|
try:
|
|
status_info = client.get_my_tool_status(tool_name)
|
|
current_status = status_info.get("status", "")
|
|
if current_status == "rejected":
|
|
# Bump version to allow resubmission
|
|
old_version = tool.version
|
|
new_version = _bump_patch_version(old_version)
|
|
|
|
# Update local config
|
|
if config_path.exists():
|
|
config_data = yaml.safe_load(config_path.read_text()) or {}
|
|
config_data["version"] = new_version
|
|
config_path.write_text(yaml.dump(config_data, default_flow_style=False, sort_keys=False))
|
|
|
|
# Reload tool from same file (not by name, to preserve path)
|
|
from ..tool import Tool
|
|
config_data = yaml.safe_load(config_path.read_text()) or {}
|
|
tool = Tool.from_dict(config_data)
|
|
tool.path = config_path
|
|
version_bumped = True
|
|
print(f" (Version bumped {old_version} -> {new_version} due to previous rejection)")
|
|
except RegistryError as e:
|
|
if e.code != "TOOL_NOT_FOUND":
|
|
pass # Other errors - continue with publish attempt
|
|
|
|
# Load README/defaults if exists
|
|
readme_path = tool_dir / "README.md"
|
|
readme = readme_path.read_text() if readme_path.exists() else ""
|
|
defaults_path = tool_dir / "defaults.yaml"
|
|
defaults = defaults_path.read_text() if defaults_path.exists() else ""
|
|
|
|
try:
|
|
config_yaml = yaml.safe_dump(tool.to_dict(), sort_keys=False)
|
|
result = client.publish_tool(config_yaml, readme=readme, defaults=defaults, dry_run=False)
|
|
|
|
# Update local tool config with registry metadata (use same config_path from above)
|
|
config_hash = result.get("config_hash")
|
|
moderation_status = result.get("status", "pending")
|
|
|
|
if config_hash and config_path.exists():
|
|
try:
|
|
config_data = yaml.safe_load(config_path.read_text()) or {}
|
|
config_data["registry_hash"] = config_hash
|
|
config_data["registry_status"] = moderation_status
|
|
registry_owner = result.get("owner")
|
|
if registry_owner:
|
|
config_data["registry_owner"] = registry_owner
|
|
# Clear any old feedback when republishing
|
|
if "registry_feedback" in config_data:
|
|
del config_data["registry_feedback"]
|
|
config_path.write_text(yaml.dump(config_data, default_flow_style=False, sort_keys=False))
|
|
except Exception:
|
|
pass # Non-critical
|
|
|
|
# Check if pending moderation
|
|
is_pending = moderation_status == "pending" or result.get("pr_url")
|
|
return {
|
|
"success": True,
|
|
"pending": is_pending,
|
|
"status": moderation_status,
|
|
"config_hash": config_hash,
|
|
"pr_url": result.get("pr_url"),
|
|
"version_bumped": version_bumped,
|
|
"version": tool.version if tool else None
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
def _resume_collection_publish(collection, client, user_slug) -> int:
|
|
"""Resume publishing a collection after tools are approved."""
|
|
from ..registry_client import RegistryError
|
|
|
|
print(f"Checking status of pending tools...")
|
|
|
|
still_pending = []
|
|
for tool_name in collection.pending_tools:
|
|
try:
|
|
# Use authenticated endpoint to check user's own tools
|
|
status_info = client.get_my_tool_status(tool_name)
|
|
status = status_info.get("status", "pending")
|
|
if status != "approved":
|
|
still_pending.append(tool_name)
|
|
print(f" {user_slug}/{tool_name}: {status}")
|
|
else:
|
|
print(f" {user_slug}/{tool_name}: approved")
|
|
except RegistryError:
|
|
still_pending.append(tool_name)
|
|
print(f" {user_slug}/{tool_name}: not found")
|
|
|
|
if still_pending:
|
|
collection.pending_tools = still_pending
|
|
collection.save()
|
|
print(f"\nStill waiting for {len(still_pending)} tool(s) to be approved")
|
|
return 0
|
|
|
|
# All approved - publish collection
|
|
print(f"\nAll tools approved! Publishing collection...")
|
|
|
|
# Rebuild registry refs AND transform pinned keys
|
|
registry_refs = []
|
|
transformed_pinned = {}
|
|
for tool_ref in collection.tools:
|
|
if "/" in tool_ref:
|
|
registry_refs.append(tool_ref)
|
|
# Copy pinned if exists
|
|
if tool_ref in collection.pinned:
|
|
transformed_pinned[tool_ref] = collection.pinned[tool_ref]
|
|
else:
|
|
full_ref = f"{user_slug}/{tool_ref}"
|
|
registry_refs.append(full_ref)
|
|
# Transform pinned key from local name to owner/name
|
|
if tool_ref in collection.pinned:
|
|
transformed_pinned[full_ref] = collection.pinned[tool_ref]
|
|
|
|
return _publish_collection_to_registry(collection, registry_refs, transformed_pinned, client, user_slug)
|
|
|
|
|
|
def _publish_collection_to_registry(collection, registry_refs, transformed_pinned, client, user_slug) -> int:
|
|
"""Actually publish the collection to registry."""
|
|
print(f"\nPublishing collection '{collection.display_name}'...")
|
|
|
|
payload = {
|
|
"name": collection.name,
|
|
"display_name": collection.display_name,
|
|
"description": collection.description,
|
|
"maintainer": user_slug,
|
|
"tools": registry_refs,
|
|
"pinned": transformed_pinned,
|
|
"tags": collection.tags,
|
|
}
|
|
|
|
try:
|
|
result = client.publish_collection(payload)
|
|
|
|
# Update local state
|
|
collection.published = True
|
|
collection.registry_name = collection.name
|
|
collection.pending_approval = False
|
|
collection.pending_tools = []
|
|
collection.maintainer = user_slug
|
|
collection.save()
|
|
|
|
print(f"\nCollection published: {collection.name}")
|
|
print(f" View at: {collection.get_registry_url()}")
|
|
return 0
|
|
|
|
except Exception as e:
|
|
print(f"\nError publishing collection: {e}")
|
|
return 1
|
|
|
|
|
|
def cmd_collections_status(args):
|
|
"""Check status of a pending collection publish."""
|
|
from ..collection import get_collection
|
|
from ..registry_client import get_client, RegistryError
|
|
|
|
collection = get_collection(args.name)
|
|
if not collection:
|
|
print(f"Error: Collection '{args.name}' not found")
|
|
return 1
|
|
|
|
if not collection.pending_approval:
|
|
if collection.published:
|
|
print(f"Collection '{collection.name}' is published")
|
|
else:
|
|
print(f"Collection '{collection.name}' is local (not published)")
|
|
return 0
|
|
|
|
# Check pending tools using authenticated endpoint
|
|
client = get_client()
|
|
|
|
# Validate auth first
|
|
is_valid, error = client.validate_token()
|
|
if not is_valid:
|
|
print(f"Error: Not authenticated. Run: cmdforge config connect")
|
|
return 1
|
|
|
|
user_slug = collection.maintainer
|
|
|
|
print(f"\nCollection: {collection.display_name}")
|
|
print(f"Status: Pending tool approvals\n")
|
|
|
|
all_approved = True
|
|
has_rejected = False
|
|
has_changes_requested = False
|
|
rejected_tools = []
|
|
changes_tools = []
|
|
|
|
for tool_name in collection.pending_tools:
|
|
try:
|
|
# Use authenticated endpoint to see pending/private tools
|
|
status_info = client.get_my_tool_status(tool_name)
|
|
status = status_info.get("status", "pending")
|
|
if status == "approved":
|
|
print(f" {user_slug}/{tool_name}: approved")
|
|
else:
|
|
print(f" {user_slug}/{tool_name}: {status}")
|
|
all_approved = False
|
|
if status == "rejected":
|
|
has_rejected = True
|
|
rejected_tools.append(tool_name)
|
|
elif status == "changes_requested":
|
|
has_changes_requested = True
|
|
changes_tools.append(tool_name)
|
|
except RegistryError as e:
|
|
print(f" {user_slug}/{tool_name}: {e.message}")
|
|
all_approved = False
|
|
|
|
if all_approved:
|
|
print(f"\nAll tools approved!")
|
|
print(f"Run: cmdforge collections publish {collection.name} --continue")
|
|
elif has_rejected:
|
|
print(f"\nSome tools were rejected:")
|
|
for t in rejected_tools:
|
|
print(f" - {t}")
|
|
print(f"\nTo proceed, you need to:")
|
|
print(f" 1. Fix the issues with rejected tools")
|
|
print(f" 2. Republish them: cmdforge registry publish <toolname>")
|
|
print(f" 3. Then republish collection: cmdforge collections publish {collection.name}")
|
|
|
|
# Clear pending state since tools were rejected
|
|
collection.pending_approval = False
|
|
collection.pending_tools = []
|
|
collection.save()
|
|
print(f"\n(Collection pending state cleared)")
|
|
elif has_changes_requested:
|
|
print(f"\nSome tools need changes:")
|
|
for t in changes_tools:
|
|
print(f" - {t}")
|
|
print(f"\nAddress the feedback and republish the tools.")
|
|
else:
|
|
print(f"\nWaiting for tool approvals...")
|
|
|
|
return 0
|