rob
/
CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add admin publish-as-official, hash-based status sync, and GUI fixes 

Registry:
- Add admin owner override for publishing tools as "official"
- Add POST /api/v1/tools/status-by-hash batch endpoint for status lookup
  scoped to publisher_id (works for tools published under any owner)

GUI:
- Add "Publish as" dropdown in publish dialog for admin users
- Add "installed" tool state (teal with arrow indicator) for registry-installed tools
- Fix tool editing for official/* qualified tool names (_get_qualified_name helper)
- Fix cancel navigation returning to wrong page (Welcome instead of Tools)
- Fix collections tab not refreshing after publish
- Refactor StatusSyncWorker to use batch hash lookup (1 request instead of N)
  with chunking (100 max) and hash collision handling

CLI:
- Switch registry status sync to hash-based lookup
- Add collection dependency checking and unpublished dep detection

Publish dialog cleanup:
- Move yaml import to module level, remove duplicate _bump_patch_version
- Fix owner combo using currentText() for reliable selection

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
rob 2026-01-29 17:18:37 -04:00
parent 071ade0ffb
commit 0ee21f27f7
13 changed files with 1459 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
src/cmdforge/cli/collections_commands.py
View File

@ -640,22 +640,79 @@ def cmd_collections_publish(args):
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
tool = load_tool(tool_name)
# 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"}
# Load README/defaults if exists
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"
@ -665,8 +722,7 @@ def _publish_single_tool(tool_name: str, client) -> dict:
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
config_path = get_tools_dir() / tool_name / "config.yaml"
# 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")
@ -689,7 +745,9 @@ def _publish_single_tool(tool_name: str, client) -> dict:
"pending": is_pending,
"status": moderation_status,
"config_hash": config_hash,
"pr_url": result.get("pr_url")
"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)}
@ -810,6 +868,11 @@ def cmd_collections_status(args):
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
@ -820,6 +883,12 @@ def cmd_collections_status(args):
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
@ -827,6 +896,25 @@ def cmd_collections_status(args):
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...")

162
src/cmdforge/cli/registry_commands.py
View File

@ -440,6 +440,122 @@ def _cmd_registry_publish(args):
print("\nCancelled.")
return 1
# Check for unpublished dependencies
from ..collection import gather_local_unpublished_deps
from ..tool import load_tool, Tool, ToolStep
dep_result = None
try:
client = get_client()
# Get user slug for proper path resolution
try:
me = client.get_me()
my_owner = me.get("slug", "")
except Exception:
my_owner = None
tool = load_tool(name)
if tool:
dep_result = gather_local_unpublished_deps([name], client, my_owner)
else:
# Tool isn't installed locally; derive deps from config.yaml
temp_tool = Tool.from_dict(data)
dep_names = []
for dep in temp_tool.dependencies:
if '/' not in dep:
dep_names.append(dep)
for step in temp_tool.steps:
if isinstance(step, ToolStep) and '/' not in step.tool:
dep_names.append(step.tool)
dep_names = sorted(set(dep_names))
if dep_names:
dep_result = gather_local_unpublished_deps(dep_names, client, my_owner)
except Exception as e:
print(f"Warning: Could not check dependencies: {e}", file=sys.stderr)
dep_result = None
if dep_result and dep_result.unpublished:
print(f"Warning: This tool has unpublished dependencies:")
for dep in dep_result.unpublished:
print(f" - {dep}")
if dep_result.cycles:
print(f"\nWarning: Circular dependencies detected:")
for cycle in dep_result.cycles:
print(f" {' -> '.join(cycle)}")
if dep_result.skipped:
print(f"\nNote: Could not check {len(dep_result.skipped)} dep(s) due to errors")
print()
# Non-interactive mode: warn but continue
if not sys.stdin.isatty():
print("Non-interactive mode: proceeding without publishing dependencies.")
else:
try:
choice = input("Publish dependencies first? [Y/n/skip] ").lower().strip()
except (EOFError, KeyboardInterrupt):
print("\nCancelled.")
return 1
if choice == 'skip':
pass # Continue without deps
elif choice in ('n', 'no'):
print("Cancelled.")
return 1
elif choice in ('', 'y', 'yes'):
# Publish deps in topological order (only unpublished ones)
from .collections_commands import _publish_single_tool
for dep in dep_result.publish_order:
if dep in dep_result.unpublished and dep != name:
print(f"Publishing {dep}...")
result = _publish_single_tool(dep, client)
if not result.get("success"):
print(f" Failed: {result.get('error')}", file=sys.stderr)
return 1
elif result.get("pending"):
print(f" Submitted for review (pending approval)")
else:
print(f" Published successfully")
print()
else:
print("Cancelled.")
return 1
# Check if tool was previously rejected - if so, bump version
try:
client = get_client()
status_info = client.get_my_tool_status(name)
current_status = status_info.get("status", "")
if current_status == "rejected":
# Bump version to allow resubmission
def bump_patch(v):
import re as re_mod
match = re_mod.match(r'^(\d+)\.(\d+)\.(\d+)(.*)?$', v)
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}"
return f"{v}.1"
new_version = bump_patch(version)
print(f"Previous version was rejected. Bumping version {version} -> {new_version}")
# Update config_yaml with new version
data["version"] = new_version
config_yaml = yaml.dump(data, default_flow_style=False, sort_keys=False)
version = new_version
# Also update local file
config_path.write_text(config_yaml)
except RegistryError as e:
if e.code != "TOOL_NOT_FOUND":
pass # Other errors - continue with publish
except Exception:
pass # Continue with publish
print(f"Publishing {name}@{version}...")
try:
@ -560,14 +676,34 @@ def _cmd_registry_my_tools(args):
def _cmd_registry_status(args):
"""Check moderation status of a tool."""
from ..registry_client import RegistryError, get_client
from ..tool import get_tools_dir
from ..tool import get_tools_dir, load_tool
tool_name = args.tool
do_sync = getattr(args, 'sync', False)
# Check if tool exists locally
config_path = get_tools_dir() / tool_name / "config.yaml"
if not config_path.exists():
# Try to find the tool - prefer qualified (owned) path, then unqualified
config_path = None
tools_dir = get_tools_dir()
# Try to get user slug for owned path resolution
try:
client = get_client()
me = client.get_me()
my_owner = me.get("slug", "")
if my_owner:
owned_path = tools_dir / my_owner / tool_name / "config.yaml"
if owned_path.exists():
config_path = owned_path
except Exception:
pass # No auth or network error, try unqualified
# Fall back to unqualified path
if not config_path:
unqualified_path = tools_dir / tool_name / "config.yaml"
if unqualified_path.exists():
config_path = unqualified_path
if not config_path:
print(f"Error: Tool '{tool_name}' not found locally", file=sys.stderr)
return 1
@ -587,14 +723,20 @@ def _cmd_registry_status(args):
print("Publish with: cmdforge registry publish")
return 0
# If syncing, fetch from server
# If syncing, fetch from server using hash-based lookup
if do_sync:
try:
client = get_client()
status_data = client.get_my_tool_status(tool_name)
# Use hash-based batch lookup (works for tools published under any owner)
results = client.get_tool_status_by_hashes([local_hash])
status_data = results.get(local_hash)
if not status_data:
print(f"Tool '{tool_name}' not found in registry.", file=sys.stderr)
return 1
new_status = status_data.get("status", "pending")
new_hash = status_data.get("config_hash")
new_feedback = status_data.get("feedback")
changed = False
@ -602,10 +744,6 @@ def _cmd_registry_status(args):
config_data["registry_status"] = new_status
local_status = new_status
changed = True
if new_hash and local_hash != new_hash:
config_data["registry_hash"] = new_hash
local_hash = new_hash
changed = True
if new_feedback != local_feedback:
if new_feedback:
config_data["registry_feedback"] = new_feedback
@ -622,8 +760,6 @@ def _cmd_registry_status(args):
except RegistryError as e:
if e.code == "UNAUTHORIZED":
print("Not logged in. Set your registry token to sync.", file=sys.stderr)
elif e.code == "TOOL_NOT_FOUND":
print(f"Tool '{tool_name}' not found in registry.", file=sys.stderr)
else:
print(f"Error syncing: {e.message}", file=sys.stderr)
return 1

207
src/cmdforge/collection.py
View File

@ -126,17 +126,35 @@ class ToolResolutionResult:
registry_tool_issues: List[tuple] # (ref, reason) - registry tools that aren't public/approved
def _get_tool_visibility_from_yaml(tool_name: str) -> str:
def _get_tool_visibility_from_yaml(tool_name: str, my_owner: str = None) -> str:
"""
Read visibility directly from tool's config.yaml file.
NOTE: The Tool dataclass doesn't have a visibility field, so we read
the raw YAML to check this. This is a workaround until Tool model is updated.
Args:
tool_name: The tool name (unqualified)
my_owner: Current user's slug for resolving owned tool paths
"""
from .tool import get_tools_dir
config_path = get_tools_dir() / tool_name / "config.yaml"
if not config_path.exists():
tools_dir = get_tools_dir()
# Try owned path first if my_owner is set
config_path = None
if my_owner:
owned_path = tools_dir / my_owner / tool_name / "config.yaml"
if owned_path.exists():
config_path = owned_path
# Fall back to unqualified path
if not config_path:
unqualified_path = tools_dir / tool_name / "config.yaml"
if unqualified_path.exists():
config_path = unqualified_path
if not config_path:
return "public" # Default if not found
try:
@ -202,7 +220,7 @@ def resolve_tool_references(
registry_tool_issues.append((ref, f"not accessible ({e.message})"))
else:
# Local tool - check visibility from raw YAML config
local_visibility = _get_tool_visibility_from_yaml(name)
local_visibility = _get_tool_visibility_from_yaml(name, user_slug)
if local_visibility != 'public':
visibility_issues.append((name, local_visibility))
@ -235,3 +253,184 @@ def resolve_tool_references(
visibility_issues=visibility_issues,
registry_tool_issues=registry_tool_issues
)
@dataclass
class DepCheckResult:
"""Result of dependency checking."""
unpublished: List[str] # Local deps not yet in registry
publish_order: List[str] # Topological order for publishing
cycles: List[List[str]] # Detected cycles (for warning)
skipped: List[str] # Deps we couldn't check (network errors)
def gather_local_unpublished_deps(
tool_names: List[str],
client,
my_owner: str = None
) -> DepCheckResult:
"""
Gather all local dependencies and check which are unpublished.
Args:
tool_names: List of local tool names to check
client: Authenticated registry client
my_owner: Current user's slug (from client.get_me())
Returns:
DepCheckResult with unpublished deps, publish order, cycles, and skipped
"""
from .tool import ToolStep, TOOLS_DIR
from .registry_client import RegistryError
# Get my_owner if not provided
if my_owner is None:
try:
me = client.get_me()
my_owner = me.get("slug", "")
except Exception:
my_owner = ""
all_deps = set()
visited = set()
dep_graph = {}
cycles = []
def get_my_local_tool_path(name: str):
"""Get path to a local tool owned by me, or None if not found.
Resolution order:
1. ~/.cmdforge/<my_owner>/<name>/ (preferred if my_owner is set)
2. ~/.cmdforge/<name>/ (fallback for unqualified local)
"""
# Prefer owned local if my_owner is set
if my_owner:
owned_path = TOOLS_DIR / my_owner / name / "config.yaml"
if owned_path.exists():
return owned_path.parent
# Fallback to unqualified local
unqualified_path = TOOLS_DIR / name / "config.yaml"
if unqualified_path.exists():
return unqualified_path.parent
return None
def is_my_local_tool(name: str) -> bool:
"""Check if a tool is a local tool owned by me."""
return get_my_local_tool_path(name) is not None
cyclic_nodes = set() # Track nodes involved in cycles
def traverse(name: str, path: List[str] = None):
path = path or []
# Cycle detection
if name in path:
cycle_start = path.index(name)
cycle = path[cycle_start:] + [name]
cycles.append(cycle)
cyclic_nodes.update(cycle)
return
if name in visited:
return
visited.add(name)
# Use explicit path resolution and load by path
tool_path = get_my_local_tool_path(name)
if not tool_path:
return
# Load tool directly from resolved path (not by name)
from .tool import Tool
config_file = tool_path / "config.yaml"
try:
with open(config_file) as f:
data = yaml.safe_load(f)
tool = Tool.from_dict(data)
tool.path = config_file # Set path for consistency with load_tool()
except Exception:
return
dep_graph[name] = []
new_path = path + [name]
# Get explicit dependencies
for dep in tool.dependencies:
if '/' in dep:
continue # Skip registry refs
if not is_my_local_tool(dep):
continue # Skip tools I don't own
all_deps.add(dep)
dep_graph[name].append(dep)
traverse(dep, new_path)
# Get implicit dependencies from ToolStep
for step in tool.steps:
if isinstance(step, ToolStep):
if '/' in step.tool:
continue
if not is_my_local_tool(step.tool):
continue
all_deps.add(step.tool)
dep_graph[name].append(step.tool)
traverse(step.tool, new_path)
for name in tool_names:
traverse(name)
# Check which are unpublished or not approved (distinguish 404 from other errors)
# Check BOTH dependencies AND the original tools (they might be rejected too)
unpublished = []
skipped = []
tools_to_check = set(all_deps) | set(tool_names)
tools_to_check = {t for t in tools_to_check if is_my_local_tool(t)}
for dep in tools_to_check:
try:
status_info = client.get_my_tool_status(dep)
status = status_info.get("status", "")
# Only "approved" tools are actually usable
# "pending", "rejected", "changes_requested" all need (re)publishing
if status != "approved":
unpublished.append(dep)
except RegistryError as e:
if e.code == "TOOL_NOT_FOUND":
unpublished.append(dep)
else:
# Network/auth error - skip this check, don't assume unpublished
skipped.append(dep)
except Exception:
skipped.append(dep)
# Topological sort (excluding cyclic nodes)
def topo_sort(names):
result = []
perm_visited = set()
def visit(n):
if n in perm_visited or n in cyclic_nodes:
return # Skip cyclic nodes
perm_visited.add(n)
for child in dep_graph.get(n, []):
if child not in perm_visited:
visit(child)
result.append(n)
for n in names:
if n not in perm_visited:
visit(n)
return result
# Filter out cyclic nodes from unpublished
unpublished_safe = [u for u in unpublished if u not in cyclic_nodes]
# Order: unpublished deps first, then original tools
all_to_publish = unpublished_safe + list(tool_names)
ordered = topo_sort(all_to_publish)
return DepCheckResult(
unpublished=unpublished_safe, # Excludes cyclic
publish_order=ordered,
cycles=cycles,
skipped=skipped
)

271
src/cmdforge/gui/dialogs/publish_dialog.py
View File

@ -8,7 +8,9 @@ from PySide6.QtWidgets import (
)
from PySide6.QtCore import QThread, Signal
from ...tool import Tool
import yaml
from ...tool import Tool, get_all_categories
from ...registry_client import RegistryClient, RegistryError
from ...config import load_config
@ -35,7 +37,7 @@ def bump_version(version: str, bump_type: str) -> str:
class VersionFetchWorker(QThread):
"""Background worker to fetch registry version info."""
finished = Signal(dict) # Dict with 'my_version' and 'original_versions'
finished = Signal(dict) # Dict with 'my_version', 'original_versions', and 'user_role'
error = Signal(str)
def __init__(self, name: str, original_owner: Optional[str] = None):
@ -54,10 +56,20 @@ class VersionFetchWorker(QThread):
"my_status": None,
"original_versions": [],
"original_owner": self.original_owner,
"user_role": "user",
"user_slug": "",
}
# Get my published version of this tool
# Get my published version of this tool and user info
if config.registry.token:
try:
# Get user info including role
me = client.get_me()
result["user_role"] = me.get("role", "user")
result["user_slug"] = me.get("slug", "")
except RegistryError:
pass
try:
status = client.get_my_tool_status(self.name)
result["my_version"] = status.get("version")
@ -84,11 +96,12 @@ class PublishWorker(QThread):
success = Signal(dict)
error = Signal(str)
def __init__(self, config_yaml: str, readme: str = "", defaults: str = ""):
def __init__(self, config_yaml: str, readme: str = "", defaults: str = "", owner: str = ""):
super().__init__()
self.config_yaml = config_yaml
self.readme = readme
self.defaults = defaults
self.owner = owner
def run(self):
try:
@ -96,7 +109,7 @@ class PublishWorker(QThread):
client = RegistryClient()
client.token = config.registry.token
result = client.publish_tool(self.config_yaml, self.readme, self.defaults)
result = client.publish_tool(self.config_yaml, self.readme, self.defaults, owner=self.owner)
self.success.emit(result)
except Exception as e:
self.error.emit(str(e))
@ -108,7 +121,7 @@ class PublishDialog(QDialog):
def __init__(self, parent, tool: Tool):
super().__init__(parent)
self.setWindowTitle("Publish Tool")
self.setMinimumSize(500, 450)
self.setMinimumSize(500, 580)
self._tool = tool
self._worker = None
self._version_worker = None
@ -117,6 +130,8 @@ class PublishDialog(QDialog):
self._original_versions: List[str] = []
self._local_version = self._get_local_version()
self._original_owner = self._get_original_owner()
self._user_role: str = "user"
self._user_slug: str = ""
self._setup_ui()
self._fetch_registry_versions()
@ -124,7 +139,6 @@ class PublishDialog(QDialog):
"""Get the version from local tool config."""
if hasattr(self._tool, 'path') and self._tool.path:
try:
import yaml
config = yaml.safe_load(self._tool.path.read_text())
return config.get("version", "1.0.0")
except Exception:
@ -135,7 +149,6 @@ class PublishDialog(QDialog):
"""Get the original owner if this tool was installed from registry."""
if hasattr(self._tool, 'path') and self._tool.path:
try:
import yaml
config = yaml.safe_load(self._tool.path.read_text())
# Check if it has forked_from or was installed from registry
if config.get("forked_from"):
@ -158,7 +171,10 @@ class PublishDialog(QDialog):
self._my_registry_version = result.get("my_version")
self._my_registry_status = result.get("my_status")
self._original_versions = result.get("original_versions", [])
self._user_role = result.get("user_role", "user")
self._user_slug = result.get("user_slug", "")
self._update_version_info()
self._update_admin_fields()
def _on_versions_error(self, error: str):
"""Handle version fetch error (non-critical)."""
@ -202,6 +218,18 @@ class PublishDialog(QDialog):
"""Show or hide version bump buttons."""
self.bump_widget.setVisible(show)
def _update_admin_fields(self):
"""Show/hide admin-only fields based on user role."""
is_admin = self._user_role == "admin"
self.owner_widget.setVisible(is_admin)
self.owner_label.setVisible(is_admin)
if is_admin and self.owner_combo.count() == 0:
# Only populate once to avoid resetting user's selection
self.owner_combo.addItem(self._user_slug)
self.owner_combo.addItem("official")
self.owner_combo.setCurrentIndex(0)
def _bump_patch(self):
"""Bump patch version."""
if self._my_registry_version:
@ -263,6 +291,7 @@ class PublishDialog(QDialog):
self.version_input.setText(self._local_version)
self.version_input.setPlaceholderText("1.0.0")
self.version_input.setMaximumWidth(100)
self.version_input.setMinimumHeight(28)
version_row.addWidget(self.version_input)
# Bump buttons (hidden until we know registry version)
@ -273,21 +302,21 @@ class PublishDialog(QDialog):
btn_patch = QPushButton("+Patch")
btn_patch.setObjectName("secondary")
btn_patch.setFixedWidth(60)
btn_patch.setFixedWidth(75)
btn_patch.setToolTip("Bump patch version (bug fixes)")
btn_patch.clicked.connect(self._bump_patch)
bump_layout.addWidget(btn_patch)
btn_minor = QPushButton("+Minor")
btn_minor.setObjectName("secondary")
btn_minor.setFixedWidth(60)
btn_minor.setFixedWidth(75)
btn_minor.setToolTip("Bump minor version (new features)")
btn_minor.clicked.connect(self._bump_minor)
bump_layout.addWidget(btn_minor)
btn_major = QPushButton("+Major")
btn_major.setObjectName("secondary")
btn_major.setFixedWidth(60)
btn_major.setFixedWidth(75)
btn_major.setToolTip("Bump major version (breaking changes)")
btn_major.clicked.connect(self._bump_major)
bump_layout.addWidget(btn_major)
@ -298,14 +327,34 @@ class PublishDialog(QDialog):
version_row.addStretch()
form.addRow("Version:", version_row)
# Category
# Owner (admin only) - hidden by default
self.owner_widget = QWidget()
owner_layout = QHBoxLayout(self.owner_widget)
owner_layout.setContentsMargins(0, 0, 0, 0)
self.owner_combo = QComboBox()
self.owner_combo.setMinimumWidth(150)
self.owner_combo.setMinimumHeight(28)
# Make it editable but read-only to fix hover styling issues
self.owner_combo.setEditable(True)
self.owner_combo.lineEdit().setReadOnly(True)
owner_layout.addWidget(self.owner_combo)
owner_hint = QLabel("(admin only)")
owner_hint.setStyleSheet("color: #805ad5; font-size: 11px;")
owner_layout.addWidget(owner_hint)
owner_layout.addStretch()
self.owner_widget.setVisible(False) # Hidden until we know user is admin
self.owner_label = QLabel("Publish as:")
self.owner_label.setVisible(False)
form.addRow(self.owner_label, self.owner_widget)
# Category - populated from defaults + local tools
self.category_combo = QComboBox()
self.category_combo.setEditable(True)
categories = [
"Text Processing", "Code", "Data", "System",
"Writing", "Analysis", "Productivity", "Other"
]
self.category_combo.addItems(categories)
self.category_combo.setMinimumHeight(28)
self.category_combo.addItems(get_all_categories())
if self._tool.category:
idx = self.category_combo.findText(self._tool.category)
if idx >= 0:
@ -314,9 +363,18 @@ class PublishDialog(QDialog):
self.category_combo.setCurrentText(self._tool.category)
form.addRow("Category:", self.category_combo)
# Tags
# Tags - load from config.yaml if available
self.tags_input = QLineEdit()
self.tags_input.setMinimumHeight(28)
self.tags_input.setPlaceholderText("ai, text, productivity (comma separated)")
if self._tool.path:
try:
config_data = yaml.safe_load(self._tool.path.read_text()) or {}
existing_tags = config_data.get("tags", [])
if existing_tags:
self.tags_input.setText(", ".join(existing_tags))
except Exception:
pass # No existing tags or error reading
form.addRow("Tags:", self.tags_input)
layout.addLayout(form)
@ -328,7 +386,8 @@ class PublishDialog(QDialog):
self.desc_input = QTextEdit()
self.desc_input.setPlaceholderText("Describe what your tool does...")
self.desc_input.setPlainText(self._tool.description or "")
self.desc_input.setMaximumHeight(100)
self.desc_input.setMinimumHeight(80)
self.desc_input.setMaximumHeight(120)
layout.addWidget(self.desc_input)
# Status
@ -336,12 +395,18 @@ class PublishDialog(QDialog):
self.status_label.setStyleSheet("color: #718096;")
layout.addWidget(self.status_label)
# Progress
# Progress - use fixed height container so layout doesn't shift
self.progress = QProgressBar()
self.progress.setRange(0, 0)
self.progress.hide()
self.progress.setFixedHeight(20)
self.progress.setVisible(False)
layout.addWidget(self.progress)
# Spacer that matches progress bar height when hidden
self.progress_spacer = QWidget()
self.progress_spacer.setFixedHeight(20)
layout.addWidget(self.progress_spacer)
layout.addStretch()
# Buttons
@ -359,10 +424,81 @@ class PublishDialog(QDialog):
layout.addLayout(buttons)
def _check_unpublished_dependencies(self):
"""Check for unpublished dependencies."""
from ...collection import gather_local_unpublished_deps, DepCheckResult
try:
config = load_config()
client = RegistryClient()
client.token = config.registry.token
if not client.token:
# Can't check without auth, skip the check
return DepCheckResult([], [], [], [])
# Get user slug for proper path resolution
try:
me = client.get_me()
my_owner = me.get("slug", "")
except Exception:
my_owner = None
return gather_local_unpublished_deps([self._tool.name], client, my_owner)
except Exception:
return DepCheckResult([], [], [], [])
def _publish_with_dependencies(self, dep_result):
"""Publish dependencies first, then the main tool."""
from ...cli.collections_commands import _publish_single_tool
config = load_config()
client = RegistryClient()
client.token = config.registry.token
# Get tools to publish in order (only unpublished deps, exclude main tool)
tools_to_publish = [
t for t in dep_result.publish_order
if t in dep_result.unpublished and t != self._tool.name
]
if not tools_to_publish:
# No deps to publish, proceed with main publish
self._do_publish()
return
# Publish dependencies first
self.btn_publish.setEnabled(False)
self.btn_cancel.setEnabled(False)
self.progress.setVisible(True)
self.progress_spacer.setVisible(False)
failed = []
for tool_name in tools_to_publish:
self.status_label.setText(f"Publishing dependency: {tool_name}...")
result = _publish_single_tool(tool_name, client)
if not result.get("success"):
failed.append((tool_name, result.get("error", "Unknown error")))
if failed:
self.progress.setVisible(False)
self.progress_spacer.setVisible(True)
self.btn_publish.setEnabled(True)
self.btn_cancel.setEnabled(True)
errors = "\n".join(f" - {name}: {err}" for name, err in failed)
QMessageBox.warning(
self, "Dependency Publish Failed",
f"Failed to publish some dependencies:\n\n{errors}\n\n"
"Fix these issues and try again."
)
return
# Now publish the main tool
self.status_label.setText("Publishing main tool...")
self._do_publish()
def _publish(self):
"""Publish the tool."""
import yaml
version = self.version_input.text().strip()
if not version:
QMessageBox.warning(self, "Validation", "Version is required")
@ -373,6 +509,75 @@ class PublishDialog(QDialog):
QMessageBox.warning(self, "Validation", "Description is required")
return
# Check for unpublished dependencies
dep_result = self._check_unpublished_dependencies()
# Filter out the tool itself - we're about to publish it
unpublished_deps = [d for d in dep_result.unpublished if d != self._tool.name]
if unpublished_deps:
dialog = QMessageBox(self)
dialog.setIcon(QMessageBox.Warning)
dialog.setWindowTitle("Unpublished Dependencies")
message = "This tool depends on unpublished tools:\n\n"
message += "\n".join(f" - {t}" for t in unpublished_deps)
message += "\n\n"
if dep_result.cycles:
message += "Warning: Circular dependencies detected:\n"
for cycle in dep_result.cycles:
message += f" {' -> '.join(cycle)}\n"
message += "(Cyclic deps excluded from auto-publish)\n\n"
if dep_result.skipped:
message += f"Note: Could not check {len(dep_result.skipped)} dep(s).\n\n"
message += "Choose how to proceed:"
dialog.setText(message)
btn_publish = dialog.addButton("Publish Dependencies First", QMessageBox.AcceptRole)
btn_anyway = dialog.addButton("Publish Anyway", QMessageBox.DestructiveRole)
btn_cancel = dialog.addButton("Cancel", QMessageBox.RejectRole)
for btn in [btn_publish, btn_anyway, btn_cancel]:
btn.setMinimumWidth(btn.fontMetrics().horizontalAdvance(btn.text()) + 30)
dialog.exec()
clicked = dialog.clickedButton()
if clicked == btn_cancel:
return
elif clicked == btn_publish:
self._publish_with_dependencies(dep_result)
return
# btn_anyway falls through to normal publish
self._do_publish()
def _do_publish(self):
"""Actually perform the publish operation."""
version = self.version_input.text().strip()
description = self.desc_input.toPlainText().strip()
# Check if tool was previously rejected - if so, bump version
try:
config = load_config()
client = RegistryClient()
client.token = config.registry.token
if client.token:
status_info = client.get_my_tool_status(self._tool.name)
current_status = status_info.get("status", "")
if current_status == "rejected":
new_version = bump_version(version, "patch")
self.version_input.setText(new_version)
version = new_version
self.status_label.setText(f"Version bumped to {new_version} (previous was rejected)")
except RegistryError as e:
if e.code != "TOOL_NOT_FOUND":
pass # Other errors - continue
except Exception:
pass # Continue with publish
category = self.category_combo.currentText()
tags = [t.strip() for t in self.tags_input.text().split(",") if t.strip()]
@ -415,22 +620,29 @@ class PublishDialog(QDialog):
self.btn_publish.setEnabled(False)
self.btn_cancel.setEnabled(False)
self.progress.show()
self.progress.setVisible(True)
self.progress_spacer.setVisible(False)
self.status_label.setText("Publishing...")
# Store the config we're publishing so we can save it on success
self._published_config = tool_config
self._worker = PublishWorker(config_yaml, readme, defaults)
# Get owner if admin selected a different one
owner = ""
if self._user_role == "admin" and self.owner_widget.isVisible():
selected_text = self.owner_combo.currentText().strip()
if selected_text == "official":
owner = "official"
self._worker = PublishWorker(config_yaml, readme, defaults, owner=owner)
self._worker.success.connect(self._on_success)
self._worker.error.connect(self._on_error)
self._worker.start()
def _on_success(self, result: dict):
"""Handle publish success."""
import yaml
self.progress.hide()
self.progress.setVisible(False)
self.progress_spacer.setVisible(True)
self.status_label.setText("Published successfully!")
self.status_label.setStyleSheet("color: #38a169; font-weight: 600;")
@ -463,7 +675,8 @@ class PublishDialog(QDialog):
def _on_error(self, error: str):
"""Handle publish error."""
self.progress.hide()
self.progress.setVisible(False)
self.progress_spacer.setVisible(True)
self.btn_publish.setEnabled(True)
self.btn_cancel.setEnabled(True)
self.status_label.setText(f"Error: {error}")

6
src/cmdforge/gui/main_window.py
View File

@ -198,8 +198,10 @@ class MainWindow(QMainWindow):
self.pages.removeWidget(current)
current.deleteLater()
# Return to tools page
self.sidebar.setCurrentRow(0)
# Return to tools page - set both sidebar and stacked widget explicitly
# to avoid race conditions with widget removal
self.pages.setCurrentWidget(self.tools_page)
self.sidebar.setCurrentRow(1)
self.tools_page.refresh()
def _setup_shortcuts(self):

321
src/cmdforge/gui/pages/collections_page.py
View File

@ -15,9 +15,139 @@ from PySide6.QtGui import QFont
from ...collection import (
Collection, list_collections, get_collection, COLLECTIONS_DIR,
classify_tool_reference, resolve_tool_references
classify_tool_reference, resolve_tool_references, gather_local_unpublished_deps
)
from ...tool import list_tools
from ...config import load_config
class CollectionStatusSyncWorker(QThread):
"""Background worker to sync collection statuses from registry."""
collection_updated = Signal(str) # collection name
finished = Signal()
def __init__(self, pending_collections: list):
"""
Args:
pending_collections: List of (collection_name, pending_tools) tuples
"""
super().__init__()
self.pending_collections = pending_collections
def run(self):
from ...registry_client import get_client, RegistryError
try:
client = get_client()
# Validate token first
is_valid, _ = client.validate_token()
if not is_valid:
return
for coll_name, pending_tools in self.pending_collections:
try:
self._sync_collection(client, coll_name, pending_tools)
except Exception:
pass # Skip collections that fail to sync
except Exception:
pass # Silently fail - this is background sync
finally:
self.finished.emit()
def _sync_collection(self, client, coll_name: str, pending_tools: list):
"""Sync a single collection's pending tool statuses."""
from ...registry_client import RegistryError
coll = get_collection(coll_name)
if not coll or not coll.pending_approval:
return
all_approved = True
has_rejected = False
still_pending = []
for tool_name in pending_tools:
try:
status_info = client.get_my_tool_status(tool_name)
status = status_info.get("status", "pending")
if status == "approved":
pass # Good
elif status == "rejected":
has_rejected = True
all_approved = False
else:
# Still pending or changes_requested
still_pending.append(tool_name)
all_approved = False
except RegistryError:
# Tool not found or other error - treat as still pending
still_pending.append(tool_name)
all_approved = False
# Update collection state based on results
changed = False
if all_approved:
# All tools approved - now actually publish the collection to registry!
try:
user_slug = coll.maintainer
if not user_slug:
me = client.get_me()
user_slug = me.get("slug", "")
# Build registry refs
registry_refs = []
transformed_pinned = {}
for tool_ref in coll.tools:
if "/" in tool_ref:
registry_refs.append(tool_ref)
if tool_ref in coll.pinned:
transformed_pinned[tool_ref] = coll.pinned[tool_ref]
else:
full_ref = f"{user_slug}/{tool_ref}"
registry_refs.append(full_ref)
if tool_ref in coll.pinned:
transformed_pinned[full_ref] = coll.pinned[tool_ref]
# Publish collection
payload = {
"name": coll.name,
"display_name": coll.display_name,
"description": coll.description,
"maintainer": user_slug,
"tools": registry_refs,
"pinned": transformed_pinned,
"tags": coll.tags,
}
client.publish_collection(payload)
# Update local state
coll.published = True
coll.registry_name = coll.name
coll.pending_approval = False
coll.pending_tools = []
coll.maintainer = user_slug
changed = True
except Exception:
# Failed to publish - just clear pending state
coll.pending_approval = False
coll.pending_tools = []
changed = True
elif has_rejected:
# Some tools rejected - clear pending state so user can retry
coll.pending_approval = False
coll.pending_tools = []
changed = True
elif still_pending != pending_tools:
# Some tools approved, update the list
coll.pending_tools = still_pending
changed = True
if changed:
coll.save()
self.collection_updated.emit(coll_name)
class CollectionCreateDialog(QDialog):
@ -201,6 +331,7 @@ class PublishAnalysisWorker(QThread):
"error": None,
"user_slug": None,
"resolution": None,
"dep_result": None, # Transitive dependency check result
}
try:
@ -233,6 +364,13 @@ class PublishAnalysisWorker(QThread):
)
result["resolution"] = resolution
# Gather transitive dependencies from local tools in the collection
local_tool_names = [ref for ref in self.collection.tools if '/' not in ref]
if local_tool_names:
dep_result = gather_local_unpublished_deps(local_tool_names, client, user_slug)
result["dep_result"] = dep_result
result["success"] = True
self.finished.emit(result)
@ -350,6 +488,12 @@ class CollectionsPage(QWidget):
self._install_worker = None
self._pending_analysis = None # Stores analysis result for publish flow
# Status sync
self._sync_worker = None
self._poll_timer = None
self._has_pending_collections = False
self._syncing = False
def _setup_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(20, 20, 20, 20)
@ -474,9 +618,16 @@ class CollectionsPage(QWidget):
def refresh(self):
"""Refresh both local and registry collections."""
self._has_pending_collections = False
self._load_local_collections()
self._load_registry_collections()
# Start background sync for pending collections
self._start_background_sync()
# Manage polling timer based on pending state
self._manage_poll_timer()
def _load_local_collections(self):
"""Load local collections into the list."""
self.collection_list.clear()
@ -490,6 +641,7 @@ class CollectionsPage(QWidget):
# Show status indicator
if coll.pending_approval:
item.setText(f"{coll.display_name} (pending)")
self._has_pending_collections = True
elif coll.published:
item.setText(f"{coll.display_name} (published)")
@ -537,6 +689,109 @@ class CollectionsPage(QWidget):
except Exception as e:
self.registry_list.addItem(f"Error: {str(e)}")
def _start_background_sync(self):
"""Start background sync for collections with pending tool approvals."""
if self._syncing:
return
config = load_config()
if not config.registry.token:
return # No auth, can't check status
# Gather pending collections
pending_collections = []
for name in list_collections():
coll = get_collection(name)
if coll and coll.pending_approval and coll.pending_tools:
pending_collections.append((name, list(coll.pending_tools)))
if not pending_collections:
return
# Stop any existing sync
if self._sync_worker and self._sync_worker.isRunning():
self._sync_worker.wait(1000)
# Start new sync
self._sync_worker = CollectionStatusSyncWorker(pending_collections)
self._sync_worker.collection_updated.connect(self._on_collection_status_updated)
self._sync_worker.finished.connect(self._on_sync_finished)
self._syncing = True
self._sync_worker.start()
def _on_sync_finished(self):
"""Handle background sync completion."""
self._syncing = False
def _manage_poll_timer(self):
"""Start or stop the polling timer based on pending collections."""
config = load_config()
should_poll = self._has_pending_collections and config.registry.token
if should_poll:
if not self._poll_timer:
# Create timer that polls every 30 seconds
self._poll_timer = QTimer(self)
self._poll_timer.timeout.connect(self._poll_status)
if not self._poll_timer.isActive():
self._poll_timer.start(30000) # 30 seconds
else:
# No pending collections, stop polling
if self._poll_timer and self._poll_timer.isActive():
self._poll_timer.stop()
def _poll_status(self):
"""Timer callback to poll status for pending collections."""
if self._syncing:
return # Already syncing
# Check if we still have pending collections
pending_collections = []
for name in list_collections():
coll = get_collection(name)
if coll and coll.pending_approval and coll.pending_tools:
pending_collections.append((name, list(coll.pending_tools)))
if pending_collections:
# Stop existing sync if any
if self._sync_worker and self._sync_worker.isRunning():
self._sync_worker.wait(1000)
# Start new sync
self._sync_worker = CollectionStatusSyncWorker(pending_collections)
self._sync_worker.collection_updated.connect(self._on_collection_status_updated)
self._sync_worker.finished.connect(self._on_sync_finished)
self._syncing = True
self._sync_worker.start()
else:
# No more pending collections, stop timer
if self._poll_timer and self._poll_timer.isActive():
self._poll_timer.stop()
self._has_pending_collections = False
def _on_collection_status_updated(self, coll_name: str):
"""Handle background sync updating a collection's status."""
# Refresh the display
self._load_local_collections()
# Re-select the current item if it was the updated one
current = self.collection_list.currentItem()
if current and current.data(Qt.UserRole) == coll_name:
coll = get_collection(coll_name)
if coll:
self._show_collection_details(coll)
# Show status message
coll = get_collection(coll_name)
if coll:
if coll.pending_approval:
self.main_window.show_status(f"Collection '{coll_name}' still has pending tools")
elif coll.published:
self.main_window.show_status(f"Collection '{coll_name}' published to registry!")
self._load_registry_collections() # Refresh registry tab when auto-published
else:
self.main_window.show_status(f"Collection '{coll_name}' status updated (some tools may have been rejected)")
def _on_selection_changed(self, current, previous):
"""Handle collection selection change."""
if not current:
@ -784,32 +1039,75 @@ class CollectionsPage(QWidget):
)
return
# Handle unpublished local tools
# Gather all unpublished tools (collection tools + transitive deps)
dep_result = result.get("dep_result")
# Start with tools not in registry at all
collection_tools_unpub = list(resolution.local_unpublished)
# Also include tools that exist but aren't approved (rejected, pending, changes_requested)
for name, status, has_approved in resolution.local_published:
if status != "approved" and name not in collection_tools_unpub:
collection_tools_unpub.append(name)
dependency_tools_unpub = []
if dep_result and dep_result.unpublished:
for dep in dep_result.unpublished:
if dep not in collection_tools_unpub:
dependency_tools_unpub.append(dep)
all_unpublished = collection_tools_unpub + dependency_tools_unpub
# Handle unpublished local tools (both collection tools and their deps)
publish_list = []
skip_list = []
if resolution.local_unpublished:
tools_list = "\n".join(f" - {t}" for t in resolution.local_unpublished)
if all_unpublished:
# Build detailed message
message = "Some tools are not published yet.\n\n"
if collection_tools_unpub:
message += "Collection tools:\n"
message += "\n".join(f" - {t}" for t in collection_tools_unpub)
message += "\n\n"
if dependency_tools_unpub:
message += "Dependencies:\n"
message += "\n".join(f" - {t}" for t in dependency_tools_unpub)
message += "\n\n"
if dep_result and dep_result.cycles:
message += "Warning: Circular dependencies detected:\n"
for cycle in dep_result.cycles:
message += f" {' -> '.join(cycle)}\n"
message += "(Cyclic deps excluded from auto-publish)\n\n"
if dep_result and dep_result.skipped:
message += f"Note: Could not check {len(dep_result.skipped)} dep(s) due to errors.\n\n"
message += "Choose how to proceed:"
dialog = QMessageBox(self)
dialog.setIcon(QMessageBox.Warning)
dialog.setWindowTitle("Unpublished Tools")
dialog.setText(
"Some local tools are not published yet.\n\n"
f"{tools_list}\n\n"
"Choose how to proceed:"
)
dialog.setText(message)
btn_publish = dialog.addButton("Publish Tools First", QMessageBox.AcceptRole)
btn_skip = dialog.addButton("Skip Unpublished", QMessageBox.DestructiveRole)
btn_cancel = dialog.addButton("Cancel", QMessageBox.RejectRole)
dialog.setDefaultButton(btn_publish)
# Ensure buttons are wide enough for their text
for btn in [btn_publish, btn_skip, btn_cancel]:
btn.setMinimumWidth(btn.fontMetrics().horizontalAdvance(btn.text()) + 30)
dialog.exec()
clicked = dialog.clickedButton()
if clicked == btn_publish:
publish_list = resolution.local_unpublished
# Use topological order if available
if dep_result and dep_result.publish_order:
# Filter to only unpublished tools
publish_list = [t for t in dep_result.publish_order if t in all_unpublished]
else:
publish_list = all_unpublished
elif clicked == btn_skip:
skip_list = resolution.local_unpublished
skip_list = all_unpublished
else:
return
@ -854,6 +1152,7 @@ class CollectionsPage(QWidget):
if success:
QMessageBox.information(self, "Published", message)
self._load_local_collections()
self._load_registry_collections() # Also refresh registry tab
else:
QMessageBox.warning(self, "Publish Failed", message)
self.btn_publish.setEnabled(True)

4
src/cmdforge/gui/pages/registry_page.py
View File

@ -11,7 +11,7 @@ from PySide6.QtGui import QColor
from ...registry_client import RegistryClient, RegistryError
from ...config import load_config
from ...tool import list_tools, load_tool
from ...tool import list_tools, load_tool, get_all_categories
class SearchWorker(QThread):
@ -152,7 +152,7 @@ class RegistryPage(QWidget):
cat_label = QLabel("Category:")
filters_layout.addWidget(cat_label)
self.category_combo = QComboBox()
self.category_combo.addItems(["All", "Text", "Developer", "Data", "Other"])
self.category_combo.addItems(["All"] + get_all_categories())
self.category_combo.setMinimumWidth(100)
self.category_combo.setToolTip("Filter by tool category")
self.category_combo.currentTextChanged.connect(self._on_filter_changed)

8
src/cmdforge/gui/pages/tool_builder_page.py
View File

@ -13,7 +13,7 @@ from PySide6.QtCore import Qt
from ...tool import (
Tool, ToolArgument, PromptStep, CodeStep, ToolStep,
load_tool, save_tool, validate_tool_name, DEFAULT_CATEGORIES,
load_tool, save_tool, validate_tool_name, get_all_categories,
ensure_settings
)
from ..widgets.icons import get_prompt_icon, get_code_icon, get_tool_icon
@ -92,7 +92,7 @@ class ToolBuilderPage(QWidget):
self.category_combo = QComboBox()
self.category_combo.setEditable(True)
self.category_combo.setToolTip("Category for organizing tools (select or type custom)")
for cat in DEFAULT_CATEGORIES:
for cat in get_all_categories():
self.category_combo.addItem(cat)
info_layout.addRow("Category:", self.category_combo)
@ -1070,8 +1070,8 @@ class ToolBuilderPage(QWidget):
return
# Validate name
error = validate_tool_name(name)
if error:
is_valid, error = validate_tool_name(name)
if not is_valid:
QMessageBox.warning(self, "Validation", error)
return

109
src/cmdforge/gui/pages/tools_page.py
View File

@ -16,14 +16,14 @@ from PySide6.QtGui import QFont, QColor, QBrush, QShortcut, QKeySequence
from ...tool import (
Tool, ToolArgument, PromptStep, CodeStep, ToolStep,
list_tools, load_tool, delete_tool, DEFAULT_CATEGORIES,
list_tools, load_tool, delete_tool, get_all_categories,
get_tools_dir
)
from ...config import load_config
class StatusSyncWorker(QThread):
"""Background worker to sync tool statuses from registry."""
"""Background worker to sync tool statuses from registry using hash-based batch lookup."""
finished = Signal()
tool_updated = Signal(str) # Emits tool name when status changes
@ -32,8 +32,9 @@ class StatusSyncWorker(QThread):
self.tool_names = tool_names
def run(self):
"""Sync status for all tools with registry_hash."""
"""Sync status for all tools with registry_hash using batch hash lookup."""
from ...registry_client import RegistryClient, RegistryError
from collections import defaultdict
try:
config = load_config()
@ -43,43 +44,59 @@ class StatusSyncWorker(QThread):
client = RegistryClient()
client.token = config.registry.token
# Collect all tools that have a registry_hash
# Multiple tools can share the same hash (copies/forks)
tools_dir = get_tools_dir()
tool_hashes = defaultdict(list) # hash -> [(tool_name, config_path), ...]
for tool_name in self.tool_names:
config_path = tools_dir / tool_name / "config.yaml"
if not config_path.exists():
continue
try:
self._sync_tool(client, tool_name)
config_data = yaml.safe_load(config_path.read_text()) or {}
h = config_data.get("registry_hash")
if h:
tool_hashes[h].append((tool_name, config_path))
except Exception:
pass # Skip tools that fail to sync
continue
if not tool_hashes:
return
# Batch requests in chunks of 100 (server limit)
all_hashes = list(tool_hashes.keys())
results = {}
for i in range(0, len(all_hashes), 100):
chunk = all_hashes[i:i + 100]
results.update(client.get_tool_status_by_hashes(chunk))
# Update local configs from results
for h, tool_entries in tool_hashes.items():
if h in results:
for tool_name, config_path in tool_entries:
try:
self._update_local_config(tool_name, config_path, results[h])
except Exception:
pass
except Exception:
pass # Silently fail - this is background sync
finally:
self.finished.emit()
def _sync_tool(self, client, tool_name: str):
"""Sync a single tool's status."""
config_path = get_tools_dir() / tool_name / "config.yaml"
if not config_path.exists():
return
def _update_local_config(self, tool_name: str, config_path, status_data: dict):
"""Update a tool's local config with registry status."""
config_data = yaml.safe_load(config_path.read_text()) or {}
if not config_data.get("registry_hash"):
return # Not published
# Get status from registry
status_data = client.get_my_tool_status(tool_name)
new_status = status_data.get("status", "pending")
new_hash = status_data.get("config_hash")
new_feedback = status_data.get("feedback")
old_status = config_data.get("registry_status", "pending")
old_hash = config_data.get("registry_hash")
old_feedback = config_data.get("registry_feedback")
changed = False
if old_status != new_status:
config_data["registry_status"] = new_status
changed = True
if new_hash and old_hash != new_hash:
config_data["registry_hash"] = new_hash
changed = True
if new_feedback != old_feedback:
if new_feedback:
config_data["registry_feedback"] = new_feedback
@ -102,6 +119,7 @@ def get_tool_publish_state(tool_name: str) -> Tuple[str, Optional[str]]:
- "pending" - submitted but awaiting moderation
- "changes_requested" - admin requested changes before approval
- "rejected" - rejected by admin
- "installed" - installed from registry (has hash but no explicit status)
- "local" - no registry_hash (never published)
"""
config_path = get_tools_dir() / tool_name / "config.yaml"
@ -111,20 +129,27 @@ def get_tool_publish_state(tool_name: str) -> Tuple[str, Optional[str]]:
try:
config = yaml.safe_load(config_path.read_text())
registry_hash = config.get("registry_hash")
registry_status = config.get("registry_status", "pending")
if not registry_hash:
return ("local", None)
# Return the moderation status directly - no local hash comparison
# Only use explicit registry_status - don't default to "pending"
registry_status = config.get("registry_status")
if not registry_status:
# Has registry_hash but no status = installed from registry
return ("installed", registry_hash)
if registry_status == "approved":
return ("published", registry_hash)
elif registry_status == "changes_requested":
return ("changes_requested", registry_hash)
elif registry_status == "rejected":
return ("rejected", registry_hash)
else:
elif registry_status == "pending":
return ("pending", registry_hash)
else:
return ("installed", registry_hash)
except Exception:
return ("local", None)
@ -285,11 +310,8 @@ class ToolsPage(QWidget):
category = tool.category if tool.category else "Other"
tools_by_category[category].append((name, tool))
# Build tree with categories
all_categories = list(DEFAULT_CATEGORIES)
for cat in tools_by_category:
if cat not in all_categories:
all_categories.append(cat)
# Build tree with categories (get_all_categories returns defaults + custom)
all_categories = get_all_categories()
for category in all_categories:
if category in tools_by_category and tools_by_category[category]:
@ -327,6 +349,10 @@ class ToolsPage(QWidget):
display_name = f"{name} ✗"
tooltip = "Rejected by moderator"
color = QColor(220, 38, 38) # Red
elif state == "installed":
display_name = f"{name} ↓"
tooltip = "Installed from registry"
color = QColor(56, 178, 172) # Teal
else:
display_name = name
tooltip = "Local tool - not published"
@ -475,6 +501,13 @@ class ToolsPage(QWidget):
self._show_tool_info(tool)
self._update_buttons()
def _get_qualified_name(self) -> Optional[str]:
"""Get the qualified name (e.g., 'official/summarize') of the selected tool."""
items = self.tool_tree.selectedItems()
if items:
return items[0].data(0, Qt.UserRole)
return self._current_tool.name if self._current_tool else None
def _on_double_click(self, item, column):
"""Handle double-click on tool."""
tool_name = item.data(0, Qt.UserRole)
@ -509,6 +542,12 @@ class ToolsPage(QWidget):
"border-radius: 4px; margin-bottom: 12px; font-size: 12px;'>"
"✓ Published to registry - approved</p>"
)
elif state == "installed":
lines.append(
"<p style='background: #e2e8f0; color: #4a5568; padding: 6px 10px; "
"border-radius: 4px; margin-bottom: 12px; font-size: 12px;'>"
"↓ Installed from registry</p>"
)
elif state == "pending":
lines.append(
"<p style='background: #fef3c7; color: #92400e; padding: 6px 10px; "
@ -622,7 +661,7 @@ class ToolsPage(QWidget):
def _edit_tool(self):
"""Edit the selected tool."""
if self._current_tool:
self.main_window.open_tool_builder(self._current_tool.name)
self.main_window.open_tool_builder(self._get_qualified_name())
def _configure_tool(self):
"""Open settings dialog for the selected tool."""
@ -639,19 +678,21 @@ class ToolsPage(QWidget):
return
from ..dialogs.settings_dialog import SettingsDialog
dialog = SettingsDialog(self, self._current_tool.name)
qualified_name = self._get_qualified_name()
dialog = SettingsDialog(self, qualified_name)
if dialog.exec():
self.main_window.show_status(f"Settings saved for '{self._current_tool.name}'")
self.main_window.show_status(f"Settings saved for '{qualified_name}'")
def _delete_tool(self):
"""Delete the selected tool."""
if not self._current_tool:
return
qualified_name = self._get_qualified_name()
reply = QMessageBox.question(
self,
"Delete Tool",
f"Are you sure you want to delete '{self._current_tool.name}'?\n\n"
f"Are you sure you want to delete '{qualified_name}'?\n\n"
"This will remove the tool configuration and wrapper script.",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No
@ -659,8 +700,8 @@ class ToolsPage(QWidget):
if reply == QMessageBox.Yes:
try:
delete_tool(self._current_tool.name)
self.main_window.show_status(f"Deleted tool '{self._current_tool.name}'")
delete_tool(qualified_name)
self.main_window.show_status(f"Deleted tool '{qualified_name}'")
self.refresh()
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to delete tool:\n{e}")

66
src/cmdforge/registry/app.py
View File

@ -2385,7 +2385,34 @@ def create_app() -> Flask:
if len(str(tag)) > MAX_TAG_LEN:
return error_response("VALIDATION_ERROR", "Tag exceeds 32 characters")
# Determine owner - admins can publish as "official" or other owners
owner = g.current_publisher["slug"]
requested_owner = payload.get("owner", "").strip()
if requested_owner and requested_owner != owner:
# Only admins can publish as a different owner
if g.current_publisher.get("role") != "admin":
return error_response(
"FORBIDDEN",
"Only admins can publish as a different owner",
403
)
# Validate the requested owner
if requested_owner == "official":
owner = "official"
else:
# Check if it's an existing publisher
existing_pub = query_one(
g.db,
"SELECT slug FROM publishers WHERE slug = ?",
[requested_owner]
)
if not existing_pub:
return error_response(
"VALIDATION_ERROR",
f"Owner '{requested_owner}' does not exist. Use 'official' or a valid publisher slug.",
400
)
owner = requested_owner
# Compute config hash early for idempotency check
config_hash = compute_yaml_hash(config_text)
@ -2650,6 +2677,45 @@ def create_app() -> Flask:
return jsonify({"data": result})
@app.route("/api/v1/tools/status-by-hash", methods=["POST"])
@require_token
def tool_status_by_hash() -> Response:
"""Look up tool statuses by config hash. Batch endpoint scoped to current publisher."""
data = request.get_json(silent=True) or {}
hashes = data.get("hashes", [])
if not hashes or not isinstance(hashes, list):
return error_response("VALIDATION_ERROR", "hashes list is required", 400)
if len(hashes) > 100:
return error_response("VALIDATION_ERROR", "Max 100 hashes per request", 400)
results = {}
for h in hashes:
row = query_one(
g.db,
"""
SELECT owner, name, version, moderation_status, moderation_note, config_hash
FROM tools
WHERE config_hash = ? AND publisher_id = ?
ORDER BY published_at DESC
LIMIT 1
""",
[h, g.current_publisher["id"]],
)
if row:
entry = {
"owner": row["owner"],
"name": row["name"],
"version": row["version"],
"status": row["moderation_status"],
"config_hash": row["config_hash"],
}
if row["moderation_status"] in ("changes_requested", "rejected") and row["moderation_note"]:
entry["feedback"] = row["moderation_note"]
results[h] = entry
return jsonify({"data": results})
@app.route("/api/v1/tools/<owner>/<name>/deprecate", methods=["POST"])
@require_token
def deprecate_tool(owner: str, name: str) -> Response:

27
src/cmdforge/registry_client.py
View File

@ -611,7 +611,8 @@ class RegistryClient:
readme: str = "",
defaults: str = "",
dry_run: bool = False,
visibility: str = "public"
visibility: str = "public",
owner: str = ""
) -> Dict[str, Any]:
"""
Publish a tool to the registry.
@ -622,6 +623,7 @@ class RegistryClient:
defaults: Default settings YAML content (defaults.yaml)
dry_run: If True, validate without publishing
visibility: Tool visibility - "public", "private", or "unlisted"
owner: Optional owner override (admin only, use "official" for official tools)
Returns:
Dict with PR URL or validation results
@ -634,6 +636,8 @@ class RegistryClient:
}
if defaults:
payload["defaults"] = defaults
if owner:
payload["owner"] = owner
response = self._request(
"POST",
@ -717,6 +721,27 @@ class RegistryClient:
return response.json().get("data", {})
def get_tool_status_by_hashes(self, hashes: List[str]) -> Dict[str, Dict[str, Any]]:
"""
Look up tool statuses by config hash (batch).
Args:
hashes: List of config hash strings
Returns:
Dict mapping hash -> status info (owner, name, version, status, config_hash, feedback)
"""
response = self._request(
"POST", "/tools/status-by-hash",
json_data={"hashes": hashes},
require_auth=True
)
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def get_me(self) -> Dict[str, Any]:
"""
Get current user info.

23
src/cmdforge/tool.py
View File

@ -235,6 +235,29 @@ class ToolSource:
DEFAULT_CATEGORIES = ["Text", "Developer", "Data", "Other"]
def get_all_categories() -> list:
"""
Get all unique categories from default list and local tools.
Returns sorted list with defaults first, then custom categories.
"""
categories = set(DEFAULT_CATEGORIES)
# Add categories from local tools
try:
for name in list_tools():
tool = load_tool(name)
if tool and tool.category:
categories.add(tool.category)
except Exception:
pass # If tool loading fails, just use defaults
# Sort: defaults first (in order), then others alphabetically
defaults = [c for c in DEFAULT_CATEGORIES if c in categories]
others = sorted([c for c in categories if c not in DEFAULT_CATEGORIES])
return defaults + others
@dataclass
class Tool:
"""A CmdForge tool definition."""

263
tests/test_collection.py
View File

@ -9,7 +9,8 @@ import yaml
from cmdforge.collection import (
Collection, list_collections, get_collection, COLLECTIONS_DIR,
classify_tool_reference, resolve_tool_references, ToolResolutionResult
classify_tool_reference, resolve_tool_references, ToolResolutionResult,
gather_local_unpublished_deps, DepCheckResult
)
@ -296,3 +297,263 @@ class TestToolResolutionResult:
assert len(result.registry_refs) == 2
assert len(result.local_unpublished) == 1
assert len(result.visibility_issues) == 1
class TestDepCheckResult:
"""Tests for DepCheckResult dataclass."""
def test_empty_result(self):
result = DepCheckResult(
unpublished=[],
publish_order=[],
cycles=[],
skipped=[]
)
assert len(result.unpublished) == 0
assert len(result.cycles) == 0
def test_with_data(self):
result = DepCheckResult(
unpublished=["tool-a", "tool-b"],
publish_order=["tool-b", "tool-a", "main"],
cycles=[["x", "y", "x"]],
skipped=["tool-z"]
)
assert result.unpublished == ["tool-a", "tool-b"]
assert result.publish_order == ["tool-b", "tool-a", "main"]
assert result.cycles == [["x", "y", "x"]]
assert result.skipped == ["tool-z"]
class TestGatherLocalUnpublishedDeps:
"""Tests for transitive dependency checking."""
@pytest.fixture
def mock_client(self):
"""Create a mock registry client."""
client = MagicMock()
client.get_me.return_value = {"slug": "testuser"}
return client
@pytest.fixture
def temp_tools_dir(self, tmp_path):
"""Create a temporary tools directory."""
with patch('cmdforge.tool.TOOLS_DIR', tmp_path):
yield tmp_path
def _create_tool(self, tools_dir, name, deps=None, tool_steps=None):
"""Helper to create a tool config in the temp dir."""
tool_dir = tools_dir / name
tool_dir.mkdir(parents=True, exist_ok=True)
config = {
"name": name,
"description": f"Test tool {name}",
"steps": [],
"output": "{input}"
}
if deps:
config["dependencies"] = deps
if tool_steps:
for step in tool_steps:
config["steps"].append({
"type": "tool",
"tool": step,
"output_var": "result"
})
(tool_dir / "config.yaml").write_text(yaml.dump(config))
def test_no_dependencies_approved(self, mock_client, temp_tools_dir):
"""Approved tool with no deps should return empty unpublished list."""
mock_client.get_my_tool_status.return_value = {"status": "approved"}
self._create_tool(temp_tools_dir, "standalone")
result = gather_local_unpublished_deps(["standalone"], mock_client, "testuser")
assert result.unpublished == []
assert result.cycles == []
def test_no_dependencies_not_published(self, mock_client, temp_tools_dir):
"""Unpublished tool with no deps should be in unpublished list."""
from cmdforge.registry_client import RegistryError
mock_client.get_my_tool_status.side_effect = RegistryError(
code="TOOL_NOT_FOUND", message="Not found"
)
self._create_tool(temp_tools_dir, "standalone")
result = gather_local_unpublished_deps(["standalone"], mock_client, "testuser")
# The original tool itself is now checked and included if not approved
assert "standalone" in result.unpublished
assert result.cycles == []
def test_gathers_explicit_deps(self, mock_client, temp_tools_dir):
"""Should find explicit dependencies."""
from cmdforge.registry_client import RegistryError
# dep-tool is unpublished
mock_client.get_my_tool_status.side_effect = RegistryError(
code="TOOL_NOT_FOUND", message="Not found"
)
self._create_tool(temp_tools_dir, "dep-tool")
self._create_tool(temp_tools_dir, "main-tool", deps=["dep-tool"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
assert "dep-tool" in result.unpublished
# dep-tool should come before main-tool in publish order
assert result.publish_order.index("dep-tool") < result.publish_order.index("main-tool")
def test_gathers_tool_step_deps(self, mock_client, temp_tools_dir):
"""Should find implicit dependencies from ToolStep."""
from cmdforge.registry_client import RegistryError
mock_client.get_my_tool_status.side_effect = RegistryError(
code="TOOL_NOT_FOUND", message="Not found"
)
self._create_tool(temp_tools_dir, "called-tool")
self._create_tool(temp_tools_dir, "caller-tool", tool_steps=["called-tool"])
result = gather_local_unpublished_deps(["caller-tool"], mock_client, "testuser")
assert "called-tool" in result.unpublished
def test_gathers_transitive_deps(self, mock_client, temp_tools_dir):
"""Should find dependencies of dependencies."""
from cmdforge.registry_client import RegistryError
mock_client.get_my_tool_status.side_effect = RegistryError(
code="TOOL_NOT_FOUND", message="Not found"
)
# a -> b -> c
self._create_tool(temp_tools_dir, "tool-c")
self._create_tool(temp_tools_dir, "tool-b", deps=["tool-c"])
self._create_tool(temp_tools_dir, "tool-a", deps=["tool-b"])
result = gather_local_unpublished_deps(["tool-a"], mock_client, "testuser")
assert "tool-b" in result.unpublished
assert "tool-c" in result.unpublished
# Order: c before b before a
assert result.publish_order.index("tool-c") < result.publish_order.index("tool-b")
assert result.publish_order.index("tool-b") < result.publish_order.index("tool-a")
def test_skips_qualified_refs(self, mock_client, temp_tools_dir):
"""Should skip owner/name registry references but check main tool."""
from cmdforge.registry_client import RegistryError
mock_client.get_my_tool_status.side_effect = RegistryError(
code="TOOL_NOT_FOUND", message="Not found"
)
# Tool depends on a registry tool
self._create_tool(temp_tools_dir, "main-tool", deps=["official/summarize"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
# Registry refs should NOT be in unpublished
assert "official/summarize" not in result.unpublished
# But the main tool itself is checked and included since not found
assert "main-tool" in result.unpublished
def test_skips_qualified_refs_with_approved_main(self, mock_client, temp_tools_dir):
"""Should skip owner/name refs; approved main tool = empty unpublished."""
mock_client.get_my_tool_status.return_value = {"status": "approved"}
# Tool depends on a registry tool
self._create_tool(temp_tools_dir, "main-tool", deps=["official/summarize"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
assert "official/summarize" not in result.unpublished
assert result.unpublished == []
def test_detects_cycles(self, mock_client, temp_tools_dir):
"""Should detect and report circular dependencies."""
from cmdforge.registry_client import RegistryError
mock_client.get_my_tool_status.side_effect = RegistryError(
code="TOOL_NOT_FOUND", message="Not found"
)
# a -> b -> a (cycle)
self._create_tool(temp_tools_dir, "tool-a", deps=["tool-b"])
self._create_tool(temp_tools_dir, "tool-b", deps=["tool-a"])
result = gather_local_unpublished_deps(["tool-a"], mock_client, "testuser")
assert len(result.cycles) > 0
# Cyclic nodes should NOT be in unpublished list
assert "tool-a" not in result.unpublished
assert "tool-b" not in result.unpublished
def test_distinguishes_404_from_network_error(self, mock_client, temp_tools_dir):
"""404 = unpublished, other errors = skipped."""
from cmdforge.registry_client import RegistryError
def status_side_effect(name):
if name == "dep-404":
raise RegistryError(code="TOOL_NOT_FOUND", message="Not found")
elif name == "dep-network":
raise RegistryError(code="CONNECTION_ERROR", message="Network error")
return {"status": "approved"}
mock_client.get_my_tool_status.side_effect = status_side_effect
self._create_tool(temp_tools_dir, "dep-404")
self._create_tool(temp_tools_dir, "dep-network")
self._create_tool(temp_tools_dir, "main", deps=["dep-404", "dep-network"])
result = gather_local_unpublished_deps(["main"], mock_client, "testuser")
assert "dep-404" in result.unpublished
assert "dep-network" in result.skipped
def test_published_dep_not_in_unpublished(self, mock_client, temp_tools_dir):
"""Already approved deps should not be in unpublished list."""
mock_client.get_my_tool_status.return_value = {"status": "approved"}
self._create_tool(temp_tools_dir, "published-dep")
self._create_tool(temp_tools_dir, "main-tool", deps=["published-dep"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
assert "published-dep" not in result.unpublished
assert result.unpublished == []
def test_rejected_dep_in_unpublished(self, mock_client, temp_tools_dir):
"""Rejected deps should be in unpublished list (need republishing)."""
mock_client.get_my_tool_status.return_value = {"status": "rejected"}
self._create_tool(temp_tools_dir, "rejected-dep")
self._create_tool(temp_tools_dir, "main-tool", deps=["rejected-dep"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
assert "rejected-dep" in result.unpublished
def test_pending_dep_in_unpublished(self, mock_client, temp_tools_dir):
"""Pending deps should be in unpublished list (not yet usable)."""
mock_client.get_my_tool_status.return_value = {"status": "pending"}
self._create_tool(temp_tools_dir, "pending-dep")
self._create_tool(temp_tools_dir, "main-tool", deps=["pending-dep"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
assert "pending-dep" in result.unpublished
def test_changes_requested_dep_in_unpublished(self, mock_client, temp_tools_dir):
"""Deps with changes_requested should be in unpublished list."""
mock_client.get_my_tool_status.return_value = {"status": "changes_requested"}
self._create_tool(temp_tools_dir, "needs-changes-dep")
self._create_tool(temp_tools_dir, "main-tool", deps=["needs-changes-dep"])
result = gather_local_unpublished_deps(["main-tool"], mock_client, "testuser")
assert "needs-changes-dep" in result.unpublished