Add automatic background sync for tool statuses

When the Tools page loads, automatically sync statuses for all published
tools in the background. This means users don't need to manually click
"Sync Status" - the indicators will update automatically.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
rob 2026-01-16 08:11:17 -04:00
parent 7f976d3f8c
commit f22e95d119
1 changed files with 113 additions and 1 deletions

View File

@ -11,7 +11,7 @@ from PySide6.QtWidgets import (
QTreeWidget, QTreeWidgetItem, QTextEdit, QLabel,
QPushButton, QGroupBox, QMessageBox, QFrame
)
from PySide6.QtCore import Qt
from PySide6.QtCore import Qt, QThread, Signal
from PySide6.QtGui import QFont, QColor, QBrush
from ...tool import (
@ -23,6 +23,68 @@ from ...config import load_config
from ...hash_utils import compute_config_hash
class StatusSyncWorker(QThread):
"""Background worker to sync tool statuses from registry."""
finished = Signal()
tool_updated = Signal(str) # Emits tool name when status changes
def __init__(self, tool_names: list):
super().__init__()
self.tool_names = tool_names
def run(self):
"""Sync status for all tools with registry_hash."""
from ...registry_client import RegistryClient, RegistryError
try:
config = load_config()
if not config.registry.token:
return
client = RegistryClient()
client.token = config.registry.token
for tool_name in self.tool_names:
try:
self._sync_tool(client, tool_name)
except Exception:
pass # Skip tools that fail to sync
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
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")
old_status = config_data.get("registry_status", "pending")
old_hash = config_data.get("registry_hash")
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 changed:
config_path.write_text(yaml.dump(config_data, default_flow_style=False, sort_keys=False))
self.tool_updated.emit(tool_name)
def get_tool_publish_state(tool_name: str) -> Tuple[str, Optional[str]]:
"""
Get the publish state of a tool.
@ -75,6 +137,8 @@ class ToolsPage(QWidget):
super().__init__()
self.main_window = main_window
self._current_tool = None
self._sync_worker = None
self._syncing = False # Prevent re-sync during update
self._setup_ui()
self.refresh()
@ -279,6 +343,54 @@ class ToolsPage(QWidget):
"or browse the Registry to install community tools."
)
# Start background sync for published tools
self._start_background_sync(tools)
def _start_background_sync(self, tools: list):
"""Start background sync for tools that have been published."""
if self._syncing:
return # Avoid re-syncing during an update
config = load_config()
if not config.registry.token:
return # Not connected
# Collect tools with registry_hash
published_tools = []
for name in tools:
config_path = get_tools_dir() / name / "config.yaml"
if config_path.exists():
try:
config_data = yaml.safe_load(config_path.read_text()) or {}
if config_data.get("registry_hash"):
published_tools.append(name)
except Exception:
pass
if not published_tools:
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 = StatusSyncWorker(published_tools)
self._sync_worker.tool_updated.connect(self._on_tool_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 _on_tool_status_updated(self, tool_name: str):
"""Handle background sync updating a tool's status."""
# Refresh the display - _syncing flag prevents re-triggering sync
self.refresh()
self.main_window.show_status(f"Status updated for '{tool_name}'")
def _on_selection_changed(self):
"""Handle tool selection change."""
items = self.tool_tree.selectedItems()