From ace814b2ea159010897ecc60f15eaf7b5c32a04e Mon Sep 17 00:00:00 2001 From: hannahblair Date: Mon, 20 Jul 2026 15:57:13 +0100 Subject: [PATCH 01/13] add workflow run history persistence --- gradio/routes.py | 2 +- gradio/workflow.py | 171 +++++- gradio/workflow_api.py | 66 ++- gradio/workflow_history.py | 308 +++++++++++ .../workflow/WorkflowCanvas.svelte | 123 +++- .../workflow/WorkflowHistoryConnect.svelte | 301 ++++++++++ .../workflow/WorkflowHistoryPanel.svelte | 523 ++++++++++++++++++ .../workflow/workflow-executor.ts | 2 +- .../workflow/workflow-migration.ts | 12 +- test/test_workflow_history.py | 323 +++++++++++ 10 files changed, 1818 insertions(+), 13 deletions(-) create mode 100644 gradio/workflow_history.py create mode 100644 js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte create mode 100644 js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte create mode 100644 test/test_workflow_history.py diff --git a/gradio/routes.py b/gradio/routes.py index 8af6be360f7..095b800be84 100644 --- a/gradio/routes.py +++ b/gradio/routes.py @@ -1405,7 +1405,7 @@ async def queue_join_helper( ) error_map = { "queue_full": status.HTTP_503_SERVICE_UNAVAILABLE, - "validator_error": status.HTTP_422_UNPROCESSABLE_CONTENT, + "validator_error": getattr(status, "HTTP_422_UNPROCESSABLE_CONTENT", status.HTTP_422_UNPROCESSABLE_ENTITY), "error": status.HTTP_400_BAD_REQUEST, "success": status.HTTP_200_OK, } diff --git a/gradio/workflow.py b/gradio/workflow.py index 21bf1cf50df..4197c649fac 100644 --- a/gradio/workflow.py +++ b/gradio/workflow.py @@ -72,6 +72,11 @@ class _CuratedCache(TypedDict): _CURATED_CACHE: _CuratedCache = {"fetched_at": 0.0, "items": None} _CURATED_LOCK = threading.Lock() +# pipeline_tag lookup cache: model_id → tag string, avoids re-hitting Hub on +# every execution when the node was added without a stored pipeline_tag. +_MODEL_TAG_CACHE: dict[str, str] = {} +_MODEL_TAG_LOCK = threading.Lock() + def _bundled_snapshot_path() -> str: return os.path.join(os.path.dirname(__file__), "_workflow_curated_snapshot.json") @@ -611,6 +616,22 @@ def call_model( provider = data[4] if len(data) > 4 and data[4] else "auto" client = InferenceClient(model=model_id, token=hf_token, provider=provider) args = json.loads(args_json) + + # Autodetect the pipeline_tag from the Hub when the node doesn't carry + # one, so we never mis-route a text-to-image model as text-generation. + if not pipeline_tag: + with _MODEL_TAG_LOCK: + pipeline_tag = _MODEL_TAG_CACHE.get(model_id) + if not pipeline_tag: + try: + info = HfApi(token=hf_token).model_info(model_id) + pipeline_tag = info.pipeline_tag or "" + if pipeline_tag: + with _MODEL_TAG_LOCK: + _MODEL_TAG_CACHE[model_id] = pipeline_tag + except Exception: + pass + task = pipeline_tag or "text-generation" a0 = args[0] if args else "" a1 = args[1] if len(args) > 1 else "" @@ -1267,6 +1288,7 @@ def __init__( *, bind: dict[str, Callable] | list[Callable] | None = None, edges: list[tuple[str, str]] | None = None, + history: str | bool | None = None, ): """ Parameters: @@ -1289,6 +1311,11 @@ def __init__( ("shout", "reverse"), # first output → first input ("clean.output", "tag.text"), # by port label ] + history: HF Hub dataset repo ID used to persist generation history. + Pass a string like ``"username/my-workflow-history"`` to use an + existing or auto-created dataset repo. Pass ``True`` to + auto-name the repo as ``"{hf_user}/{workflow-slug}-history"``. + Defaults to ``None`` (no persistence). """ if graph is None: caller_filename = sys._getframe(1).f_code.co_filename @@ -1307,6 +1334,7 @@ def __init__( ) self._bound: dict[str, Callable] = bind or {} self._edges: list[tuple[str, str]] = edges or [] + self._history_param: str | bool | None = history if Context.root_block is not None: raise ValueError( @@ -1333,6 +1361,9 @@ def _build(self): self._workflow_file, ) + # Resolve the history repo ID (may require a Hub API call for history=True). + self._wh = self._resolve_history() + # Callable so each browser session re-reads `workflow.json`, picking up # writes from `save_workflow` instead of the construction-time snapshot. def _load_initial() -> str | None: @@ -1483,11 +1514,104 @@ def save_workflow( "Workflow: endpoint sync after save failed", exc_info=True, ) + # Sync workflow graph definition to Hub alongside generations. + if self._wh is not None: + import threading as _threading + + _threading.Thread( + target=self._wh.push_graph_file, args=(payload,), daemon=True + ).start() return "ok" except Exception as e: logger.error("save_workflow failed: %s", e, exc_info=True) return json.dumps({"error": str(e)}) + def list_history( + data=None, + _request: Optional[Request] = None, + _token: Optional[OAuthToken] = None, + ) -> str: + """Return recent generation records for the History panel.""" + if self._wh is None: + return json.dumps({"records": [], "repo_id": None}) + subgraph = data[0] if data else None + limit = int(data[1]) if data and len(data) > 1 and data[1] is not None else 50 + records = self._wh.list(limit=limit, subgraph=subgraph or None) + return json.dumps({"records": records, "repo_id": self._wh.repo_id}) + + def connect_history( + data=None, + request: Optional[Request] = None, + token: Optional[OAuthToken] = None, + ) -> str: + """Connect (or switch) the workflow to a HF Hub dataset for history. + + data[0]: repo_id string, e.g. ``"user/my-history"``. + data[1]: optional bool — if truthy, auto-derive repo_id from the + HF username and workflow name (same as ``history=True``). + """ + if not has_write_access(request, token): + return json.dumps({"error": "Write access required", "error_type": "auth"}) + try: + from gradio.workflow_history import WorkflowHistory + + auto = data[1] if data and len(data) > 1 else False + # Prefer the caller's OAuth bearer token over the local CLI token so + # this works on Spaces where hf_get_token() returns None. + hf_token = (token.token if token is not None else None) or hf_get_token() + if auto: + user = HfApi(token=hf_token).whoami()["name"] + slug = re.sub(r"[^a-z0-9-]", "-", self._workflow_name.lower()).strip("-") + repo_id = f"{user}/{slug}-history" + else: + repo_id = (data[0] if data else "").strip() + if not re.fullmatch(r"[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-][a-zA-Z0-9_./-]*", repo_id): + return json.dumps({"error": f"Invalid dataset repo ID: {repo_id}"}) + + wh = WorkflowHistory(repo_id=repo_id, token=hf_token) + wh.ensure_repo() + self._wh = wh + + # Persist the repo_id into workflow.json so it survives restarts. + # Hold _save_lock for the full read-modify-write to avoid racing + # with concurrent save_workflow() calls on the same file. + try: + with _save_lock: + raw = _load_initial() + if raw: + parsed = json.loads(raw) + parsed["history_repo"] = repo_id + with open(workflow_file, "w", encoding="utf-8") as f: + json.dump(parsed, f, ensure_ascii=False) + except Exception: + logger.debug("connect_history: could not persist to workflow.json", exc_info=True) + + return json.dumps({"ok": True, "repo_id": repo_id}) + except Exception as e: + logger.error("connect_history failed: %s", e, exc_info=True) + return json.dumps({"error": str(e)}) + + def push_history( + data=None, + _request: Optional[Request] = None, + _token: Optional[OAuthToken] = None, + ) -> str: + """Accept a pre-built record from a client-side run and push it to Hub. + + data[0]: JSON string of the history record dict. + """ + if self._wh is None: + return json.dumps({"ok": False, "reason": "no_history"}) + try: + record = json.loads(data[0]) if data else {} + if not record.get("id"): + return json.dumps({"ok": False, "reason": "invalid_record"}) + self._wh.push(record) + return json.dumps({"ok": True}) + except Exception as e: + logger.debug("push_history failed: %s", e, exc_info=True) + return json.dumps({"ok": False, "reason": str(e)}) + server_functions = [ get_token, get_write_access, @@ -1507,6 +1631,9 @@ def save_workflow( list_bound_fns, get_workflow_api, save_workflow, + list_history, + connect_history, + push_history, ] from gradio.workflow_api import WorkflowGraph, register_workflow_endpoints @@ -1580,7 +1707,49 @@ async def wrapper( # Expose each subject (output) as a named API endpoint reusing /info + # /call. The manager re-syncs on every save_workflow, so adding, # removing, renaming, or retyping an output updates the live API. - self._api_endpoints = register_workflow_endpoints(self, _current_graph, callers) + self._api_endpoints = register_workflow_endpoints( + self, _current_graph, callers, get_history=lambda: self._wh + ) + + def _resolve_history(self): + """Instantiate WorkflowHistory from ``self._history_param``, the saved + ``history_repo`` field in workflow.json, or return None.""" + from gradio.workflow_history import WorkflowHistory + + param = self._history_param + + # If no constructor arg, check whether workflow.json already has a + # ``history_repo`` written by a previous ``connect_history()`` call. + if param is None: + try: + with open(self._workflow_file, encoding="utf-8") as f: + saved = json.load(f) + repo_id = saved.get("history_repo") + if repo_id: + return WorkflowHistory(repo_id=repo_id, token=hf_get_token()) + except Exception: + pass + return None + + if param is True: + try: + token = hf_get_token() + user = HfApi(token=token).whoami()["name"] + slug = re.sub(r"[^a-z0-9-]", "-", self._workflow_name.lower()).strip( + "-" + ) + repo_id = f"{user}/{slug}-history" + except Exception: + logger.warning( + "Workflow: history=True requires a logged-in HF account " + "(run `huggingface-cli login`). History disabled.", + exc_info=True, + ) + return None + else: + repo_id = str(param) + token = hf_get_token() + return WorkflowHistory(repo_id=repo_id, token=token) def launch(self, *args, **kwargs): # type: ignore[override] """Launch the workflow as a Gradio app. Accepts the same arguments as `gr.Blocks.launch()`. diff --git a/gradio/workflow_api.py b/gradio/workflow_api.py index be59bf2f660..3a9d6c11cbd 100644 --- a/gradio/workflow_api.py +++ b/gradio/workflow_api.py @@ -20,9 +20,14 @@ import json import logging import re +import secrets +import threading from collections import deque from collections.abc import Callable -from typing import Any, Optional +from typing import TYPE_CHECKING, Any, Optional + +if TYPE_CHECKING: + from gradio.workflow_history import WorkflowHistory logger = logging.getLogger(__name__) @@ -534,7 +539,7 @@ def _run_model(self, node: dict, data_map: dict[str, dict[str, Any]]) -> None: resolved = self._resolve_inputs(node, data_map) self._require(node, resolved) args = [resolved[p["id"]] for p in node.get("inputs") or []] - tag = node.get("pipeline_tag") or "text-generation" + tag = node.get("pipeline_tag") or node.get("task") or "text-generation" provider = node.get("provider") or "auto" output_data = self._call( "model", [node.get("model_id"), tag, json.dumps(args), None, provider] @@ -693,6 +698,9 @@ def _build_endpoint_fn( subject_ids: list[str], free_ids: list[str], callers: dict[str, Callable], + api_name: str = "", + get_history: Optional[Callable[[], Optional["WorkflowHistory"]]] = None, + free_items: Optional[list[dict]] = None, ): """Build the callable backing one subgraph endpoint. `subject_ids` are all the outputs of the subgraph; with more than one the endpoint returns a tuple @@ -716,6 +724,15 @@ def endpoint(*args): results = WorkflowExecutor(graph, callers).run_many( subject_ids, inputs, request, token ) + history = get_history() if get_history else None + if history is not None: + try: + _record_generation( + history, api_name, graph, free_items or [], input_values, + subject_ids, results, request, token, + ) + except Exception: + logger.debug("_record_generation failed", exc_info=True) return results[0] if len(results) == 1 else tuple(results) params = [ @@ -746,6 +763,43 @@ def endpoint(*args): return endpoint +def _record_generation( + history: WorkflowHistory, + api_name: str, + graph: WorkflowGraph, + free_items: list[dict], + input_values: list[Any], + subject_ids: list[str], + results: list[Any], + request: Any, + token: Any, +) -> None: + """Fire-and-forget: push a generation record to WorkflowHistory.""" + from gradio.workflow_history import build_history_record + + user: str | None = None + try: + if token is not None and hasattr(token, "name"): + user = token.name + elif hasattr(request, "username"): + user = request.username + except Exception: + pass + + gen_id = secrets.token_hex(12) + record = build_history_record( + gen_id=gen_id, + subgraph=api_name, + graph=graph, + free_items=free_items, + input_values=list(input_values), + subject_ids=subject_ids, + results=list(results), + user=user, + ) + history.push(record) + + class WorkflowEndpointManager: """Owns the lifecycle of the per-subject API endpoints and keeps them in sync with the workflow graph. @@ -763,10 +817,12 @@ def __init__( blocks, get_graph: Callable[[], Optional[WorkflowGraph]], callers: dict[str, Callable], + get_history: Optional[Callable[[], Optional[WorkflowHistory]]] = None, ): self.blocks = blocks self.get_graph = get_graph self.callers = callers + self.get_history = get_history self._blocks_created: list = [] self._fn_ids: list[int] = [] self.api_names: list[str] = [] @@ -824,6 +880,9 @@ def _register(self, graph: WorkflowGraph) -> None: [s["id"] for s in group], [f["node"]["id"] for f in frees], self.callers, + api_name=api_name, + get_history=self.get_history, + free_items=frees, ) trigger = gr.Button(visible=False) self._blocks_created.append(trigger) @@ -862,10 +921,11 @@ def register_workflow_endpoints( blocks, get_graph: Callable[[], Optional[WorkflowGraph]], callers: dict[str, Callable], + get_history: Optional[Callable[[], Optional[WorkflowHistory]]] = None, ) -> WorkflowEndpointManager: """Create a `WorkflowEndpointManager` and register the initial endpoint set from the current graph. Returns the manager so the caller can `.sync()` it again whenever the graph is saved.""" - manager = WorkflowEndpointManager(blocks, get_graph, callers) + manager = WorkflowEndpointManager(blocks, get_graph, callers, get_history=get_history) manager.sync() return manager diff --git a/gradio/workflow_history.py b/gradio/workflow_history.py new file mode 100644 index 00000000000..d7be0afb122 --- /dev/null +++ b/gradio/workflow_history.py @@ -0,0 +1,308 @@ +"""WorkflowHistory — persists gr.Workflow generation records to a HF Hub dataset.""" + +from __future__ import annotations + +import json +import logging +import os +import pathlib +import secrets +import tempfile +import threading +import time +from datetime import datetime, timezone +from typing import Any + +from huggingface_hub import HfApi, hf_hub_download +from huggingface_hub import get_token as hf_get_token + +logger = logging.getLogger(__name__) + +MEDIA_PORT_TYPES = {"image", "audio", "video"} + +# Cache TTL for list() so rapid panel refreshes don't hammer Hub. +_LIST_CACHE_TTL = 10.0 + +# Max records to read from Hub when listing. +_MAX_FILES_SCAN = 50 + + +class WorkflowHistory: + """Persists workflow generation records to a private HF Hub dataset repo. + + Each Gradio process maintains one growing JSONL file on Hub under + ``data/{proc_id}.jsonl`` so concurrent processes don't conflict. Media + outputs (images, audio, video) are uploaded to ``media/`` and the local + path is replaced with a stable Hub URL before the record is stored. + + Args: + repo_id: HF Hub dataset repo identifier, e.g. ``"user/my-history"``. + token: HF access token. Falls back to the cached CLI token. + """ + + def __init__(self, repo_id: str, token: str | None = None) -> None: + self.repo_id = repo_id + self._token = token or hf_get_token() + self._api = HfApi(token=self._token) + + # Unique per-process ID so multiple Gradio instances write separate files + # without conflicts. + self._proc_id = secrets.token_hex(6) + self._data_path = f"data/{self._proc_id}.jsonl" + + # Local buffer file (appended in-process, then re-uploaded to Hub). + self._local_file = os.path.join( + tempfile.gettempdir(), + f"wf_hist_{repo_id.replace('/', '_')}_{self._proc_id}.jsonl", + ) + + self._write_lock = threading.Lock() + self._repo_ready = False + self._repo_lock = threading.Lock() + + # list() cache + self._cache: list[dict] | None = None + self._cache_at: float = 0.0 + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def push(self, record: dict) -> None: + """Persist *record* to Hub in a background thread (non-blocking).""" + threading.Thread(target=self._push_sync, args=(record,), daemon=True).start() + + def list( + self, limit: int = 50, subgraph: str | None = None + ) -> list[dict]: + """Return recent generation records, newest first. + + Results are cached for ``_LIST_CACHE_TTL`` seconds to avoid + hammering the Hub on rapid panel refreshes. + """ + now = time.monotonic() + if self._cache is not None and (now - self._cache_at) < _LIST_CACHE_TTL: + records = self._cache + else: + records = self._fetch_records() + self._cache = records + self._cache_at = now + + if subgraph: + records = [r for r in records if r.get("subgraph") == subgraph] + return records[:limit] + + def push_graph_file(self, workflow_json: str) -> None: + """Upload the current ``workflow.json`` to the dataset repo for versioning.""" + try: + self.ensure_repo() + if not self._repo_ready: + return + self._api.upload_file( + path_or_fileobj=workflow_json.encode("utf-8"), + path_in_repo="workflow.json", + repo_id=self.repo_id, + repo_type="dataset", + commit_message="Update workflow definition", + ) + except Exception: + logger.debug("WorkflowHistory: graph file upload failed", exc_info=True) + + # ------------------------------------------------------------------ + # Internals + # ------------------------------------------------------------------ + + def ensure_repo(self) -> None: + with self._repo_lock: + if self._repo_ready: + return + try: + self._api.create_repo( + self.repo_id, + repo_type="dataset", + private=True, + exist_ok=True, + ) + self._repo_ready = True + except Exception: + logger.warning( + "WorkflowHistory: could not create dataset repo %s", + self.repo_id, + exc_info=True, + ) + + def _push_sync(self, record: dict) -> None: + """Called in a daemon thread — uploads media, appends JSONL, uploads file.""" + try: + self.ensure_repo() + if not self._repo_ready: + return + + # Upload media outputs → replace local paths with stable Hub URLs. + for sid, output in list(record.get("outputs", {}).items()): + value = output.get("value") + port_type = output.get("type", "text") + if port_type in MEDIA_PORT_TYPES and isinstance(value, str): + hub_url = self._upload_media(record["id"], sid, value, port_type) + if hub_url: + record["outputs"][sid]["value"] = hub_url + elif os.path.isfile(value): + # Upload was attempted but failed — don't store a dead local path. + record["outputs"][sid]["value"] = None + + # Append to local buffer and re-upload to Hub. + with self._write_lock: + with open(self._local_file, "a", encoding="utf-8") as f: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + self._api.upload_file( + path_or_fileobj=self._local_file, + path_in_repo=self._data_path, + repo_id=self.repo_id, + repo_type="dataset", + commit_message="Add generation", + ) + # Invalidate inside the lock so no concurrent list() refills the + # cache with stale data after the upload but before this clear. + self._cache = None + except Exception: + logger.debug("WorkflowHistory: push failed", exc_info=True) + + def _upload_media( + self, gen_id: str, sid: str, value: str, port_type: str + ) -> str | None: + """Upload a local media file to the dataset repo's ``media/`` dir.""" + if not os.path.isfile(value): + # Already a URL or non-existent — keep as-is. + return None + ext = pathlib.Path(value).suffix or { + "image": ".png", + "audio": ".mp3", + "video": ".mp4", + }.get(port_type, ".bin") + path_in_repo = f"media/{gen_id}_{sid}{ext}" + try: + self._api.upload_file( + path_or_fileobj=value, + path_in_repo=path_in_repo, + repo_id=self.repo_id, + repo_type="dataset", + commit_message="Add generation media", + ) + return ( + f"https://huggingface.co/datasets/{self.repo_id}" + f"/resolve/main/{path_in_repo}" + ) + except Exception: + logger.debug("WorkflowHistory: media upload failed", exc_info=True) + return None + + def _fetch_records(self) -> list[dict]: + """Download JSONL files from Hub, merge, and sort newest-first.""" + try: + all_paths = sorted( + ( + f.rfilename + for f in self._api.list_repo_tree( + self.repo_id, + repo_type="dataset", + path_in_repo="data", + recursive=False, + ) + if f.rfilename.endswith(".jsonl") + ), + reverse=True, + )[:_MAX_FILES_SCAN] + + records: list[dict] = [] + for path in all_paths: + try: + local = hf_hub_download( + self.repo_id, + path, + repo_type="dataset", + token=self._token, + ) + with open(local, encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if line: + records.append(json.loads(line)) + except Exception: + continue + + records.sort(key=lambda r: r.get("timestamp", ""), reverse=True) + return records + except Exception: + logger.debug("WorkflowHistory: fetch failed", exc_info=True) + return [] + + +def build_history_record( + gen_id: str, + subgraph: str, + graph: Any, + free_items: list[dict], + input_values: list[Any], + subject_ids: list[str], + results: list[Any], + user: str | None, +) -> dict: + """Build a history record dict from a completed subgraph execution. + + Args: + gen_id: UUID string for this generation. + subgraph: API name of the subgraph endpoint. + graph: ``WorkflowGraph`` instance (for node metadata). + free_items: List of free-input dicts from ``group_free_inputs()``. + input_values: Positional input values passed to the executor. + subject_ids: Subject node IDs in the same order as *results*. + results: Output values from ``WorkflowExecutor.run_many()``. + user: HF username (or None for anonymous). + """ + ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + inputs: dict[str, dict] = {} + for item, value in zip(free_items, input_values): + node = item["node"] + node_id = node["id"] + port_type = item.get("type", "text") + label = item.get("label") or node.get("label", node_id) + port = item.get("port") or {} + port_id = port.get("id", "out_0") + # Serialise the value as a JSON-safe primitive. + safe_value: Any + if isinstance(value, (str, int, float, bool)) or value is None: + safe_value = value + else: + safe_value = str(value) + inputs[node_id] = { + "value": safe_value, + "type": port_type, + "label": label, + "port_id": port_id, + } + + outputs: dict[str, dict] = {} + for sid, result in zip(subject_ids, results): + node = graph.node_by_id.get(sid, {}) + in_ports = node.get("inputs") or [] + port_type = (in_ports[0].get("type") if in_ports else None) or node.get( + "asset_type", "text" + ) + label = node.get("label", sid) + safe_result: Any + if isinstance(result, (str, int, float, bool)) or result is None: + safe_result = result + else: + safe_result = str(result) + outputs[sid] = {"value": safe_result, "type": port_type, "label": label} + + return { + "id": gen_id, + "timestamp": ts, + "subgraph": subgraph, + "subject_ids": subject_ids, + "inputs": inputs, + "outputs": outputs, + "user": user, + } diff --git a/js/workflowcanvas/workflow/WorkflowCanvas.svelte b/js/workflowcanvas/workflow/WorkflowCanvas.svelte index 8654c6558c4..190f5ed36c7 100644 --- a/js/workflowcanvas/workflow/WorkflowCanvas.svelte +++ b/js/workflowcanvas/workflow/WorkflowCanvas.svelte @@ -8,6 +8,8 @@ import NodeModelPicker from "./NodeModelPicker.svelte"; import WorkflowEmptyState from "./WorkflowEmptyState.svelte"; import WorkflowApiPanel from "./WorkflowApiPanel.svelte"; + import WorkflowHistoryPanel from "./WorkflowHistoryPanel.svelte"; + import WorkflowHistoryConnect from "./WorkflowHistoryConnect.svelte"; import CheckIcon from "./icons/CheckIcon.svelte"; import LayoutIcon from "./icons/LayoutIcon.svelte"; import InfoIcon from "./icons/InfoIcon.svelte"; @@ -188,6 +190,23 @@ .catch(() => {}); }); + // Probe whether the backend has history configured. The history panel is + // only shown when repo_id is non-null in the response. + $effect(() => { + if (!server?.list_history) return; + void server + .list_history([null, 0]) + .then((raw: string) => { + try { + const parsed = JSON.parse(raw); + historyAvailable = parsed?.repo_id != null; + } catch { + /* ignore */ + } + }) + .catch(() => {}); + }); + $effect(() => { window.addEventListener("keydown", handleKeydown); window.addEventListener("keyup", handle_keyup); @@ -288,6 +307,9 @@ let showShortcuts = $state(false); let showUserMenu = $state(false); let showApiPanel = $state(false); + let showHistoryPanel = $state(false); + let showHistoryConnect = $state(false); + let historyAvailable = $state(false); // Popover shown when the "Run only" badge is clicked, explaining why editing // is disabled and how to enable it. let showAccessInfo = $state(false); @@ -1567,6 +1589,9 @@ } : undefined; + // Capture node outputs live for history recording. + const capturedOutputs: Record = {}; + await executeWorkflow( wfToRun, (nodeId, status, error, errorType) => { @@ -1610,6 +1635,8 @@ }, (nodeId, portId, value) => { updateNodeData(nodeId, portId, value); + if (!capturedOutputs[nodeId]) capturedOutputs[nodeId] = []; + capturedOutputs[nodeId].push({ portId, value }); }, abortController.signal, callSpaceWithToken, @@ -1630,9 +1657,53 @@ ); running = false; + const wasAborted = abortController?.signal.aborted ?? false; abortController = null; - const hasErrors = Object.values(nodeStatus).some((s) => s === "error"); + + // Push a history record for client-side runs (the server-side API path + // records history itself; the canvas executor doesn't go through it). + if (!wasAborted && !hasErrors && server?.push_history) { + try { + const genInputs: Record = {}; + for (const ref of wfToRun.references) { + const node = legacyView.nodes.find((n) => n.id === ref.id); + if (!node) continue; + const outPort = node.outputs[0]; + if (!outPort) continue; + const captured = capturedOutputs[ref.id]; + const value = captured + ? (captured.find((o) => o.portId === outPort.id) ?? captured[0])?.value ?? null + : node.data?.[outPort.id] ?? null; + genInputs[ref.id] = { value, type: outPort.type, label: node.label }; + } + const genOutputs: Record = {}; + for (const subj of wfToRun.subjects) { + const node = legacyView.nodes.find((n) => n.id === subj.id); + if (!node) continue; + const inPort = node.inputs[0]; + if (!inPort) continue; + const captured = capturedOutputs[subj.id]; + const value = captured + ? (captured.find((o) => o.portId === inPort.id) ?? captured[0])?.value ?? null + : node.data?.[inPort.id] ?? null; + genOutputs[subj.id] = { value, type: inPort.type, label: node.label }; + } + const record = { + id: crypto.randomUUID().replace(/-/g, "").slice(0, 24), + timestamp: new Date().toISOString().replace(/\.\d{3}Z$/, "Z"), + subgraph: wfToRun.name ?? "default", + subject_ids: wfToRun.subjects.map((s) => s.id), + inputs: genInputs, + outputs: genOutputs, + user: null + }; + server.push_history([JSON.stringify(record)]).catch(() => {}); + } catch { + // history push is best-effort + } + } + showToast( hasErrors ? "Workflow finished with errors" : "Workflow complete", hasErrors ? 5000 : 3000, @@ -2204,6 +2275,23 @@ View API + {#if historyAvailable} + + {:else if server?.connect_history} + + {/if} {#if saveIndicator} (showApiPanel = false)} /> {/if} + + {#if showHistoryConnect} + { + historyAvailable = true; + showHistoryConnect = false; + showHistoryPanel = true; + }} + onclose={() => (showHistoryConnect = false)} + /> + {/if} + + {#if showHistoryPanel} + (showHistoryPanel = false)} + onchange={() => { + showHistoryPanel = false; + showHistoryConnect = true; + }} + onload={(inputs) => { + // Load historical inputs back into the canvas reference nodes. + for (const [nodeId, input] of Object.entries(inputs as Record)) { + const portId: string = (input as any).port_id ?? "out_0"; + const value = (input as any).value; + updateNodeData(nodeId, portId, value); + } + showHistoryPanel = false; + }} + /> + {/if} diff --git a/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte b/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte new file mode 100644 index 00000000000..eb99d275c4e --- /dev/null +++ b/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte @@ -0,0 +1,523 @@ + + + + + diff --git a/js/workflowcanvas/workflow/workflow-executor.ts b/js/workflowcanvas/workflow/workflow-executor.ts index 84f99332504..0e08311c5eb 100644 --- a/js/workflowcanvas/workflow/workflow-executor.ts +++ b/js/workflowcanvas/workflow/workflow-executor.ts @@ -419,7 +419,7 @@ export async function executeWorkflow( // Prefer browser-side streaming for chat-completion-compatible // text tasks so the UI receives tokens as they arrive. The // Python path stays for every other task. - const tag = node.pipeline_tag ?? "text-generation"; + const tag = node.pipeline_tag ?? (node as any).task ?? "text-generation"; const streamable = (tag === "text-generation" || tag === "text2text-generation" || diff --git a/js/workflowcanvas/workflow/workflow-migration.ts b/js/workflowcanvas/workflow/workflow-migration.ts index 2dcbd285cd3..a27f1dcdc97 100644 --- a/js/workflowcanvas/workflow/workflow-migration.ts +++ b/js/workflowcanvas/workflow/workflow-migration.ts @@ -171,8 +171,8 @@ export function toLegacyShape(wf: Workflow): { kind: "component", label: n.label, source: "local", - inputs: n.inputs, - outputs: n.outputs, + inputs: n.inputs ?? [], + outputs: n.outputs ?? [], x: n.x, y: n.y, width: n.width, @@ -196,8 +196,8 @@ export function toLegacyShape(wf: Workflow): { pipeline_tag: n.pipeline_tag, provider: n.provider, fn: n.fn, - inputs: n.inputs, - outputs: n.outputs, + inputs: n.inputs ?? [], + outputs: n.outputs ?? [], x: n.x, y: n.y, width: n.width, @@ -211,8 +211,8 @@ export function toLegacyShape(wf: Workflow): { kind: "component", label: n.label, source: "local", - inputs: n.inputs, - outputs: n.outputs, + inputs: n.inputs ?? [], + outputs: n.outputs ?? [], x: n.x, y: n.y, width: n.width, diff --git a/test/test_workflow_history.py b/test/test_workflow_history.py new file mode 100644 index 00000000000..2947027fd46 --- /dev/null +++ b/test/test_workflow_history.py @@ -0,0 +1,323 @@ +"""Unit tests for gradio.workflow_history.WorkflowHistory and build_history_record.""" + +from __future__ import annotations + +import json +import os +import tempfile +from unittest.mock import MagicMock, patch + +import pytest + +from gradio.workflow_history import WorkflowHistory, build_history_record + + +# ─── build_history_record ───────────────────────────────────────────────────── + + +class FakeGraph: + def __init__(self, nodes: dict): + self.node_by_id = nodes + + +def test_build_history_record_text_inputs(): + graph = FakeGraph( + { + "subj_0": { + "id": "subj_0", + "label": "Result", + "inputs": [{"id": "in_0", "type": "text"}], + } + } + ) + free_items = [ + { + "node": {"id": "ref_0", "label": "Prompt"}, + "port": {"id": "out_0", "type": "text"}, + "type": "text", + "label": "Prompt", + } + ] + record = build_history_record( + gen_id="abc123", + subgraph="generate", + graph=graph, + free_items=free_items, + input_values=["hello world"], + subject_ids=["subj_0"], + results=["world hello"], + user="alice", + ) + assert record["id"] == "abc123" + assert record["subgraph"] == "generate" + assert record["user"] == "alice" + assert record["inputs"]["ref_0"]["value"] == "hello world" + assert record["inputs"]["ref_0"]["port_id"] == "out_0" + assert record["inputs"]["ref_0"]["type"] == "text" + assert record["outputs"]["subj_0"]["value"] == "world hello" + assert record["outputs"]["subj_0"]["type"] == "text" + assert "timestamp" in record + + +def test_build_history_record_non_serializable_value(): + graph = FakeGraph({"subj_0": {"id": "subj_0", "label": "Out", "inputs": []}}) + free_items = [ + { + "node": {"id": "ref_0", "label": "Input"}, + "port": {"id": "out_0", "type": "text"}, + "type": "text", + "label": "Input", + } + ] + # Pass a non-serializable object — should be coerced to str + record = build_history_record( + gen_id="x", + subgraph="sg", + graph=graph, + free_items=free_items, + input_values=[object()], + subject_ids=["subj_0"], + results=[None], + user=None, + ) + assert isinstance(record["inputs"]["ref_0"]["value"], str) + assert record["outputs"]["subj_0"]["value"] is None + assert record["user"] is None + + +def test_build_history_record_missing_port_uses_default(): + graph = FakeGraph({"subj_0": {"id": "subj_0", "label": "Out", "inputs": []}}) + free_items = [ + { + "node": {"id": "ref_0", "label": "Input"}, + "port": None, # No port + "type": "text", + "label": "Input", + } + ] + record = build_history_record( + gen_id="x", + subgraph="sg", + graph=graph, + free_items=free_items, + input_values=["val"], + subject_ids=["subj_0"], + results=["out"], + user=None, + ) + assert record["inputs"]["ref_0"]["port_id"] == "out_0" + + +# ─── WorkflowHistory ────────────────────────────────────────────────────────── + + +def _make_history(tmp_path, repo_id="user/test-history"): + with ( + patch("gradio.workflow_history.HfApi") as MockApi, + patch("gradio.workflow_history.hf_get_token", return_value="tok"), + ): + mock_api = MockApi.return_value + wh = WorkflowHistory(repo_id=repo_id, token="tok") + wh._api = mock_api + wh._repo_ready = True + # Point local file to tmp dir + wh._local_file = str(tmp_path / "buf.jsonl") + return wh, mock_api + + +def test_push_appends_to_local_file(tmp_path): + wh, mock_api = _make_history(tmp_path) + record = { + "id": "gen1", + "timestamp": "2026-06-30T12:00:00Z", + "subgraph": "generate", + "subject_ids": ["subj_0"], + "inputs": {"ref_0": {"value": "cat", "type": "text", "label": "Prompt", "port_id": "out_0"}}, + "outputs": {"subj_0": {"value": "a cute cat", "type": "text", "label": "Result"}}, + "user": None, + } + wh._push_sync(record) + + assert os.path.exists(wh._local_file) + with open(wh._local_file) as f: + lines = [json.loads(l) for l in f if l.strip()] + assert len(lines) == 1 + assert lines[0]["id"] == "gen1" + # upload_file should have been called (once for JSONL, zero times for media since value is text) + mock_api.upload_file.assert_called_once() + + +def test_push_uploads_image_to_media(tmp_path): + wh, mock_api = _make_history(tmp_path) + # Create a fake local image file + img_path = str(tmp_path / "out.png") + with open(img_path, "wb") as f: + f.write(b"\x89PNG\r\n") + + mock_api.upload_file.return_value = None + record = { + "id": "gen2", + "timestamp": "2026-06-30T13:00:00Z", + "subgraph": "img", + "subject_ids": ["subj_img"], + "inputs": {}, + "outputs": { + "subj_img": {"value": img_path, "type": "image", "label": "Image"} + }, + "user": None, + } + wh._push_sync(record) + + # Two upload_file calls: media upload + JSONL upload + assert mock_api.upload_file.call_count == 2 + media_call = mock_api.upload_file.call_args_list[0] + assert "media/" in media_call.kwargs.get("path_in_repo", "") + # Record in local file should have the Hub URL, not the local path + with open(wh._local_file) as f: + saved = json.loads(f.read().strip()) + assert "huggingface.co" in saved["outputs"]["subj_img"]["value"] + + +def test_push_skips_media_upload_for_urls(tmp_path): + wh, mock_api = _make_history(tmp_path) + record = { + "id": "gen3", + "timestamp": "2026-06-30T14:00:00Z", + "subgraph": "img", + "subject_ids": ["subj_img"], + "inputs": {}, + "outputs": { + "subj_img": { + "value": "https://example.com/img.png", + "type": "image", + "label": "Image", + } + }, + "user": None, + } + wh._push_sync(record) + + # Only JSONL upload (no media upload since value is already a URL) + assert mock_api.upload_file.call_count == 1 + with open(wh._local_file) as f: + saved = json.loads(f.read().strip()) + # URL preserved as-is + assert saved["outputs"]["subj_img"]["value"] == "https://example.com/img.png" + + +def test_list_returns_cached_results(tmp_path): + wh, mock_api = _make_history(tmp_path) + wh._cache = [{"id": "cached", "timestamp": "2026-06-30T00:00:00Z"}] + wh._cache_at = float("inf") # never expires + + results = wh.list(limit=10) + # _fetch_records should NOT be called since cache is warm + mock_api.list_repo_tree.assert_not_called() + assert results[0]["id"] == "cached" + + +def test_list_filters_by_subgraph(tmp_path): + wh, mock_api = _make_history(tmp_path) + wh._cache = [ + {"id": "a", "timestamp": "2026-06-30T01:00:00Z", "subgraph": "foo"}, + {"id": "b", "timestamp": "2026-06-30T02:00:00Z", "subgraph": "bar"}, + {"id": "c", "timestamp": "2026-06-30T03:00:00Z", "subgraph": "foo"}, + ] + wh._cache_at = float("inf") + + filtered = wh.list(subgraph="foo") + assert len(filtered) == 2 + assert all(r["subgraph"] == "foo" for r in filtered) + + +def test_list_respects_limit(tmp_path): + wh, mock_api = _make_history(tmp_path) + wh._cache = [ + {"id": str(i), "timestamp": f"2026-06-30T{i:02d}:00:00Z", "subgraph": "sg"} + for i in range(20) + ] + wh._cache_at = float("inf") + assert len(wh.list(limit=5)) == 5 + + +def test_push_graph_file(tmp_path): + wh, mock_api = _make_history(tmp_path) + wh.push_graph_file('{"schema_version": "2"}') + mock_api.upload_file.assert_called_once() + call_kwargs = mock_api.upload_file.call_args.kwargs + assert call_kwargs["path_in_repo"] == "workflow.json" + + +def test_ensure_repo_creates_dataset(tmp_path): + with ( + patch("gradio.workflow_history.HfApi") as MockApi, + patch("gradio.workflow_history.hf_get_token", return_value="tok"), + ): + mock_api = MockApi.return_value + wh = WorkflowHistory("user/new-repo", token="tok") + wh._api = mock_api + wh.ensure_repo() + mock_api.create_repo.assert_called_once_with( + "user/new-repo", + repo_type="dataset", + private=True, + exist_ok=True, + ) + assert wh._repo_ready is True + + +def test_history_true_auto_names_repo(): + """history=True should derive the repo ID from the HF username and workflow name.""" + import warnings + + with ( + patch("gradio.workflow.HfApi") as MockHfApi, + patch("gradio.workflow.hf_get_token", return_value="tok"), + warnings.catch_warnings(), + ): + warnings.simplefilter("ignore") + mock_api_inst = MockHfApi.return_value + mock_api_inst.whoami.return_value = {"name": "testuser"} + + from gradio.workflow import Workflow + + wf = Workflow.__new__(Workflow) + wf._workflow_name = "My Workflow" + wf._history_param = True + result = wf._resolve_history() + + assert result is not None + assert result.repo_id == "testuser/my-workflow-history" + + +def test_history_string_uses_explicit_repo(): + import warnings + + with ( + patch("gradio.workflow.hf_get_token", return_value="tok"), + warnings.catch_warnings(), + ): + warnings.simplefilter("ignore") + from gradio.workflow import Workflow + + wf = Workflow.__new__(Workflow) + wf._workflow_name = "Workflow" + wf._history_param = "acme/custom-history" + result = wf._resolve_history() + + assert result is not None + assert result.repo_id == "acme/custom-history" + + +def test_history_none_returns_none(): + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + from gradio.workflow import Workflow + + wf = Workflow.__new__(Workflow) + wf._history_param = None + result = wf._resolve_history() + + assert result is None From cb6b706c05dc5560a3536b03375a2809d9af9f7f Mon Sep 17 00:00:00 2001 From: gradio-pr-bot Date: Mon, 20 Jul 2026 14:59:26 +0000 Subject: [PATCH 02/13] add changeset --- .changeset/late-poets-see.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/late-poets-see.md diff --git a/.changeset/late-poets-see.md b/.changeset/late-poets-see.md new file mode 100644 index 00000000000..0a65e7544d1 --- /dev/null +++ b/.changeset/late-poets-see.md @@ -0,0 +1,6 @@ +--- +"@gradio/workflowcanvas": minor +"gradio": minor +--- + +feat:add workflow generation history persistence From 86974c939828f72136fc945710e1be5f79e47394 Mon Sep 17 00:00:00 2001 From: hannahblair Date: Wed, 22 Jul 2026 15:21:08 +0100 Subject: [PATCH 03/13] add bucket --- .../workflow/WorkflowHistoryConnect.svelte | 68 +++++++++++++++++-- .../workflow/WorkflowHistoryPanel.svelte | 2 +- 2 files changed, 64 insertions(+), 6 deletions(-) diff --git a/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte b/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte index bd4158ae56f..a0175910931 100644 --- a/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte +++ b/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte @@ -14,6 +14,7 @@ let repoInput = $state(""); let connecting = $state(false); let error = $state(null); + let repoType = $state<"dataset" | "bucket">("dataset"); const suggestedName = $derived( workflowName @@ -28,7 +29,7 @@ error = null; connecting = true; try { - const raw = await server.connect_history([repoInput.trim(), auto]); + const raw = await server.connect_history([repoInput.trim(), auto, repoType]); const data = typeof raw === "string" ? JSON.parse(raw) : raw; if (data.error) { error = data.error; @@ -58,10 +59,29 @@

