feat: add idempotency guard, env config overrides, and param_count

- execution_history: add find_duplicate_run() for job_id dedup and
  job_id field in run records to prevent agent retry storms
- runtime_config: support COMFYUI_SERVER_URL and COMFYUI_SERVER_ID
  env vars to override config.json (useful for Docker/CI)
- services: expose param_count in WorkflowSummary so agents can
  assess workflow complexity before execution

Co-Authored-By: HuangYuChuh <HuangYuChuh@users.noreply.github.com>

scripts/shared/execution_history.py

@@ -21,6 +21,23 @@ def file_sha256(path: Path) -> str:
     return digest.hexdigest()
 
 
+def find_duplicate_run(server_id: str, workflow_id: str, job_id: str) -> dict[str, Any] | None:
+    """Check if a run with this job_id already exists (idempotency guard)."""
+    if not job_id:
+        return None
+    history_dir = get_server_history_dir(server_id, workflow_id)
+    if not history_dir.exists():
+        return None
+    for path in history_dir.glob("*.json"):
+        try:
+            payload = load_json(path)
+        except Exception:
+            continue
+        if isinstance(payload, dict) and payload.get("job_id") == job_id:
+            return payload
+    return None
+
+
 def build_run_record(
     server_id: str,
     workflow_id: str,
@@ -28,9 +45,12 @@ def build_run_record(
     raw_args: dict[str, Any],
     workflow_path: Path,
     schema_path: Path,
+    *,
+    job_id: str = "",
 ) -> dict[str, Any]:
     timestamp = utc_now_iso()
     return {
+        "job_id": job_id,
         "run_id": run_id,
         "server_id": server_id,
         "workflow_id": workflow_id,

scripts/shared/runtime_config.py

@@ -125,19 +125,44 @@ def migrate_workflow_storage_layout() -> None:
         _migrate_server_storage_layout(server_dir.name)
 
 
+def _apply_env_overrides(config: dict[str, object]) -> dict[str, object]:
+    """Apply environment variable overrides to server config.
+
+    Supported:
+      COMFYUI_SERVER_URL   — override default server's URL
+      COMFYUI_SERVER_ID    — override which server is default
+    """
+    import os
+
+    env_url = os.environ.get("COMFYUI_SERVER_URL")
+    env_default = os.environ.get("COMFYUI_SERVER_ID")
+
+    if env_default:
+        config["default_server"] = env_default
+
+    if env_url:
+        default_id = str(config.get("default_server", DEFAULT_SERVER_ID))
+        for server in config.get("servers", []):
+            if isinstance(server, dict) and server.get("id") == default_id:
+                server["url"] = env_url
+                break
+
+    return config
+
+
 def get_runtime_config() -> dict[str, object]:
     """Load the full multi-server config, migrating from legacy format if needed."""
     migrate_legacy_config()
     migrate_workflow_storage_layout()
 
     if not CONFIG_PATH.exists():
-        return default_config()
+        return _apply_env_overrides(default_config())
 
     loaded = load_json(CONFIG_PATH)
     if not isinstance(loaded, dict) or "servers" not in loaded:
-        return default_config()
+        return _apply_env_overrides(default_config())
 
-    return loaded
+    return _apply_env_overrides(loaded)
 
 
 def get_server_by_id(server_id: str) -> dict[str, object] | None:

ui/services.py

@@ -75,6 +75,7 @@ class WorkflowSummary:
     source_label: str = ""
     tags: list[str] = field(default_factory=list)
     has_history: bool = False
+    param_count: int = 0
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -87,6 +88,7 @@ class WorkflowSummary:
             "origin": self.origin,
             "source_label": self.source_label,
             "tags": self.tags,
+            "param_count": self.param_count,
             # Keep the key for backward compatibility with older bundles,
             # but always disable the list-level history entry point.
             "has_history": False,
@@ -230,6 +232,7 @@ class UIStorageService:
                 origin = ""
                 source_label = ""
                 tags: list[str] = []
+                param_count = 0
                 try:
                     schema_data = _read_json(schema_path, fallback={})
                     if isinstance(schema_data, dict):
@@ -240,6 +243,9 @@ class UIStorageService:
                         raw_tags = schema_data.get("tags")
                         if isinstance(raw_tags, list):
                             tags = [str(tag) for tag in raw_tags if str(tag).strip()]
+                        params = schema_data.get("parameters")
+                        if isinstance(params, dict):
+                            param_count = len(params)
                 except Exception:
                     enabled = True
 
@@ -257,6 +263,7 @@ class UIStorageService:
                     source_label=source_label,
                     tags=tags,
                     has_history=False,
+                    param_count=param_count,
                 ))
 
             server_workflows.sort(