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 diff --git a/gradio/history.py b/gradio/history.py new file mode 100644 index 00000000000..5a5db02e0c8 --- /dev/null +++ b/gradio/history.py @@ -0,0 +1,290 @@ +"""BucketHistory — core Gradio primitive for persisting per-app history to a HF Hub bucket. + +Used by ``gr.Workflow`` for generation history, but usable by any Gradio app that +wants an append-only record store backed by a private HF Hub bucket. +""" + +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 +from huggingface_hub import get_token as hf_get_token + +logger = logging.getLogger(__name__) + +MEDIA_PORT_TYPES = {"image", "audio", "video"} + +_LIST_CACHE_TTL = 10.0 + +_MAX_FILES_SCAN = 50 + + +class BucketHistory: + """Persists app records to a private HF Hub bucket. + + Each record is stored as an individual JSON file under + ``data/_.json``. Media outputs (images, audio, video) + are uploaded to ``media/`` and a ``bucket_url`` is added to the record + for durable reference; the original ``value`` (usually a Gradio-served + URL) is preserved for in-session display. + + Args: + repo_id: HF Hub bucket 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) + self._repo_ready = False + self._repo_lock = threading.Lock() + self._cache_lock = threading.Lock() + self._cache: list[dict] | None = None + self._cache_at: float = 0.0 + + 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 records, newest first. + + Results are cached for ``_LIST_CACHE_TTL`` seconds to avoid + hammering the Hub on rapid panel refreshes. + """ + now = time.monotonic() + with self._cache_lock: + if self._cache is None or (now - self._cache_at) >= _LIST_CACHE_TTL: + self._cache = self._fetch_records() + self._cache_at = time.monotonic() + records = self._cache + + if subgraph: + records = [r for r in records if r.get("subgraph") == subgraph] + return records[:limit] + + def delete(self, record_id: str, timestamp: str) -> bool: + """Delete a single record from the bucket. + + Returns True on success, False if the record could not be deleted. + """ + try: + safe_ts = timestamp.replace(":", "-") + path_in_repo = f"data/{safe_ts}_{record_id}.json" + self._api.batch_bucket_files( + bucket_id=self.repo_id, + delete=[path_in_repo], + ) + with self._cache_lock: + self._cache = None + return True + except Exception: + logger.debug("BucketHistory: delete failed", exc_info=True) + return False + + def ensure_repo(self) -> None: + with self._repo_lock: + if self._repo_ready: + return + try: + self._api.create_bucket(self.repo_id, private=True, exist_ok=True) + self._repo_ready = True + except Exception: + logger.warning( + "BucketHistory: could not create bucket %s", + self.repo_id, + exc_info=True, + ) + + def _push_sync(self, record: dict) -> None: + """Called in a daemon thread — uploads media then writes the record.""" + try: + self.ensure_repo() + if not self._repo_ready: + return + + 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): + fs_path = ( + value[len("/gradio_api/file=") :] + if value.startswith("/gradio_api/file=") + else value + ) + hub_url = self._upload_media(record["id"], sid, fs_path, port_type) + if hub_url: + record["outputs"][sid]["bucket_url"] = hub_url + + 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)], + ) + with self._cache_lock: + self._cache = None + + except Exception: + logger.debug("BucketHistory: 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 bucket's ``media/`` dir.""" + if not os.path.isfile(value): + return None + if not os.path.realpath(value).startswith( + os.path.realpath(tempfile.gettempdir()) + ): + logger.debug("BucketHistory: refusing to upload non-temp path: %s", value) + 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: + 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("BucketHistory: media upload failed", exc_info=True) + return None + + def _fetch_records(self) -> list[dict]: + """Download record files from the bucket and sort newest-first.""" + try: + 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("BucketHistory: bucket download failed", exc_info=True) + return [] + + for _, local in downloads: + try: + with open(local, encoding="utf-8") as fh: + records.append(json.load(fh)) + except Exception: + continue + + records.sort(key=lambda r: r.get("timestamp", ""), reverse=True) + return records + except Exception: + logger.debug("BucketHistory: 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") + 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/gradio/http_server.py b/gradio/http_server.py index 84924f13738..d3f84cda8ff 100644 --- a/gradio/http_server.py +++ b/gradio/http_server.py @@ -48,7 +48,9 @@ def __init__( self, config: Config, reloader: _ServerReloaderT | None = None, - watchfn: Callable[[_ServerReloaderT], None] = watchfn, # ty: ignore[invalid-parameter-default] + watchfn: Callable[ + [_ServerReloaderT], None + ] = watchfn, # ty: ignore[invalid-parameter-default] ) -> None: self.running_app = config.app super().__init__(config) diff --git a/gradio/routes.py b/gradio/routes.py index 8af6be360f7..e2d47244c24 100644 --- a/gradio/routes.py +++ b/gradio/routes.py @@ -66,6 +66,7 @@ import gradio from gradio import ( caching, + oauth, route_utils, themes, utils, @@ -695,6 +696,101 @@ def main( "the frontend by running /scripts/build_frontend.sh" ) from err + _bucket_repo_re = re.compile( + r"^[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-][a-zA-Z0-9_./-]*$" + ) + + def _oauth_token(request: fastapi.Request) -> str | None: + try: + info = oauth._get_valid_oauth_info_from_session(request.session) + except Exception: + return None + return info.get("access_token") if info else None + + async def _history_body(request: fastapi.Request) -> dict: + try: + data = await request.json() + return data if isinstance(data, dict) else {} + except Exception: + return {} + + def _bucket_for(request: fastapi.Request, bucket_id: str): + """Return a cached ``BucketHistory`` for the caller's OAuth token + and the requested bucket, or None if unauthenticated / invalid.""" + hf_token = _oauth_token(request) + if not hf_token or not _bucket_repo_re.fullmatch(bucket_id or ""): + return None + from gradio.history import BucketHistory + + cache = app.state.bucket_history_cache + key = (hf_token, bucket_id) + if key not in cache: + cache[key] = BucketHistory(bucket_id, token=hf_token) + return cache[key] + + app.state.bucket_history_cache = {} + + @app.post("/gradio_api/history/list") + async def _history_list(request: fastapi.Request): + body = await _history_body(request) + wh = _bucket_for(request, str(body.get("bucket_id") or "")) + if wh is None: + return JSONResponse({"records": []}) + try: + limit = int(body.get("limit", 50) or 0) + except Exception: + limit = 50 + if limit == 0: + return JSONResponse({"records": []}) + records = wh.list(limit=limit, subgraph=body.get("subgraph") or None) + return JSONResponse({"records": records}) + + @app.post("/gradio_api/history/push") + async def _history_push(request: fastapi.Request): + body = await _history_body(request) + wh = _bucket_for(request, str(body.get("bucket_id") or "")) + if wh is None: + return JSONResponse({"ok": False, "reason": "auth"}, status_code=403) + record = body.get("record") or {} + if isinstance(record, str): + try: + record = orjson.loads(record) + except Exception: + return JSONResponse({"ok": False, "reason": "invalid_record"}) + if not isinstance(record, dict) or not record.get("id"): + return JSONResponse({"ok": False, "reason": "invalid_record"}) + wh.push(record) + return JSONResponse({"ok": True}) + + @app.post("/gradio_api/history/delete") + async def _history_delete(request: fastapi.Request): + body = await _history_body(request) + wh = _bucket_for(request, str(body.get("bucket_id") or "")) + if wh is None: + return JSONResponse({"ok": False, "reason": "auth"}, status_code=403) + record_id = str(body.get("id") or "") + timestamp = str(body.get("timestamp") or "") + if not record_id or not timestamp: + return JSONResponse({"ok": False, "reason": "missing_fields"}) + return JSONResponse({"ok": wh.delete(record_id, timestamp)}) + + @app.get("/gradio_api/history/buckets") + async def _history_buckets(request: fastapi.Request): + """List the authenticated user's own buckets (for a picker UI).""" + hf_token = _oauth_token(request) + if not hf_token: + return JSONResponse({"buckets": []}) + try: + from huggingface_hub import HfApi as _HfApi + + buckets = [ + {"id": b.id, "private": getattr(b, "private", True)} + for b in _HfApi(token=hf_token).list_buckets(token=hf_token) + ] + return JSONResponse({"buckets": buckets}) + except Exception: + return JSONResponse({"buckets": []}) + @app.get("/gradio_api/deep_link") def deep_link(session_hash: str): if session_hash in app.state_holder: @@ -1405,7 +1501,9 @@ 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", 422 + ), "error": status.HTTP_400_BAD_REQUEST, "success": status.HTTP_200_OK, } diff --git a/gradio/workflow.py b/gradio/workflow.py index 9df0af98272..9e2fd1a658d 100644 --- a/gradio/workflow.py +++ b/gradio/workflow.py @@ -72,6 +72,10 @@ class _CuratedCache(TypedDict): _CURATED_CACHE: _CuratedCache = {"fetched_at": 0.0, "items": None} _CURATED_LOCK = threading.Lock() +_MODEL_TAG_CACHE: dict[str, tuple[str, float]] = {} +_MODEL_TAG_LOCK = threading.Lock() +_MODEL_TAG_TTL = 3600.0 + def _bundled_snapshot_path() -> str: return os.path.join(os.path.dirname(__file__), "_workflow_curated_snapshot.json") @@ -375,7 +379,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: @@ -773,10 +788,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", @@ -808,8 +819,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"}, @@ -821,8 +830,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() @@ -851,7 +858,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 @@ -860,7 +866,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", "")} ) @@ -884,7 +889,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", @@ -937,6 +941,28 @@ 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) + + if not pipeline_tag: + import time as _time + + now = _time.monotonic() + with _MODEL_TAG_LOCK: + cached = _MODEL_TAG_CACHE.get(model_id) + if cached and (now - cached[1]) < _MODEL_TAG_TTL: + pipeline_tag = cached[0] + 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, + _time.monotonic(), + ) + except Exception: + pass + if isinstance(args, dict): endpoint = pipeline_tag or "" return _dispatch_model_endpoint(client, endpoint, args) @@ -963,8 +989,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 @@ -1819,7 +1843,9 @@ 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 + ) def launch(self, *args, **kwargs): # type: ignore[override] """Launch the workflow as a Gradio app. Accepts the same arguments as `gr.Blocks.launch()`. @@ -1835,6 +1861,7 @@ def launch(self, *args, **kwargs): # type: ignore[override] kwargs.update(dict(zip(names, args))) kwargs["allowed_paths"] = [ tempfile.gettempdir(), + os.path.realpath(tempfile.gettempdir()), *(kwargs.get("allowed_paths") or []), ] # We need the edit link to print (and the browser to open to it) before diff --git a/gradio/workflow_api.py b/gradio/workflow_api.py index 15ac52499bb..e2470c6c234 100644 --- a/gradio/workflow_api.py +++ b/gradio/workflow_api.py @@ -550,7 +550,7 @@ def _run_model(self, node: dict, data_map: dict[str, dict[str, Any]]) -> None: ] else: 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" call_data = [node.get("model_id"), tag, json.dumps(args), None, provider] output_data = self._call("model", call_data) self._map_outputs(node, output_data, data_map) diff --git a/js/workflowcanvas/workflow/WorkflowCanvas.svelte b/js/workflowcanvas/workflow/WorkflowCanvas.svelte index e8c8a0c2369..0fe8556fd9b 100644 --- a/js/workflowcanvas/workflow/WorkflowCanvas.svelte +++ b/js/workflowcanvas/workflow/WorkflowCanvas.svelte @@ -1674,9 +1674,10 @@ ); running = false; + const wasAborted = abortController?.signal.aborted ?? false; abortController = null; - const hasErrors = Object.values(nodeStatus).some((s) => s === "error"); + showToast( hasErrors ? "Workflow finished with errors" : "Workflow complete", hasErrors ? 5000 : 3000, @@ -2747,6 +2748,7 @@ onClose={() => (showApiPanel = false)} /> {/if} +