- Save every generation to a private HF Hub dataset. Browse and reload past + Save every generation to a private HF Hub repo. Browse and reload past outputs from the History panel.

+
+ + +
+
+

@@ -101,7 +107,9 @@ { @@ -212,7 +220,9 @@ font-weight: 500; padding: 5px 0; cursor: pointer; - transition: background 0.1s, color 0.1s; + transition: + background 0.1s, + color 0.1s; } .type-btn.active { diff --git a/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte b/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte index 57e4442c2d6..e88499bdf76 100644 --- a/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte +++ b/js/workflowcanvas/workflow/WorkflowHistoryPanel.svelte @@ -118,7 +118,12 @@ rel="noopener noreferrer" title="View dataset on Hugging Face" > - + @@ -126,14 +131,22 @@ {repoId} {#if onchange} -

{record.subgraph}
-
{formatRelativeTime(record.timestamp)}
+
+ {formatRelativeTime(record.timestamp)} +
{#if record.user}
{record.user}
{/if} diff --git a/js/workflowcanvas/workflow/workflow-executor.ts b/js/workflowcanvas/workflow/workflow-executor.ts index 0e08311c5eb..a8c8ec9da29 100644 --- a/js/workflowcanvas/workflow/workflow-executor.ts +++ b/js/workflowcanvas/workflow/workflow-executor.ts @@ -419,7 +419,8 @@ export async function executeWorkflow( // Prefer browser-side streaming for chat-completion-compatible // text tasks so the UI receives tokens as they arrive. The // Python path stays for every other task. - const tag = node.pipeline_tag ?? (node as any).task ?? "text-generation"; + const tag = + node.pipeline_tag ?? (node as any).task ?? "text-generation"; const streamable = (tag === "text-generation" || tag === "text2text-generation" || diff --git a/test/test_workflow_history.py b/test/test_workflow_history.py index 2947027fd46..da767eb543a 100644 --- a/test/test_workflow_history.py +++ b/test/test_workflow_history.py @@ -4,14 +4,10 @@ import json import os -import tempfile -from unittest.mock import MagicMock, patch - -import pytest +from unittest.mock import patch from gradio.workflow_history import WorkflowHistory, build_history_record - # ─── build_history_record ───────────────────────────────────────────────────── @@ -132,8 +128,17 @@ def test_push_appends_to_local_file(tmp_path): "timestamp": "2026-06-30T12:00:00Z", "subgraph": "generate", "subject_ids": ["subj_0"], - "inputs": {"ref_0": {"value": "cat", "type": "text", "label": "Prompt", "port_id": "out_0"}}, - "outputs": {"subj_0": {"value": "a cute cat", "type": "text", "label": "Result"}}, + "inputs": { + "ref_0": { + "value": "cat", + "type": "text", + "label": "Prompt", + "port_id": "out_0", + } + }, + "outputs": { + "subj_0": {"value": "a cute cat", "type": "text", "label": "Result"} + }, "user": None, } wh._push_sync(record) @@ -161,9 +166,7 @@ def test_push_uploads_image_to_media(tmp_path): "subgraph": "img", "subject_ids": ["subj_img"], "inputs": {}, - "outputs": { - "subj_img": {"value": img_path, "type": "image", "label": "Image"} - }, + "outputs": {"subj_img": {"value": img_path, "type": "image", "label": "Image"}}, "user": None, } wh._push_sync(record) From 0e468d60b6f8278fca55f9116dad777293c0389a Mon Sep 17 00:00:00 2001 From: hannahblair Date: Wed, 22 Jul 2026 17:00:48 +0100 Subject: [PATCH 07/13] clean up --- gradio/workflow.py | 39 ++++++------------- gradio/workflow_history.py | 36 ++++------------- .../workflow/WorkflowCanvas.svelte | 7 ---- test/test_workflow_history.py | 14 +++---- 4 files changed, 26 insertions(+), 70 deletions(-) diff --git a/gradio/workflow.py b/gradio/workflow.py index 290a4121ef6..ac6d2b4ec12 100644 --- a/gradio/workflow.py +++ b/gradio/workflow.py @@ -72,8 +72,6 @@ class _CuratedCache(TypedDict): _CURATED_CACHE: _CuratedCache = {"fetched_at": 0.0, "items": None} _CURATED_LOCK = threading.Lock() -# pipeline_tag lookup cache: model_id → tag string, avoids re-hitting Hub on -# every execution when the node was added without a stored pipeline_tag. _MODEL_TAG_CACHE: dict[str, str] = {} _MODEL_TAG_LOCK = threading.Lock() @@ -380,7 +378,18 @@ def _save_tmp(result, ext: str) -> dict: def _img_url(a) -> str: - return a.get("url") or a.get("path", "") if isinstance(a, dict) else a + if not isinstance(a, dict): + return a + path = a.get("path", "") + url = a.get("url", "") + # Prefer the local file path: InferenceClient reads it directly. + # Only fall back to url if it's a public HTTPS address; never send + # Gradio-internal relative URLs (e.g. /gradio_api/file=...) to the Hub. + if path and os.path.isfile(path): + return path + if url.startswith("https://"): + return url + return path or url def _classify_error(e: Exception) -> dict: @@ -778,10 +787,6 @@ def process_item(item): } -# Legacy pipeline tags (as sent by older saved workflows and the browser -# executor) → InferenceClient endpoint names. depth-estimation is absent on -# purpose: InferenceClient has no such method, so it keeps its raw-POST branch -# in call_model. Unmapped tags fall through to the chat/raw-POST fallback. _PIPELINE_TAG_TO_ENDPOINT: dict[str, str] = { "text-generation": "text_generation", "text2text-generation": "text_generation", @@ -813,8 +818,6 @@ def process_item(item): } -# Client params that expect a list of strings; port values arrive as a single -# string, split on the given pattern. _ENDPOINT_LIST_KWARGS: dict[str, dict[str, str]] = { "zero_shot_classification": {"candidate_labels": r"[\n,]"}, "sentence_similarity": {"other_sentences": r"\n"}, @@ -826,8 +829,6 @@ def get_model_endpoints( ) -> str: from huggingface_hub import InferenceClient - # Only advertise endpoints the installed huggingface_hub can actually run, - # so the UI never shapes a node around a method that would fail server-side. endpoints = [ {"name": name, **schema} for name, schema in _INFERENCE_ENDPOINT_SCHEMAS.items() @@ -856,7 +857,6 @@ def _dispatch_model_endpoint(client, endpoint: str, kwargs: dict) -> str: continue legacy = re.fullmatch(r"in_(\d+)", k) if legacy and k not in schema_ids and int(legacy.group(1)) < len(schema_ids): - # positional port IDs from workflows saved before endpoint schemas k = schema_ids[int(legacy.group(1))] clean[k] = ( _img_url(v) if isinstance(v, dict) and ("url" in v or "path" in v) else v @@ -865,7 +865,6 @@ def _dispatch_model_endpoint(client, endpoint: str, kwargs: dict) -> str: if isinstance(clean.get(key), str): clean[key] = [s.strip() for s in re.split(sep, clean[key]) if s.strip()] if endpoint == "zero_shot_classification" and not clean.get("candidate_labels"): - # without labels, fall back to scoring with the model's own label set return _dispatch_model_endpoint( client, "text_classification", {"text": clean.get("text", "")} ) @@ -889,7 +888,6 @@ def _dispatch_model_endpoint(client, endpoint: str, kwargs: dict) -> str: if ext: return json.dumps([_save_tmp(result, ext)]) if isinstance(result, list) and result and hasattr(result[0], "answer"): - # question-answering-style outputs: surface the top answer result = result[0] for attr in ( "summary_text", @@ -943,8 +941,6 @@ def call_model( client = InferenceClient(model=model_id, token=hf_token, provider=provider) args = json.loads(args_json) - # Autodetect the pipeline_tag from the Hub when the node doesn't carry - # one, so we never mis-route a text-to-image model as text-generation. if not pipeline_tag: with _MODEL_TAG_LOCK: pipeline_tag = _MODEL_TAG_CACHE.get(model_id) @@ -984,8 +980,6 @@ def call_model( endpoint = _PIPELINE_TAG_TO_ENDPOINT.get(task) if endpoint: - # Positional args from legacy saved workflows and the browser - # executor map onto the endpoint schema's input order. schema_inputs = _INFERENCE_ENDPOINT_SCHEMAS[endpoint]["inputs"] kwargs = { schema_inputs[i]["id"]: val @@ -1577,7 +1571,6 @@ def _build(self): self._workflow_file, ) - # Resolve the history repo ID (may require a Hub API call for history=True). self._wh = self._resolve_history() # Callable so each browser session re-reads `workflow.json`, picking up @@ -1730,7 +1723,6 @@ def save_workflow( "Workflow: endpoint sync after save failed", exc_info=True, ) - # Sync workflow graph definition to Hub alongside generations. if self._wh is not None: import threading as _threading @@ -1777,8 +1769,6 @@ def connect_history( auto = data[1] if data and len(data) > 1 else False repo_type = (data[2] if data and len(data) > 2 else None) or "dataset" - # Prefer the caller's OAuth bearer token over the local CLI token so - # this works on Spaces where hf_get_token() returns None. hf_token = ( token.token if token is not None else None ) or hf_get_token() @@ -1801,9 +1791,6 @@ def connect_history( wh.ensure_repo() self._wh = wh - # Persist the repo_id and repo_type into workflow.json so they - # survive restarts. Hold _save_lock for the full read-modify-write - # to avoid racing with concurrent save_workflow() calls. try: with _save_lock: raw = _load_initial() @@ -1955,8 +1942,6 @@ def _resolve_history(self): param = self._history_param repo_type = self._history_repo_type - # If no constructor arg, check whether workflow.json already has a - # ``history_repo`` written by a previous ``connect_history()`` call. if param is None: try: with open(self._workflow_file, encoding="utf-8") as f: diff --git a/gradio/workflow_history.py b/gradio/workflow_history.py index 918cf05d990..f95c3c0d44d 100644 --- a/gradio/workflow_history.py +++ b/gradio/workflow_history.py @@ -20,10 +20,8 @@ MEDIA_PORT_TYPES = {"image", "audio", "video"} -# Cache TTL for list() so rapid panel refreshes don't hammer Hub. _LIST_CACHE_TTL = 10.0 -# Max files to scan when listing records. _MAX_FILES_SCAN = 200 @@ -60,17 +58,12 @@ def __init__( self._token = token or hf_get_token() self._api = HfApi(token=self._token) - # Unique per-process ID so multiple Gradio instances write separate files - # without conflicts. self._proc_id = secrets.token_hex(6) if repo_type == "bucket": - # Buckets: one JSON file per record — no growing buffer, no O(N²) uploads. self._data_path = None self._local_file = None else: - # Datasets: one growing JSONL file per process appended locally then - # re-uploaded to Hub. self._data_path = f"data/{self._proc_id}.jsonl" self._local_file = os.path.join( tempfile.gettempdir(), @@ -81,14 +74,9 @@ def __init__( self._repo_ready = False self._repo_lock = threading.Lock() - # list() cache self._cache: list[dict] | None = None self._cache_at: float = 0.0 - # ------------------------------------------------------------------ - # Public API - # ------------------------------------------------------------------ - def push(self, record: dict) -> None: """Persist *record* to Hub in a background thread (non-blocking).""" threading.Thread(target=self._push_sync, args=(record,), daemon=True).start() @@ -133,10 +121,6 @@ def push_graph_file(self, workflow_json: str) -> None: except Exception: logger.debug("WorkflowHistory: graph file upload failed", exc_info=True) - # ------------------------------------------------------------------ - # Internals - # ------------------------------------------------------------------ - def ensure_repo(self) -> None: with self._repo_lock: if self._repo_ready: @@ -167,7 +151,6 @@ def _push_sync(self, record: dict) -> None: if not self._repo_ready: return - # Upload media outputs → replace local paths with stable Hub URLs. for sid, output in list(record.get("outputs", {}).items()): value = output.get("value") port_type = output.get("type", "text") @@ -176,7 +159,6 @@ def _push_sync(self, record: dict) -> None: if hub_url: record["outputs"][sid]["value"] = hub_url elif os.path.isfile(value): - # Upload was attempted but failed — don't store a dead local path. record["outputs"][sid]["value"] = None if self.repo_type == "bucket": @@ -193,10 +175,11 @@ def _push_to_bucket(self, record: dict) -> None: Bucket writes are independent files so no local-state lock is needed; concurrent pushes proceed in parallel. """ - # Replace colons in the ISO timestamp so the path is safe on all - # platforms and S3-compatible backends (colons are illegal on Windows). - safe_ts = record["timestamp"].replace(":", "-") - path_in_repo = f"data/{safe_ts}_{record['id']}.json" + ts = record.get("timestamp") or datetime.now(timezone.utc).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + safe_ts = ts.replace(":", "-") + path_in_repo = f"data/{safe_ts}_{record.get('id', secrets.token_hex(8))}.json" data = json.dumps(record, ensure_ascii=False).encode("utf-8") self._api.batch_bucket_files( bucket_id=self.repo_id, @@ -216,8 +199,6 @@ def _push_to_dataset(self, record: dict) -> None: repo_type="dataset", commit_message="Add generation", ) - # Invalidate inside the lock so no concurrent list() refills the - # cache with stale data after the upload but before this clear. self._cache = None def _upload_media( @@ -225,7 +206,6 @@ def _upload_media( ) -> str | None: """Upload a local media file to the repo/bucket's ``media/`` dir.""" if not os.path.isfile(value): - # Already a URL or non-existent — keep as-is. return None ext = pathlib.Path(value).suffix or { "image": ".png", @@ -240,7 +220,6 @@ def _upload_media( bucket_id=self.repo_id, add=[(fh.read(), path_in_repo)], ) - # Bucket files are served at /buckets/{id}/{path} (no /resolve/main/). return f"https://huggingface.co/buckets/{self.repo_id}/{path_in_repo}" else: self._api.upload_file( @@ -270,13 +249,12 @@ def _fetch_records(self) -> list[dict]: def _fetch_from_bucket(self) -> list[dict]: """Fetch individual JSON record files from a bucket.""" - from huggingface_hub._buckets import BucketFile - all_items = sorted( ( item for item in self._api.list_bucket_tree(self.repo_id, prefix="data/") - if isinstance(item, BucketFile) and item.path.endswith(".json") + if getattr(item, "path", "").endswith(".json") + and not hasattr(item, "count") ), key=lambda f: f.path, reverse=True, diff --git a/js/workflowcanvas/workflow/WorkflowCanvas.svelte b/js/workflowcanvas/workflow/WorkflowCanvas.svelte index 4e55ff1bba1..cf7e991b579 100644 --- a/js/workflowcanvas/workflow/WorkflowCanvas.svelte +++ b/js/workflowcanvas/workflow/WorkflowCanvas.svelte @@ -193,8 +193,6 @@ .catch(() => {}); }); - // Probe whether the backend has history configured. The history panel is - // only shown when repo_id is non-null in the response. $effect(() => { if (!server?.list_history) return; void server @@ -1605,7 +1603,6 @@ } : undefined; - // Capture node outputs live for history recording. const capturedOutputs: Record = {}; @@ -1678,8 +1675,6 @@ abortController = null; const hasErrors = Object.values(nodeStatus).some((s) => s === "error"); - // Push a history record for client-side runs (the server-side API path - // records history itself; the canvas executor doesn't go through it). if (!wasAborted && !hasErrors && server?.push_history) { try { const genInputs: Record< @@ -1725,7 +1720,6 @@ }; server.push_history([JSON.stringify(record)]).catch(() => {}); } catch { - // history push is best-effort } } @@ -2823,7 +2817,6 @@ showHistoryConnect = true; }} onload={(inputs) => { - // Load historical inputs back into the canvas reference nodes. for (const [nodeId, input] of Object.entries( inputs as Record )) { diff --git a/test/test_workflow_history.py b/test/test_workflow_history.py index da767eb543a..7d3a32f6980 100644 --- a/test/test_workflow_history.py +++ b/test/test_workflow_history.py @@ -109,10 +109,10 @@ def test_build_history_record_missing_port_uses_default(): def _make_history(tmp_path, repo_id="user/test-history"): with ( - patch("gradio.workflow_history.HfApi") as MockApi, + patch("gradio.workflow_history.HfApi") as mock_api_cls, patch("gradio.workflow_history.hf_get_token", return_value="tok"), ): - mock_api = MockApi.return_value + mock_api = mock_api_cls.return_value wh = WorkflowHistory(repo_id=repo_id, token="tok") wh._api = mock_api wh._repo_ready = True @@ -145,7 +145,7 @@ def test_push_appends_to_local_file(tmp_path): assert os.path.exists(wh._local_file) with open(wh._local_file) as f: - lines = [json.loads(l) for l in f if l.strip()] + lines = [json.loads(line) for line in f if line.strip()] assert len(lines) == 1 assert lines[0]["id"] == "gen1" # upload_file should have been called (once for JSONL, zero times for media since value is text) @@ -253,10 +253,10 @@ def test_push_graph_file(tmp_path): def test_ensure_repo_creates_dataset(tmp_path): with ( - patch("gradio.workflow_history.HfApi") as MockApi, + patch("gradio.workflow_history.HfApi") as mock_api_cls, patch("gradio.workflow_history.hf_get_token", return_value="tok"), ): - mock_api = MockApi.return_value + mock_api = mock_api_cls.return_value wh = WorkflowHistory("user/new-repo", token="tok") wh._api = mock_api wh.ensure_repo() @@ -274,12 +274,12 @@ def test_history_true_auto_names_repo(): import warnings with ( - patch("gradio.workflow.HfApi") as MockHfApi, + patch("gradio.workflow.HfApi") as mock_hf_api_cls, patch("gradio.workflow.hf_get_token", return_value="tok"), warnings.catch_warnings(), ): warnings.simplefilter("ignore") - mock_api_inst = MockHfApi.return_value + mock_api_inst = mock_hf_api_cls.return_value mock_api_inst.whoami.return_value = {"name": "testuser"} from gradio.workflow import Workflow From d7970222e9a4782f77d4ceb9123b33e0a0555ffe Mon Sep 17 00:00:00 2001 From: hannahblair Date: Wed, 22 Jul 2026 17:09:05 +0100 Subject: [PATCH 08/13] format --- gradio/routes.py | 6 +++++- js/workflowcanvas/workflow/WorkflowCanvas.svelte | 3 +-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/gradio/routes.py b/gradio/routes.py index 095b800be84..4d4aafa5414 100644 --- a/gradio/routes.py +++ b/gradio/routes.py @@ -1405,7 +1405,11 @@ async def queue_join_helper( ) error_map = { "queue_full": status.HTTP_503_SERVICE_UNAVAILABLE, - "validator_error": getattr(status, "HTTP_422_UNPROCESSABLE_CONTENT", status.HTTP_422_UNPROCESSABLE_ENTITY), + "validator_error": getattr( + status, + "HTTP_422_UNPROCESSABLE_CONTENT", + status.HTTP_422_UNPROCESSABLE_ENTITY, + ), "error": status.HTTP_400_BAD_REQUEST, "success": status.HTTP_200_OK, } diff --git a/js/workflowcanvas/workflow/WorkflowCanvas.svelte b/js/workflowcanvas/workflow/WorkflowCanvas.svelte index cf7e991b579..6e9a2f31a24 100644 --- a/js/workflowcanvas/workflow/WorkflowCanvas.svelte +++ b/js/workflowcanvas/workflow/WorkflowCanvas.svelte @@ -1719,8 +1719,7 @@ user: null }; server.push_history([JSON.stringify(record)]).catch(() => {}); - } catch { - } + } catch {} } showToast( From 0c4f121d541536cc83bbca39ff824c4ce5225abb Mon Sep 17 00:00:00 2001 From: hannahblair Date: Thu, 23 Jul 2026 15:59:05 +0100 Subject: [PATCH 09/13] use only buckets --- gradio/workflow.py | 39 +-- gradio/workflow_history.py | 272 +++++------------- .../workflow/WorkflowHistoryConnect.svelte | 70 +---- 3 files changed, 87 insertions(+), 294 deletions(-) diff --git a/gradio/workflow.py b/gradio/workflow.py index ac6d2b4ec12..9cb0d0db838 100644 --- a/gradio/workflow.py +++ b/gradio/workflow.py @@ -1494,7 +1494,6 @@ def __init__( bind: dict[str, Callable] | list[Callable] | None = None, edges: list[tuple[str, str]] | None = None, history: str | bool | None = None, - history_repo_type: str = "dataset", ): """ Parameters: @@ -1517,14 +1516,11 @@ def __init__( ("shout", "reverse"), # first output → first input ("clean.output", "tag.text"), # by port label ] - history: HF Hub repo/bucket ID used to persist generation history. + history: HF Hub bucket ID used to persist generation history. Pass a string like ``"username/my-workflow-history"`` to use an - existing or auto-created repo. Pass ``True`` to auto-name the - repo as ``"{hf_user}/{workflow-slug}-history"``. + existing or auto-created bucket. Pass ``True`` to auto-name the + bucket as ``"{hf_user}/{workflow-slug}-history"``. Defaults to ``None`` (no persistence). - history_repo_type: ``"dataset"`` (default, free accounts) or - ``"bucket"`` (Hub S3-like storage, no git commit noise, - requires a paid plan). """ if graph is None: caller_filename = sys._getframe(1).f_code.co_filename @@ -1544,7 +1540,6 @@ def __init__( self._bound: dict[str, Callable] = bind or {} self._edges: list[tuple[str, str]] = edges or [] self._history_param: str | bool | None = history - self._history_repo_type: str = history_repo_type if Context.root_block is not None: raise ValueError( @@ -1754,10 +1749,10 @@ def connect_history( request: Optional[Request] = None, token: Optional[OAuthToken] = None, ) -> str: - """Connect (or switch) the workflow to a HF Hub dataset for history. + """Connect (or switch) the workflow to a HF Hub bucket for history. - data[0]: repo_id string, e.g. ``"user/my-history"``. - data[1]: optional bool — if truthy, auto-derive repo_id from the + data[0]: bucket_id string, e.g. ``"user/my-history"``. + data[1]: optional bool — if truthy, auto-derive bucket_id from the HF username and workflow name (same as ``history=True``). """ if not has_write_access(request, token): @@ -1768,7 +1763,6 @@ def connect_history( from gradio.workflow_history import WorkflowHistory auto = data[1] if data and len(data) > 1 else False - repo_type = (data[2] if data and len(data) > 2 else None) or "dataset" hf_token = ( token.token if token is not None else None ) or hf_get_token() @@ -1783,11 +1777,9 @@ def connect_history( if not re.fullmatch( r"[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-][a-zA-Z0-9_./-]*", repo_id ): - return json.dumps({"error": f"Invalid repo ID: {repo_id}"}) + return json.dumps({"error": f"Invalid bucket ID: {repo_id}"}) - wh = WorkflowHistory( - repo_id=repo_id, token=hf_token, repo_type=repo_type - ) + wh = WorkflowHistory(repo_id=repo_id, token=hf_token) wh.ensure_repo() self._wh = wh @@ -1797,7 +1789,6 @@ def connect_history( if raw: parsed = json.loads(raw) parsed["history_repo"] = repo_id - parsed["history_repo_type"] = repo_type with open(workflow_file, "w", encoding="utf-8") as f: json.dump(parsed, f, ensure_ascii=False) except Exception: @@ -1806,9 +1797,7 @@ def connect_history( exc_info=True, ) - return json.dumps( - {"ok": True, "repo_id": repo_id, "repo_type": repo_type} - ) + return json.dumps({"ok": True, "repo_id": repo_id}) except Exception as e: logger.error("connect_history failed: %s", e, exc_info=True) return json.dumps({"error": str(e)}) @@ -1940,7 +1929,6 @@ def _resolve_history(self): from gradio.workflow_history import WorkflowHistory param = self._history_param - repo_type = self._history_repo_type if param is None: try: @@ -1948,12 +1936,7 @@ def _resolve_history(self): saved = json.load(f) repo_id = saved.get("history_repo") if repo_id: - saved_type = saved.get("history_repo_type", "dataset") - return WorkflowHistory( - repo_id=repo_id, - token=hf_get_token(), - repo_type=saved_type, - ) + return WorkflowHistory(repo_id=repo_id, token=hf_get_token()) except Exception: pass return None @@ -1976,7 +1959,7 @@ def _resolve_history(self): else: repo_id = str(param) token = hf_get_token() - return WorkflowHistory(repo_id=repo_id, token=token, repo_type=repo_type) + return WorkflowHistory(repo_id=repo_id, token=token) def launch(self, *args, **kwargs): # type: ignore[override] """Launch the workflow as a Gradio app. Accepts the same arguments as `gr.Blocks.launch()`. diff --git a/gradio/workflow_history.py b/gradio/workflow_history.py index f95c3c0d44d..ac5d77eddc2 100644 --- a/gradio/workflow_history.py +++ b/gradio/workflow_history.py @@ -1,4 +1,4 @@ -"""WorkflowHistory — persists gr.Workflow generation records to a HF Hub repo or bucket.""" +"""WorkflowHistory — persists gr.Workflow generation records to a HF Hub bucket.""" from __future__ import annotations @@ -11,9 +11,9 @@ import threading import time from datetime import datetime, timezone -from typing import Any, Literal +from typing import Any -from huggingface_hub import HfApi, hf_hub_download +from huggingface_hub import HfApi from huggingface_hub import get_token as hf_get_token logger = logging.getLogger(__name__) @@ -26,54 +26,24 @@ class WorkflowHistory: - """Persists workflow generation records to a HF Hub dataset repo or bucket. + """Persists workflow generation records to a HF Hub bucket. - Two storage backends are supported: - - * ``repo_type="dataset"`` (default) — git-based dataset repo. Each process - keeps one growing JSONL file so concurrent processes don't conflict. - Works on free HF accounts. - - * ``repo_type="bucket"`` — S3-like Hub bucket. Each generation is stored as - an individual JSON file (``data/_.json``), so there is - no re-upload growth and no git commit noise. Requires a paid HF plan. - - Media outputs (images, audio, video) are uploaded to ``media/`` and their - local paths are replaced with stable Hub URLs before the record is stored. + Each generation is stored as an individual JSON file + (``data/_.json``). Media outputs (images, audio, + video) are uploaded to ``media/`` and their local paths are replaced + with stable Hub URLs before the record is stored. Args: - repo_id: HF Hub repo/bucket identifier, e.g. ``"user/my-history"``. + repo_id: HF Hub bucket identifier, e.g. ``"user/my-history"``. token: HF access token. Falls back to the cached CLI token. - repo_type: ``"dataset"`` or ``"bucket"``. """ - def __init__( - self, - repo_id: str, - token: str | None = None, - repo_type: Literal["dataset", "bucket"] = "dataset", - ) -> None: + def __init__(self, repo_id: str, token: str | None = None) -> None: self.repo_id = repo_id - self.repo_type = repo_type self._token = token or hf_get_token() self._api = HfApi(token=self._token) - - self._proc_id = secrets.token_hex(6) - - if repo_type == "bucket": - self._data_path = None - self._local_file = None - else: - self._data_path = f"data/{self._proc_id}.jsonl" - self._local_file = os.path.join( - tempfile.gettempdir(), - f"wf_hist_{repo_id.replace('/', '_')}_{self._proc_id}.jsonl", - ) - - self._write_lock = threading.Lock() self._repo_ready = False self._repo_lock = threading.Lock() - self._cache: list[dict] | None = None self._cache_at: float = 0.0 @@ -100,24 +70,15 @@ def list(self, limit: int = 50, subgraph: str | None = None) -> list[dict]: return records[:limit] def push_graph_file(self, workflow_json: str) -> None: - """Upload the current ``workflow.json`` to the repo/bucket for versioning.""" + """Upload the current ``workflow.json`` to the bucket for versioning.""" try: self.ensure_repo() if not self._repo_ready: return - if self.repo_type == "bucket": - self._api.batch_bucket_files( - bucket_id=self.repo_id, - add=[(workflow_json.encode("utf-8"), "workflow.json")], - ) - else: - self._api.upload_file( - path_or_fileobj=workflow_json.encode("utf-8"), - path_in_repo="workflow.json", - repo_id=self.repo_id, - repo_type="dataset", - commit_message="Update workflow definition", - ) + self._api.batch_bucket_files( + bucket_id=self.repo_id, + add=[(workflow_json.encode("utf-8"), "workflow.json")], + ) except Exception: logger.debug("WorkflowHistory: graph file upload failed", exc_info=True) @@ -126,20 +87,11 @@ def ensure_repo(self) -> None: if self._repo_ready: return try: - if self.repo_type == "bucket": - self._api.create_bucket(self.repo_id, private=True, exist_ok=True) - else: - self._api.create_repo( - self.repo_id, - repo_type="dataset", - private=True, - exist_ok=True, - ) + self._api.create_bucket(self.repo_id, private=True, exist_ok=True) self._repo_ready = True except Exception: logger.warning( - "WorkflowHistory: could not create %s %s", - self.repo_type, + "WorkflowHistory: could not create bucket %s", self.repo_id, exc_info=True, ) @@ -161,50 +113,25 @@ def _push_sync(self, record: dict) -> None: elif os.path.isfile(value): record["outputs"][sid]["value"] = None - if self.repo_type == "bucket": - self._push_to_bucket(record) - else: - self._push_to_dataset(record) + ts = record.get("timestamp") or datetime.now(timezone.utc).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + safe_ts = ts.replace(":", "-") + path_in_repo = f"data/{safe_ts}_{record.get('id', secrets.token_hex(8))}.json" + data = json.dumps(record, ensure_ascii=False).encode("utf-8") + self._api.batch_bucket_files( + bucket_id=self.repo_id, + add=[(data, path_in_repo)], + ) + self._cache = None except Exception: logger.debug("WorkflowHistory: push failed", exc_info=True) - def _push_to_bucket(self, record: dict) -> None: - """Write one JSON file per record — no growing buffer, no commit noise. - - Bucket writes are independent files so no local-state lock is needed; - concurrent pushes proceed in parallel. - """ - ts = record.get("timestamp") or datetime.now(timezone.utc).strftime( - "%Y-%m-%dT%H:%M:%SZ" - ) - safe_ts = ts.replace(":", "-") - path_in_repo = f"data/{safe_ts}_{record.get('id', secrets.token_hex(8))}.json" - data = json.dumps(record, ensure_ascii=False).encode("utf-8") - self._api.batch_bucket_files( - bucket_id=self.repo_id, - add=[(data, path_in_repo)], - ) - self._cache = None - - def _push_to_dataset(self, record: dict) -> None: - """Append to local JSONL buffer and re-upload to the dataset repo.""" - with self._write_lock: - with open(self._local_file, "a", encoding="utf-8") as f: - f.write(json.dumps(record, ensure_ascii=False) + "\n") - self._api.upload_file( - path_or_fileobj=self._local_file, - path_in_repo=self._data_path, - repo_id=self.repo_id, - repo_type="dataset", - commit_message="Add generation", - ) - self._cache = None - def _upload_media( self, gen_id: str, sid: str, value: str, port_type: str ) -> str | None: - """Upload a local media file to the repo/bucket's ``media/`` dir.""" + """Upload a local media file to the bucket's ``media/`` dir.""" if not os.path.isfile(value): return None ext = pathlib.Path(value).suffix or { @@ -214,116 +141,63 @@ def _upload_media( }.get(port_type, ".bin") path_in_repo = f"media/{gen_id}_{sid}{ext}" try: - if self.repo_type == "bucket": - with open(value, "rb") as fh: - self._api.batch_bucket_files( - bucket_id=self.repo_id, - add=[(fh.read(), path_in_repo)], - ) - return f"https://huggingface.co/buckets/{self.repo_id}/{path_in_repo}" - else: - self._api.upload_file( - path_or_fileobj=value, - path_in_repo=path_in_repo, - repo_id=self.repo_id, - repo_type="dataset", - commit_message="Add generation media", - ) - return ( - f"https://huggingface.co/datasets/{self.repo_id}" - f"/resolve/main/{path_in_repo}" + with open(value, "rb") as fh: + self._api.batch_bucket_files( + bucket_id=self.repo_id, + add=[(fh.read(), path_in_repo)], ) + return f"https://huggingface.co/buckets/{self.repo_id}/{path_in_repo}" except Exception: logger.debug("WorkflowHistory: media upload failed", exc_info=True) return None def _fetch_records(self) -> list[dict]: - """Download record files from Hub and sort newest-first.""" + """Download record files from the bucket and sort newest-first.""" try: - if self.repo_type == "bucket": - return self._fetch_from_bucket() - return self._fetch_from_dataset() - except Exception: - logger.debug("WorkflowHistory: fetch failed", exc_info=True) - return [] - - def _fetch_from_bucket(self) -> list[dict]: - """Fetch individual JSON record files from a bucket.""" - all_items = sorted( - ( - item - for item in self._api.list_bucket_tree(self.repo_id, prefix="data/") - if getattr(item, "path", "").endswith(".json") - and not hasattr(item, "count") - ), - key=lambda f: f.path, - reverse=True, - )[:_MAX_FILES_SCAN] - - if not all_items: - return [] - - records: list[dict] = [] - with tempfile.TemporaryDirectory() as tmpdir: - downloads = [ - (item, os.path.join(tmpdir, f"{i}.json")) - for i, item in enumerate(all_items) - ] - try: - self._api.download_bucket_files( - bucket_id=self.repo_id, - files=[(item, local) for item, local in downloads], - token=self._token, - ) - except Exception: - logger.debug("WorkflowHistory: bucket download failed", exc_info=True) + all_items = sorted( + ( + item + for item in self._api.list_bucket_tree(self.repo_id, prefix="data/") + if getattr(item, "path", "").endswith(".json") + and not hasattr(item, "count") + ), + key=lambda f: f.path, + reverse=True, + )[:_MAX_FILES_SCAN] + + if not all_items: return [] - for _, local in downloads: + records: list[dict] = [] + with tempfile.TemporaryDirectory() as tmpdir: + downloads = [ + (item, os.path.join(tmpdir, f"{i}.json")) + for i, item in enumerate(all_items) + ] try: - with open(local, encoding="utf-8") as fh: - records.append(json.load(fh)) + self._api.download_bucket_files( + bucket_id=self.repo_id, + files=[(item, local) for item, local in downloads], + token=self._token, + ) except Exception: - continue + logger.debug( + "WorkflowHistory: bucket download failed", exc_info=True + ) + return [] - records.sort(key=lambda r: r.get("timestamp", ""), reverse=True) - return records + for _, local in downloads: + try: + with open(local, encoding="utf-8") as fh: + records.append(json.load(fh)) + except Exception: + continue - def _fetch_from_dataset(self) -> list[dict]: - """Download JSONL files from a dataset repo, merge, and sort newest-first.""" - all_paths = sorted( - ( - f.rfilename - for f in self._api.list_repo_tree( - self.repo_id, - repo_type="dataset", - path_in_repo="data", - recursive=False, - ) - if f.rfilename.endswith(".jsonl") - ), - reverse=True, - )[:_MAX_FILES_SCAN] - - records: list[dict] = [] - for path in all_paths: - try: - local = hf_hub_download( - self.repo_id, - path, - repo_type="dataset", - token=self._token, - ) - with open(local, encoding="utf-8") as fh: - for line in fh: - line = line.strip() - if line: - records.append(json.loads(line)) - except Exception: - continue - - records.sort(key=lambda r: r.get("timestamp", ""), reverse=True) - return records + records.sort(key=lambda r: r.get("timestamp", ""), reverse=True) + return records + except Exception: + logger.debug("WorkflowHistory: fetch failed", exc_info=True) + return [] def build_history_record( diff --git a/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte b/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte index 1dc3d955250..e938b313c8f 100644 --- a/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte +++ b/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte @@ -14,7 +14,6 @@ let repoInput = $state(""); let connecting = $state(false); let error = $state(null); - let repoType = $state<"dataset" | "bucket">("dataset"); const suggestedName = $derived( workflowName @@ -29,11 +28,7 @@ error = null; connecting = true; try { - const raw = await server.connect_history([ - repoInput.trim(), - auto, - repoType - ]); + const raw = await server.connect_history([repoInput.trim(), auto]); const data = typeof raw === "string" ? JSON.parse(raw) : raw; if (data.error) { error = data.error; @@ -65,29 +60,10 @@

- Save every generation to a private HF Hub repo. Browse and reload past + Save every generation to a private HF Hub bucket. Browse and reload past outputs from the History panel.

-
- - -
-
{:else if server?.connect_history} {/if} {#if saveIndicator} @@ -2798,7 +2830,7 @@ { + onconnected={() => { historyAvailable = true; showHistoryConnect = false; showHistoryPanel = true; @@ -2810,6 +2842,7 @@ {#if showHistoryPanel} (showHistoryPanel = false)} onchange={() => { showHistoryPanel = false; diff --git a/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte b/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte index e938b313c8f..17420932373 100644 --- a/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte +++ b/js/workflowcanvas/workflow/WorkflowHistoryConnect.svelte @@ -1,4 +1,6 @@