diff --git a/studio/backend/core/data_recipe/jobs/constants.py b/studio/backend/core/data_recipe/jobs/constants.py new file mode 100644 index 0000000000..5d4081abbc --- /dev/null +++ b/studio/backend/core/data_recipe/jobs/constants.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +# stages parsed from data-designer logs +STAGE_CREATE = "create" +STAGE_PREVIEW = "preview" +STAGE_DAG = "dag" +STAGE_HEALTHCHECK = "healthcheck" +STAGE_SAMPLING = "sampling" +STAGE_COLUMN_CONFIG = "column_config" +STAGE_GENERATING = "generating" +STAGE_BATCH = "batch" +STAGE_PROFILING = "profiling" + +USAGE_RESET_STAGES = { + STAGE_CREATE, + STAGE_PREVIEW, + STAGE_DAG, + STAGE_HEALTHCHECK, + STAGE_SAMPLING, + STAGE_GENERATING, + STAGE_PROFILING, +} + +# job event types emitted by worker/manager +EVENT_JOB_ENQUEUED = "job.enqueued" +EVENT_JOB_STARTED = "job.started" +EVENT_JOB_CANCELLING = "job.cancelling" +EVENT_JOB_CANCELLED = "job.cancelled" +EVENT_JOB_COMPLETED = "job.completed" +EVENT_JOB_ERROR = "job.error" diff --git a/studio/backend/core/data_recipe/jobs/manager.py b/studio/backend/core/data_recipe/jobs/manager.py index eb8c10bb81..dbaa620004 100644 --- a/studio/backend/core/data_recipe/jobs/manager.py +++ b/studio/backend/core/data_recipe/jobs/manager.py @@ -13,6 +13,15 @@ from typing import Any import multiprocessing as mp +from ..jsonable import to_jsonable +from .constants import ( + EVENT_JOB_CANCELLING, + EVENT_JOB_CANCELLED, + EVENT_JOB_COMPLETED, + EVENT_JOB_ENQUEUED, + EVENT_JOB_ERROR, + EVENT_JOB_STARTED, +) from .parse import apply_update, coerce_event, parse_log_message from .types import Job from .worker import run_job_process @@ -20,29 +29,6 @@ from .worker import run_job_process _CTX = mp.get_context("spawn") -def _to_jsonable(value: Any) -> Any: - try: - import numpy as np # type: ignore - except Exception: # pragma: no cover - np = None # type: ignore - - if np is not None: - if isinstance(value, np.ndarray): - return value.tolist() - if isinstance(value, np.generic): - return value.item() - - if isinstance(value, dict): - return {str(k): _to_jsonable(v) for k, v in value.items()} - if isinstance(value, (list, tuple, set)): - return [_to_jsonable(v) for v in value] - if hasattr(value, "isoformat") and callable(value.isoformat): - try: - return value.isoformat() - except Exception: - pass - return value - @dataclass class Subscription: @@ -123,7 +109,7 @@ class JobManager: self._pump_thread = threading.Thread(target=self._pump_loop, daemon=True) self._pump_thread.start() - self._emit({"type": "job.enqueued", "ts": time.time(), "job_id": job_id}) + self._emit({"type": EVENT_JOB_ENQUEUED, "ts": time.time(), "job_id": job_id}) return job_id def cancel(self, job_id: str) -> bool: @@ -134,15 +120,15 @@ class JobManager: if self._proc is None or not self._proc.is_alive(): return True self._job.status = "cancelling" - self._emit({"type": "job.cancelling", "ts": time.time(), "job_id": job_id}) + self._emit({"type": EVENT_JOB_CANCELLING, "ts": time.time(), "job_id": job_id}) try: self._proc.terminate() - except Exception: + except (AttributeError, OSError): pass return True def get_status(self, job_id: str) -> dict | None: - """UI-friendly snapshot. Poll this if you don't want SSE.""" + """UI friendly snapshot that we need. Alternative to sse kinda of and structured""" with self._lock: if self._job is None or self._job.job_id != job_id: return None @@ -304,7 +290,7 @@ class JobManager: ).fetchdf() finally: conn.close() - except Exception: + except (RuntimeError, ValueError, duckdb.Error): return None for helper_col in ("filename", "__row_num__"): @@ -312,7 +298,7 @@ class JobManager: dataframe = dataframe.drop(columns=[helper_col]) rows = dataframe.to_dict(orient="records") - return {"dataset": _to_jsonable(rows), "total": total} + return {"dataset": to_jsonable(rows), "total": total} @staticmethod def _load_dataset_page_with_data_designer( @@ -326,7 +312,7 @@ class JobManager: dataframe = read_parquet_dataset(parquet_dir) total = int(len(dataframe.index)) rows = dataframe.iloc[offset:offset + limit].to_dict(orient="records") - return {"dataset": _to_jsonable(rows), "total": total} + return {"dataset": to_jsonable(rows), "total": total} def subscribe(self, job_id: str, *, after_seq: int | None = None) -> Subscription | None: """SSE subscribe: get replay buffer + live events stream.""" @@ -355,7 +341,7 @@ class JobManager: for q in self._subs: try: q.put_nowait(event) - except Exception: + except queue.Full: stale.append(q) if stale: self._subs = [q for q in self._subs if q not in stale] @@ -374,7 +360,7 @@ class JobManager: return coerce_event(q.get(timeout=timeout_sec)) except queue.Empty: return None - except Exception: + except (EOFError, OSError, ValueError): return None @staticmethod @@ -386,7 +372,7 @@ class JobManager: events.append(coerce_event(q.get_nowait())) except queue.Empty: return events - except Exception: + except (EOFError, OSError, ValueError): return events def _pump_loop(self) -> None: @@ -416,13 +402,10 @@ class JobManager: self._job.status = "error" self._job.error = self._job.error or "process exited" self._job.finished_at = time.time() - self._emit( - { - "type": f"job.{self._job.status}", - "ts": time.time(), - "job_id": self._job.job_id, - } + event_type = ( + EVENT_JOB_CANCELLED if self._job.status == "cancelled" else EVENT_JOB_ERROR ) + self._emit({"type": event_type, "ts": time.time(), "job_id": self._job.job_id}) return def _handle_event(self, job: Job, event: dict) -> None: @@ -433,9 +416,9 @@ class JobManager: with self._lock: if self._job is None or self._job.job_id != job.job_id: return - if et == "job.started": + if et == EVENT_JOB_STARTED: self._job.status = "active" - if et == "job.completed": + if et == EVENT_JOB_COMPLETED: self._job.status = "completed" self._job.finished_at = time.time() self._job.analysis = event.get("analysis") @@ -445,7 +428,7 @@ class JobManager: if self._job.progress.total and self._job.progress.total > 0: self._job.progress.done = self._job.progress.total self._job.progress.percent = 100.0 - if et == "job.error": + if et == EVENT_JOB_ERROR: self._job.status = "error" self._job.finished_at = time.time() self._job.error = event.get("error") or "error" diff --git a/studio/backend/core/data_recipe/jobs/parse.py b/studio/backend/core/data_recipe/jobs/parse.py index 99d1a85a79..6e2142adf2 100644 --- a/studio/backend/core/data_recipe/jobs/parse.py +++ b/studio/backend/core/data_recipe/jobs/parse.py @@ -4,6 +4,18 @@ import re from dataclasses import dataclass from typing import Any +from .constants import ( + STAGE_BATCH, + STAGE_COLUMN_CONFIG, + STAGE_CREATE, + STAGE_DAG, + STAGE_GENERATING, + STAGE_HEALTHCHECK, + STAGE_PREVIEW, + STAGE_PROFILING, + STAGE_SAMPLING, + USAGE_RESET_STAGES, +) from .types import Job, ModelUsage, Progress @@ -27,8 +39,7 @@ class ParsedUpdate: usage_rpm: float | None = None usage_section_start: bool | None = None -# welp, best effort to parse the logs and convert them to structured information so we can access it and read it properly on the client -# i couldnt find what datadesigner do for progress tracking besides the logs so il probably raise a pr in their repo to add some sort of progress monitoring +# kinda of a bummber but currently only option, Best effort parser from data-designer logs -> structured status for UI. _RE_SAMPLERS = re.compile( r"Preparing samplers to generate (?P\d+) records across (?P\d+) columns" ) @@ -52,31 +63,31 @@ def parse_log_message(msg: str) -> ParsedUpdate | None: m = _RE_SAMPLERS.search(msg) if m: return ParsedUpdate( - stage="sampling", + stage=STAGE_SAMPLING, rows=int(m.group("rows")), cols=int(m.group("cols")), ) if "Sorting column configs into a Directed Acyclic Graph" in msg: - return ParsedUpdate(stage="dag") + return ParsedUpdate(stage=STAGE_DAG) if "Running health checks for models" in msg: - return ParsedUpdate(stage="healthcheck") + return ParsedUpdate(stage=STAGE_HEALTHCHECK) if "Preview generation in progress" in msg: - return ParsedUpdate(stage="preview") + return ParsedUpdate(stage=STAGE_PREVIEW) if "Creating Data Designer dataset" in msg: - return ParsedUpdate(stage="create") + return ParsedUpdate(stage=STAGE_CREATE) if "Measuring dataset column statistics" in msg: - return ParsedUpdate(stage="profiling") + return ParsedUpdate(stage=STAGE_PROFILING) m = _RE_COLCFG.search(msg) if m: col = m.group("col") - return ParsedUpdate(stage="column_config", current_column=col) + return ParsedUpdate(stage=STAGE_COLUMN_CONFIG, current_column=col) m = _RE_PROCESSING_COL.search(msg) if m: col = m.group("col") - return ParsedUpdate(stage="generating", current_column=col) + return ParsedUpdate(stage=STAGE_GENERATING, current_column=col) m = _RE_PROGRESS.search(msg) if m: @@ -89,12 +100,12 @@ def parse_log_message(msg: str) -> ParsedUpdate | None: rate=float(m.group("rate")), eta_sec=float(m.group("eta")), ) - return ParsedUpdate(stage="generating", progress=p) + return ParsedUpdate(stage=STAGE_GENERATING, progress=p) m = _RE_BATCH.search(msg) if m: return ParsedUpdate( - stage="batch", + stage=STAGE_BATCH, batch_idx=int(m.group("idx")), batch_total=int(m.group("total")), ) @@ -132,7 +143,7 @@ def apply_update(job: Job, update: ParsedUpdate) -> None: job.stage = update.stage if update.current_column is not None: job.current_column = update.current_column - if update.stage == "generating" and update.current_column not in job._seen_generation_columns: + if update.stage == STAGE_GENERATING and update.current_column not in job._seen_generation_columns: job._seen_generation_columns.append(update.current_column) if update.rows is not None: job.rows = update.rows @@ -146,16 +157,8 @@ def apply_update(job: Job, update: ParsedUpdate) -> None: if update.batch_total is not None: job.batch.total = update.batch_total - if update.stage in { - "profiling", - "generating", - "sampling", - "healthcheck", - "dag", - "create", - "preview", - }: - # usage summary is a short block; reset once we move into the next stage. + if update.stage in USAGE_RESET_STAGES: + # usage summary is a short block so we reset once we move into the next stage. job._in_usage_summary = False if update.usage_section_start is not None: diff --git a/studio/backend/core/data_recipe/jobs/worker.py b/studio/backend/core/data_recipe/jobs/worker.py index ac27cd0d0b..8c0996b140 100644 --- a/studio/backend/core/data_recipe/jobs/worker.py +++ b/studio/backend/core/data_recipe/jobs/worker.py @@ -7,6 +7,8 @@ import traceback from pathlib import Path from typing import Any +from ..jsonable import to_jsonable +from .constants import EVENT_JOB_COMPLETED, EVENT_JOB_ERROR, EVENT_JOB_STARTED from ..service import build_config_builder, create_data_designer _PROJECT_ROOT = Path(__file__).resolve().parents[5] @@ -28,36 +30,10 @@ class _QueueLogHandler(logging.Handler): "message": record.getMessage(), } self._q.put(event) - except Exception: + except (OSError, RuntimeError, ValueError): pass -def _to_jsonable(value: Any) -> Any: - try: - import numpy as np # type: ignore - except Exception: # pragma: no cover - np = None # type: ignore - - if np is not None: - if isinstance(value, np.ndarray): - return value.tolist() - if isinstance(value, np.generic): - return value.item() - - if isinstance(value, dict): - return {str(k): _to_jsonable(v) for k, v in value.items()} - if isinstance(value, (list, tuple, set)): - return [_to_jsonable(v) for v in value] - - if hasattr(value, "isoformat") and callable(value.isoformat): - try: - return value.isoformat() - except Exception: - pass - - return value - - def run_job_process( *, event_queue, @@ -68,7 +44,7 @@ def run_job_process( Subprocess entrypoint. Sends events to `event_queue`. """ - event_queue.put({"type": "job.started", "ts": time.time()}) + event_queue.put({"type": EVENT_JOB_STARTED, "ts": time.time()}) try: from data_designer.config.run_config import RunConfig @@ -103,21 +79,21 @@ def run_job_process( analysis = ( None if results.analysis is None - else _to_jsonable(results.analysis.model_dump(mode="json")) + else to_jsonable(results.analysis.model_dump(mode="json")) ) dataset = ( [] if results.dataset is None - else _to_jsonable(results.dataset.to_dict(orient="records")) + else to_jsonable(results.dataset.to_dict(orient="records")) ) processor_artifacts = ( None if results.processor_artifacts is None - else _to_jsonable(results.processor_artifacts) + else to_jsonable(results.processor_artifacts) ) event_queue.put( { - "type": "job.completed", + "type": EVENT_JOB_COMPLETED, "ts": time.time(), "analysis": analysis, "dataset": dataset, @@ -128,13 +104,13 @@ def run_job_process( ) else: results = designer.create(builder, num_records=rows, dataset_name=dataset_name) - analysis = _to_jsonable(results.load_analysis().model_dump(mode="json")) + analysis = to_jsonable(results.load_analysis().model_dump(mode="json")) if merge_batches: _merge_batches_to_single_parquet(results.artifact_storage.base_dataset_path) artifact_path = str(results.artifact_storage.base_dataset_path) event_queue.put( { - "type": "job.completed", + "type": EVENT_JOB_COMPLETED, "ts": time.time(), "analysis": analysis, "artifact_path": artifact_path, @@ -144,7 +120,7 @@ def run_job_process( except Exception as exc: event_queue.put( { - "type": "job.error", + "type": EVENT_JOB_ERROR, "ts": time.time(), "error": str(exc), "stack": traceback.format_exc(limit=20), @@ -160,7 +136,7 @@ def _merge_batches_to_single_parquet(base_dataset_path: Path) -> None: try: from data_designer.config.utils.io_helpers import read_parquet_dataset - except Exception: + except ImportError: return dataframe = read_parquet_dataset(parquet_dir) diff --git a/studio/backend/core/data_recipe/jsonable.py b/studio/backend/core/data_recipe/jsonable.py new file mode 100644 index 0000000000..aa6e1d6b2e --- /dev/null +++ b/studio/backend/core/data_recipe/jsonable.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from typing import Any + + +def to_jsonable(value: Any) -> Any: + """Convert numpy/pandas-ish values into plain JSON-safe values.""" + try: + import numpy as np # type: ignore + except ImportError: # pragma: no cover + np = None # type: ignore + + if np is not None: + if isinstance(value, np.ndarray): + return value.tolist() + if isinstance(value, np.generic): + return value.item() + + if isinstance(value, dict): + return {str(k): to_jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple, set)): + return [to_jsonable(v) for v in value] + + if hasattr(value, "isoformat") and callable(value.isoformat): + try: + return value.isoformat() + except (TypeError, ValueError): + return value + + return value diff --git a/studio/backend/core/data_recipe/service.py b/studio/backend/core/data_recipe/service.py index b04c6bbf00..2d11cb2845 100644 --- a/studio/backend/core/data_recipe/service.py +++ b/studio/backend/core/data_recipe/service.py @@ -3,32 +3,7 @@ from __future__ import annotations import os from typing import Any -def _to_jsonable(value: Any) -> Any: - # pydantic/fastapi can't serialize numpy arrays/scalars. - try: - import numpy as np # type: ignore - except Exception: # pragma: no cover - np = None # type: ignore - - if np is not None: - if isinstance(value, np.ndarray): - return value.tolist() - if isinstance(value, np.generic): - return value.item() - - if isinstance(value, dict): - return {str(k): _to_jsonable(v) for k, v in value.items()} - if isinstance(value, (list, tuple, set)): - return [_to_jsonable(v) for v in value] - - # pandas Timestamp/date-like - if hasattr(value, "isoformat") and callable(value.isoformat): - try: - return value.isoformat() - except Exception: - pass - - return value +from .jsonable import to_jsonable def build_model_providers(recipe: dict[str, Any]): @@ -158,17 +133,17 @@ def preview_recipe( dataset: list[dict[str, Any]] = [] if results.dataset is not None: raw_rows = results.dataset.to_dict(orient="records") - dataset = [_to_jsonable(row) for row in raw_rows] + dataset = [to_jsonable(row) for row in raw_rows] artifacts = ( None if results.processor_artifacts is None - else _to_jsonable(results.processor_artifacts) + else to_jsonable(results.processor_artifacts) ) analysis = ( None if results.analysis is None - else _to_jsonable(results.analysis.model_dump(mode="json")) + else to_jsonable(results.analysis.model_dump(mode="json")) ) return dataset, artifacts, analysis diff --git a/studio/backend/routes/data_recipe/__init__.py b/studio/backend/routes/data_recipe/__init__.py new file mode 100644 index 0000000000..dc0301d1c9 --- /dev/null +++ b/studio/backend/routes/data_recipe/__init__.py @@ -0,0 +1,23 @@ +"""Data Recipe route package.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +from fastapi import APIRouter + +backend_path = Path(__file__).parent.parent.parent +if str(backend_path) not in sys.path: + sys.path.insert(0, str(backend_path)) + +from .jobs import router as jobs_router +from .seed import router as seed_router +from .validate import router as validate_router + +router = APIRouter() +router.include_router(seed_router) +router.include_router(validate_router) +router.include_router(jobs_router) + +__all__ = ["router"] diff --git a/studio/backend/routes/data_recipe/jobs.py b/studio/backend/routes/data_recipe/jobs.py new file mode 100644 index 0000000000..ffbded9474 --- /dev/null +++ b/studio/backend/routes/data_recipe/jobs.py @@ -0,0 +1,143 @@ +"""Job lifecycle endpoints for data recipe.""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import JSONResponse, StreamingResponse +from pydantic import ValidationError + +from core.data_recipe.jobs import get_job_manager +from models.data_recipe import JobCreateResponse, RecipePayload + +router = APIRouter() + + +@router.post("/jobs", response_class=JSONResponse, response_model=JobCreateResponse) +def create_job(payload: RecipePayload): + recipe = payload.recipe + if not recipe.get("columns"): + raise HTTPException(status_code=400, detail="Recipe must include columns.") + + run: dict[str, Any] = payload.run or {} + run.pop("artifact_path", None) + run.pop("dataset_name", None) + execution_type = str(run.get("execution_type") or "full").strip().lower() + if execution_type not in {"preview", "full"}: + raise HTTPException(status_code=400, detail="invalid execution_type: must be 'preview' or 'full'") + run["execution_type"] = execution_type + run_config_raw = run.get("run_config") + if run_config_raw is not None: + try: + from data_designer.config.run_config import RunConfig + + RunConfig.model_validate(run_config_raw) + except (ImportError, ValidationError, TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail=f"invalid run_config: {exc}") from exc + + mgr = get_job_manager() + try: + job_id = mgr.start(recipe=recipe, run=run) + except RuntimeError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + return {"job_id": job_id} + + +@router.get("/jobs/{job_id}/status") +def job_status(job_id: str): + mgr = get_job_manager() + state = mgr.get_status(job_id) + if state is None: + raise HTTPException(status_code=404, detail="job not found") + return state + + +@router.get("/jobs/current") +def current_job(): + mgr = get_job_manager() + state = mgr.get_current_status() + if state is None: + raise HTTPException(status_code=404, detail="no job") + return state + + +@router.post("/jobs/{job_id}/cancel") +def cancel_job(job_id: str): + mgr = get_job_manager() + ok = mgr.cancel(job_id) + if not ok: + raise HTTPException(status_code=404, detail="job not found") + return mgr.get_status(job_id) + + +@router.get("/jobs/{job_id}/analysis") +def job_analysis(job_id: str): + mgr = get_job_manager() + analysis = mgr.get_analysis(job_id) + if analysis is None: + raise HTTPException(status_code=404, detail="analysis not ready") + return analysis + + +@router.get("/jobs/{job_id}/dataset") +def job_dataset( + job_id: str, + limit: int = Query(default=20, ge=1, le=500), + offset: int = Query(default=0, ge=0), +): + mgr = get_job_manager() + result = mgr.get_dataset(job_id, limit=limit, offset=offset) + if result is None: + raise HTTPException(status_code=404, detail="dataset not ready") + if "error" in result: + raise HTTPException(status_code=422, detail=result["error"]) + return { + "dataset": result["dataset"], + "total": result["total"], + "limit": limit, + "offset": offset, + } + + +@router.get("/jobs/{job_id}/events") +async def job_events(request: Request, job_id: str): + mgr = get_job_manager() + last_id = request.headers.get("last-event-id") + after_seq: int | None = None + if last_id: + try: + after_seq = int(str(last_id).strip()) + except (TypeError, ValueError): + after_seq = None + + after_q = request.query_params.get("after") + if after_q: + try: + after_seq = int(str(after_q).strip()) + except (TypeError, ValueError): + pass + + sub = mgr.subscribe(job_id, after_seq=after_seq) + if sub is None: + raise HTTPException(status_code=404, detail="job not found") + + async def gen(): + try: + for event in sub.replay: + yield sub.format_sse(event) + + while True: + if await request.is_disconnected(): + break + event = await sub.next_event(timeout_sec=1.0) + if event is None: + continue + yield sub.format_sse(event) + finally: + mgr.unsubscribe(sub) + + return StreamingResponse(gen(), media_type="text/event-stream") diff --git a/studio/backend/routes/data_recipe.py b/studio/backend/routes/data_recipe/seed.py similarity index 53% rename from studio/backend/routes/data_recipe.py rename to studio/backend/routes/data_recipe/seed.py index 8a713f2cec..eb02ab2bbf 100644 --- a/studio/backend/routes/data_recipe.py +++ b/studio/backend/routes/data_recipe/seed.py @@ -1,42 +1,24 @@ -""" -Data Recipe routes (DataDesigner runner). -""" +"""Seed inspect endpoints for data recipe.""" from __future__ import annotations import base64 import binascii -import sys from itertools import islice from pathlib import Path from typing import Any from uuid import uuid4 -from fastapi import APIRouter, HTTPException, Query, Request -from fastapi.responses import JSONResponse, StreamingResponse +from fastapi import APIRouter, HTTPException -# same thing as other files do -backend_path = Path(__file__).parent.parent.parent -if str(backend_path) not in sys.path: - sys.path.insert(0, str(backend_path)) - -from core.data_recipe.jobs import get_job_manager -from core.data_recipe.service import ( - build_config_builder, - create_data_designer, - validate_recipe, -) from models.data_recipe import ( - JobCreateResponse, - RecipePayload, SeedInspectRequest, - SeedInspectUploadRequest, SeedInspectResponse, - ValidateError, - ValidateResponse, + SeedInspectUploadRequest, ) router = APIRouter() + DATA_EXTS = (".parquet", ".jsonl", ".json", ".csv") DEFAULT_SPLIT = "train" LOCAL_UPLOAD_EXTS = {".csv", ".json", ".jsonl"} @@ -70,20 +52,21 @@ def _normalize_optional_text(value: str | None) -> str | None: def _list_hf_data_files(*, dataset_name: str, token: str | None) -> list[str]: try: from huggingface_hub import HfApi - + from huggingface_hub.utils import HfHubHTTPError + except ImportError: + return [] + try: api = HfApi() repo_files = api.list_repo_files(dataset_name, repo_type="dataset", token=token) return [file for file in repo_files if file.lower().endswith(DATA_EXTS)] - except Exception: + except (HfHubHTTPError, OSError, ValueError): return [] -def _select_best_file(data_files: list[str], split: str | None) -> str | None: +def _select_best_file(data_files: list[str]) -> str | None: if not data_files: return None - if not split: - return data_files[0] - split_lower = split.lower() + split_lower = DEFAULT_SPLIT def score(path: str) -> tuple[int, int]: name = path.lower() @@ -102,8 +85,8 @@ def _select_best_file(data_files: list[str], split: str | None) -> str | None: return sorted(data_files, key=score)[0] -def _resolve_seed_hf_path(dataset_name: str, data_files: list[str], split: str | None) -> str | None: - selected = _select_best_file(data_files, split) +def _resolve_seed_hf_path(dataset_name: str, data_files: list[str]) -> str | None: + selected = _select_best_file(data_files) if not selected: return None @@ -177,7 +160,7 @@ def _decode_base64_payload(content_base64: str) -> bytes: def _read_preview_rows_from_local_file(path: Path, preview_size: int) -> list[dict[str, Any]]: try: import pandas as pd - except Exception as exc: + except ImportError as exc: raise HTTPException(status_code=500, detail=f"seed inspect dependencies unavailable: {exc}") from exc ext = path.suffix.lower() @@ -189,68 +172,19 @@ def _read_preview_rows_from_local_file(path: Path, preview_size: int) -> list[di elif ext == ".json": try: df = pd.read_json(path, lines=True).head(preview_size) - except Exception: + except ValueError: df = pd.read_json(path).head(preview_size) else: raise HTTPException(status_code=422, detail=f"unsupported file type: {ext}") except HTTPException: raise - except Exception as exc: + except (ValueError, OSError) as exc: raise HTTPException(status_code=422, detail=f"seed inspect failed: {exc}") from exc rows = df.to_dict(orient="records") return _serialize_preview_rows(rows) -def _collect_validation_errors(recipe: dict[str, Any]) -> list[ValidateError]: - try: - from data_designer.engine.compiler import ( - _add_internal_row_id_column_if_needed, - _get_allowed_references, - _resolve_and_add_seed_columns, - ) - from data_designer.engine.validation import ( - ViolationLevel, - validate_data_designer_config, - ) - except Exception: - return [] - - try: - builder = build_config_builder(recipe) - designer = create_data_designer(recipe) - resource_provider = designer._create_resource_provider( # type: ignore[attr-defined] - "validate-configuration", - builder, - ) - config = builder.build() - _resolve_and_add_seed_columns(config, resource_provider.seed_reader) - _add_internal_row_id_column_if_needed(config) - violations = validate_data_designer_config( - columns=config.columns, - processor_configs=config.processors or [], - allowed_references=_get_allowed_references(config), - ) - except Exception: - return [] - - errors: list[ValidateError] = [] - for violation in violations: - if violation.level != ViolationLevel.ERROR: - continue - code = getattr(violation.type, "value", None) - path = violation.column if violation.column else None - message = str(violation.message).strip() or "Validation failed." - errors.append( - ValidateError( - message=message, - path=path, - code=code, - ) - ) - return errors - - @router.post("/seed/inspect", response_model=SeedInspectResponse) def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: dataset_name = payload.dataset_name.strip() @@ -259,10 +193,10 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: try: from datasets import load_dataset - except Exception as exc: + except ImportError as exc: raise HTTPException(status_code=500, detail=f"seed inspect dependencies unavailable: {exc}") from exc - split = (payload.split or DEFAULT_SPLIT).strip() or DEFAULT_SPLIT + split = DEFAULT_SPLIT subset = _normalize_optional_text(payload.subset) token = _normalize_optional_text(payload.hf_token) preview_size = int(payload.preview_size) @@ -270,7 +204,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: preview_rows: list[dict[str, Any]] = [] data_files = _list_hf_data_files(dataset_name=dataset_name, token=token) - selected_file = _select_best_file(data_files, split) + selected_file = _select_best_file(data_files) if selected_file: try: single_file_kwargs = _build_stream_load_kwargs( @@ -285,7 +219,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: load_kwargs=single_file_kwargs, preview_size=preview_size, ) - except Exception: + except (ValueError, OSError, RuntimeError): preview_rows = [] if not preview_rows: @@ -301,7 +235,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: load_kwargs=split_kwargs, preview_size=preview_size, ) - except Exception as exc: + except (ValueError, OSError, RuntimeError) as exc: raise HTTPException(status_code=422, detail=f"seed inspect failed: {exc}") from exc if not preview_rows: @@ -310,10 +244,9 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: columns = _extract_columns(preview_rows) if not data_files: - # Best effort path fallback when file list is unavailable. resolved_path = f"datasets/{dataset_name}/**/*.parquet" else: - resolved_path = _resolve_seed_hf_path(dataset_name, data_files, split) + resolved_path = _resolve_seed_hf_path(dataset_name, data_files) if not resolved_path: raise HTTPException(status_code=422, detail="unable to resolve seed dataset path") @@ -322,7 +255,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse: resolved_path=resolved_path, columns=columns, preview_rows=preview_rows, - split=split, + split=None, subset=subset, ) @@ -363,159 +296,3 @@ def inspect_seed_upload(payload: SeedInspectUploadRequest) -> SeedInspectRespons split=None, subset=None, ) - - -@router.post("/validate", response_model=ValidateResponse) -def validate(payload: RecipePayload) -> ValidateResponse: - recipe = payload.recipe - if not recipe.get("columns"): - return ValidateResponse( - valid=False, - errors=[ValidateError(message="Recipe must include columns.")], - ) - - try: - validate_recipe(recipe) - except RuntimeError as exc: - raise HTTPException(status_code=503, detail=str(exc)) from exc - except Exception as exc: - detail = str(exc).strip() or "Validation failed." - parsed_errors = _collect_validation_errors(recipe) - return ValidateResponse( - valid=False, - errors=parsed_errors or [ValidateError(message=detail)], - raw_detail=detail, - ) - - return ValidateResponse(valid=True) - - -@router.post("/jobs", response_class=JSONResponse, response_model=JobCreateResponse) -def create_job(payload: RecipePayload): - recipe = payload.recipe - if not recipe.get("columns"): - raise HTTPException(status_code=400, detail="Recipe must include columns.") - - run: dict[str, Any] = payload.run or {} - run.pop("artifact_path", None) - run.pop("dataset_name", None) - execution_type = str(run.get("execution_type") or "full").strip().lower() - if execution_type not in {"preview", "full"}: - raise HTTPException(status_code=400, detail="invalid execution_type: must be 'preview' or 'full'") - run["execution_type"] = execution_type - run_config_raw = run.get("run_config") - if run_config_raw is not None: - try: - from data_designer.config.run_config import RunConfig - - RunConfig.model_validate(run_config_raw) - except Exception as exc: - raise HTTPException(status_code=400, detail=f"invalid run_config: {exc}") from exc - - mgr = get_job_manager() - try: - job_id = mgr.start(recipe=recipe, run=run) - except RuntimeError as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc - except ValueError as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc - - return {"job_id": job_id} - - -@router.get("/jobs/{job_id}/status") -def job_status(job_id: str): - mgr = get_job_manager() - state = mgr.get_status(job_id) - if state is None: - raise HTTPException(status_code=404, detail="job not found") - return state - - -@router.get("/jobs/current") -def current_job(): - mgr = get_job_manager() - state = mgr.get_current_status() - if state is None: - raise HTTPException(status_code=404, detail="no job") - return state - - -@router.post("/jobs/{job_id}/cancel") -def cancel_job(job_id: str): - mgr = get_job_manager() - ok = mgr.cancel(job_id) - if not ok: - raise HTTPException(status_code=404, detail="job not found") - return mgr.get_status(job_id) - - -@router.get("/jobs/{job_id}/analysis") -def job_analysis(job_id: str): - mgr = get_job_manager() - analysis = mgr.get_analysis(job_id) - if analysis is None: - raise HTTPException(status_code=404, detail="analysis not ready") - return analysis - - -@router.get("/jobs/{job_id}/dataset") -def job_dataset( - job_id: str, - limit: int = Query(default=20, ge=1, le=500), - offset: int = Query(default=0, ge=0), -): - mgr = get_job_manager() - result = mgr.get_dataset(job_id, limit=limit, offset=offset) - if result is None: - raise HTTPException(status_code=404, detail="dataset not ready") - if "error" in result: - raise HTTPException(status_code=422, detail=result["error"]) - return { - "dataset": result["dataset"], - "total": result["total"], - "limit": limit, - "offset": offset, - } - - -@router.get("/jobs/{job_id}/events") -async def job_events(request: Request, job_id: str): - mgr = get_job_manager() - last_id = request.headers.get("last-event-id") - after_seq: int | None = None - if last_id: - try: - after_seq = int(str(last_id).strip()) - except Exception: - after_seq = None - - # EventSource can't set custom headers on first connect after a full page refresh, - # so allow resume via query param too: /events?after= - after_q = request.query_params.get("after") - if after_q: - try: - after_seq = int(str(after_q).strip()) - except Exception: - pass - - sub = mgr.subscribe(job_id, after_seq=after_seq) - if sub is None: - raise HTTPException(status_code=404, detail="job not found") - - async def gen(): - try: - for event in sub.replay: - yield sub.format_sse(event) - - while True: - if await request.is_disconnected(): - break - event = await sub.next_event(timeout_sec=1.0) - if event is None: - continue - yield sub.format_sse(event) - finally: - mgr.unsubscribe(sub) - - return StreamingResponse(gen(), media_type="text/event-stream") diff --git a/studio/backend/routes/data_recipe/validate.py b/studio/backend/routes/data_recipe/validate.py new file mode 100644 index 0000000000..a8755f9410 --- /dev/null +++ b/studio/backend/routes/data_recipe/validate.py @@ -0,0 +1,90 @@ +"""Validation endpoints for data recipe.""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException + +from core.data_recipe.service import ( + build_config_builder, + create_data_designer, + validate_recipe, +) +from models.data_recipe import RecipePayload, ValidateError, ValidateResponse + +router = APIRouter() + + +def _collect_validation_errors(recipe: dict[str, Any]) -> list[ValidateError]: + try: + from data_designer.engine.compiler import ( + _add_internal_row_id_column_if_needed, + _get_allowed_references, + _resolve_and_add_seed_columns, + ) + from data_designer.engine.validation import ( + ViolationLevel, + validate_data_designer_config, + ) + except ImportError: + return [] + + try: + builder = build_config_builder(recipe) + designer = create_data_designer(recipe) + resource_provider = designer._create_resource_provider( # type: ignore[attr-defined] + "validate-configuration", + builder, + ) + config = builder.build() + _resolve_and_add_seed_columns(config, resource_provider.seed_reader) + _add_internal_row_id_column_if_needed(config) + violations = validate_data_designer_config( + columns=config.columns, + processor_configs=config.processors or [], + allowed_references=_get_allowed_references(config), + ) + except (TypeError, ValueError, AttributeError): + return [] + + errors: list[ValidateError] = [] + for violation in violations: + if violation.level != ViolationLevel.ERROR: + continue + code = getattr(violation.type, "value", None) + path = violation.column if violation.column else None + message = str(violation.message).strip() or "Validation failed." + errors.append( + ValidateError( + message=message, + path=path, + code=code, + ) + ) + return errors + + +@router.post("/validate", response_model=ValidateResponse) +def validate(payload: RecipePayload) -> ValidateResponse: + recipe = payload.recipe + if not recipe.get("columns"): + return ValidateResponse( + valid=False, + errors=[ValidateError(message="Recipe must include columns.")], + ) + + try: + validate_recipe(recipe) + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except Exception as exc: + detail = str(exc).strip() or "Validation failed." + parsed_errors = _collect_validation_errors(recipe) + return ValidateResponse( + valid=False, + errors=parsed_errors or [ValidateError(message=detail)], + raw_detail=detail, + ) + + return ValidateResponse(valid=True) diff --git a/studio/frontend/src/components/ui/combobox.tsx b/studio/frontend/src/components/ui/combobox.tsx index 9c1e970c57..8ccc40c95f 100644 --- a/studio/frontend/src/components/ui/combobox.tsx +++ b/studio/frontend/src/components/ui/combobox.tsx @@ -162,20 +162,20 @@ function ComboboxContent({ - + align={align} + alignOffset={alignOffset} + anchor={anchor} + className="isolate z-[120] pointer-events-auto" + > + ); diff --git a/studio/frontend/src/features/recipe-studio/blocks/definitions.ts b/studio/frontend/src/features/recipe-studio/blocks/definitions.ts index 75d860290e..073cce928e 100644 --- a/studio/frontend/src/features/recipe-studio/blocks/definitions.ts +++ b/studio/frontend/src/features/recipe-studio/blocks/definitions.ts @@ -125,7 +125,7 @@ const BLOCK_DEFINITIONS: BlockDefinition[] = [ { kind: "seed", type: "seed_local", - title: "Local file", + title: "Structured file", description: "Upload CSV/JSON/JSONL and use rows as seed context.", icon: DocumentCodeIcon, dialogKey: "seed", diff --git a/studio/frontend/src/features/recipe-studio/components/block-sheet.tsx b/studio/frontend/src/features/recipe-studio/components/block-sheet.tsx index d9e278c05d..be301a36fd 100644 --- a/studio/frontend/src/features/recipe-studio/components/block-sheet.tsx +++ b/studio/frontend/src/features/recipe-studio/components/block-sheet.tsx @@ -1,4 +1,5 @@ import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; import { Sheet, SheetContent, @@ -12,17 +13,24 @@ import { CodeIcon, Copy02Icon, type Database02Icon, + DragDropVerticalIcon, PlusSignIcon, Tick02Icon, Upload01Icon, } from "@hugeicons/core-free-icons"; import { HugeiconsIcon } from "@hugeicons/react"; -import { type ReactElement, useMemo, useState } from "react"; +import { + type DragEvent as ReactDragEvent, + type ReactElement, + useMemo, + useState, +} from "react"; import { RECIPE_FLOATING_ICON_BUTTON_CLASS } from "./recipe-floating-icon-button-class"; import type { LlmType, SamplerType } from "../types"; import { BLOCK_GROUPS, getBlocksForKind, + type BlockType, type SeedBlockType, } from "../blocks/registry"; @@ -62,6 +70,12 @@ type BlockSheetProps = { onImport: () => void; }; +export const RECIPE_BLOCK_DND_MIME = "application/x-recipe-studio-block"; +export type RecipeBlockDragPayload = { + kind: SheetKind; + type: BlockType; +}; + function getSheetTitle(sheetView: SheetView): string { if (sheetView === "root") { return "Add a block"; @@ -103,6 +117,15 @@ const ROOT_GROUPS: RootGroup[] = [ icon: CodeIcon, }, ]; +const SEARCHABLE_KINDS: SheetKind[] = [ + "sampler", + "seed", + "llm", + "expression", + "note", +]; +const PROCESSOR_TITLE = "Schema Transform"; +const PROCESSOR_DESCRIPTION = "Transform final dataset schema."; function BlockSheetButton({ icon, @@ -110,22 +133,30 @@ function BlockSheetButton({ description, onClick, isActive = false, + draggable = false, + onDragStart, + trailing = "chevron", }: { icon: typeof Database02Icon; title: string; description: string; onClick: () => void; isActive?: boolean; + draggable?: boolean; + onDragStart?: (event: ReactDragEvent) => void; + trailing?: "chevron" | "drag" | "none"; }): ReactElement { return ( ); } @@ -162,11 +201,17 @@ export function BlockSheet({ }: BlockSheetProps): ReactElement { const sheetTitle = getSheetTitle(sheetView); const [uncontrolledOpen, setUncontrolledOpen] = useState(false); + const [search, setSearch] = useState(""); const expressionBlocks = useMemo(() => getBlocksForKind("expression"), []); const noteBlocks = useMemo(() => getBlocksForKind("note"), []); const seedBlocks = useMemo(() => getBlocksForKind("seed"), []); const isControlled = typeof open === "boolean"; const sheetOpen = isControlled ? (open as boolean) : uncontrolledOpen; + const normalizedSearch = search.trim().toLowerCase(); + const hasSearch = normalizedSearch.length > 0; + const isProcessorView = sheetView === "processor"; + const isRootView = sheetView === "root"; + const isScopedBlockView = !isRootView && !isProcessorView; const setSheetOpen = (nextOpen: boolean) => { if (!isControlled) { @@ -174,6 +219,95 @@ export function BlockSheet({ } onOpenChange?.(nextOpen); }; + const matchesSearch = (title: string, description: string) => + title.toLowerCase().includes(normalizedSearch) || + description.toLowerCase().includes(normalizedSearch); + + const searchableBlocks = useMemo( + () => SEARCHABLE_KINDS.flatMap((kind) => getBlocksForKind(kind)), + [], + ); + const rootSearchBlocks = useMemo(() => { + if (!hasSearch) { + return []; + } + return searchableBlocks.filter((item) => + matchesSearch(item.title, item.description), + ); + }, [hasSearch, searchableBlocks, normalizedSearch]); + + const scopedBlocks = useMemo(() => { + if (!isScopedBlockView) { + return []; + } + const blocks = getBlocksForKind(VIEW_KIND[sheetView] ?? "sampler"); + if (!hasSearch) { + return blocks; + } + return blocks.filter((item) => matchesSearch(item.title, item.description)); + }, [hasSearch, isScopedBlockView, normalizedSearch, sheetView]); + + const rootGroups = useMemo(() => { + if (!hasSearch) { + return ROOT_GROUPS; + } + return ROOT_GROUPS.filter((group) => { + if (matchesSearch(group.title, group.description)) { + return true; + } + if (group.kind === "processor") { + return matchesSearch(PROCESSOR_TITLE, PROCESSOR_DESCRIPTION); + } + return getBlocksForKind(group.kind).some((item) => + matchesSearch(item.title, item.description), + ); + }); + }, [hasSearch, normalizedSearch]); + const showNoMatches = + (isRootView && hasSearch && rootSearchBlocks.length === 0) || + (isScopedBlockView && scopedBlocks.length === 0) || + (isProcessorView && + hasSearch && + !matchesSearch(PROCESSOR_TITLE, PROCESSOR_DESCRIPTION)); + + const buildDragStart = + (kind: SheetKind, type: BlockType) => + (event: ReactDragEvent) => { + const payload: RecipeBlockDragPayload = { kind, type }; + const serialized = JSON.stringify(payload); + event.dataTransfer.setData(RECIPE_BLOCK_DND_MIME, serialized); + event.dataTransfer.setData("text/plain", serialized); + event.dataTransfer.effectAllowed = "copy"; + }; + const getTrailing = (_kind: SheetKind): "drag" => "drag"; + const onBlockClick = (kind: SheetKind, type: BlockType) => { + setSheetOpen(false); + if (kind === "sampler") { + onAddSampler(type as SamplerType); + return; + } + if (kind === "seed") { + onAddSeed(type as SeedBlockType); + return; + } + if (kind === "llm") { + if (type === "model_provider") { + onAddModelProvider(); + return; + } + if (type === "model_config") { + onAddModelConfig(); + return; + } + onAddLlm(type as LlmType); + return; + } + if (kind === "expression") { + onAddExpression(); + return; + } + onAddMarkdownNote(); + }; return (
@@ -183,6 +317,7 @@ export function BlockSheet({ setSheetOpen(nextOpen); if (nextOpen) { onViewChange("root"); + setSearch(""); } }} > @@ -206,7 +341,7 @@ export function BlockSheet({ className="absolute gap-0 p-0 shadow-none" overlayClassName="bg-transparent pointer-events-none backdrop-blur-none supports-backdrop-filter:backdrop-blur-none" > - +
{sheetView !== "root" && (
+ setSearch(event.target.value)} + placeholder="Search blocks..." + className="corner-squircle mt-3 h-9" + />
- {sheetView === "root" && - ROOT_GROUPS.map((item, index) => ( + {isRootView && + hasSearch && + rootSearchBlocks.map((item, index) => ( + onBlockClick(item.kind, item.type)} + /> + ))} + {isRootView && + !hasSearch && + rootGroups.map((item, index) => ( { if (item.kind === "processor") { setSheetOpen(false); @@ -256,18 +426,20 @@ export function BlockSheet({ }} /> ))} - {sheetView === "processor" && ( - + {isProcessorView && ( + (!hasSearch || + matchesSearch(PROCESSOR_TITLE, PROCESSOR_DESCRIPTION)) && ( + + ) )} - {sheetView !== "root" && - sheetView !== "processor" && - getBlocksForKind(VIEW_KIND[sheetView] ?? "sampler").map( + {isScopedBlockView && + scopedBlocks.map( (item, index) => ( { - setSheetOpen(false); - if (item.kind === "sampler") { - onAddSampler(item.type as SamplerType); - } else if (item.kind === "seed") { - onAddSeed(item.type as SeedBlockType); - } else if (item.kind === "llm") { - if (item.type === "model_provider") { - onAddModelProvider(); - } else if (item.type === "model_config") { - onAddModelConfig(); - } else { - onAddLlm(item.type as LlmType); - } - } else if (item.kind === "expression") { - onAddExpression(); - } else { - onAddMarkdownNote(); - } - }} + draggable={true} + onDragStart={buildDragStart(item.kind, item.type)} + trailing={getTrailing(item.kind)} + onClick={() => onBlockClick(item.kind, item.type)} /> ), )} + {showNoMatches && ( +

+ No blocks match. +

+ )}
diff --git a/studio/frontend/src/features/recipe-studio/components/chip-input.tsx b/studio/frontend/src/features/recipe-studio/components/chip-input.tsx index 5162b73e03..a7dd5ccc16 100644 --- a/studio/frontend/src/features/recipe-studio/components/chip-input.tsx +++ b/studio/frontend/src/features/recipe-studio/components/chip-input.tsx @@ -1,7 +1,15 @@ import { Button } from "@/components/ui/button"; import { Cancel01Icon } from "@hugeicons/core-free-icons"; import { HugeiconsIcon } from "@hugeicons/react"; -import { type KeyboardEvent, type ReactElement, useId, useMemo, useState } from "react"; +import { + type KeyboardEvent, + type ReactElement, + useEffect, + useId, + useMemo, + useRef, + useState, +} from "react"; type ChipInputProps = { values: string[]; @@ -19,12 +27,28 @@ export function ChipInput({ suggestions, }: ChipInputProps): ReactElement { const [draft, setDraft] = useState(""); + const [isWrapped, setIsWrapped] = useState(false); + const containerRef = useRef(null); const listId = useId(); const suggestionSet = useMemo( () => new Set((suggestions ?? []).map((value) => value.trim())), [suggestions], ); + useEffect(() => { + const element = containerRef.current; + if (!element) { + return; + } + const syncWrapped = () => { + setIsWrapped(element.clientHeight > 44); + }; + syncWrapped(); + const observer = new ResizeObserver(syncWrapped); + observer.observe(element); + return () => observer.disconnect(); + }, [values.length, draft]); + function addValue(rawValue: string, allowAny: boolean): void { const trimmed = rawValue.trim(); if (!trimmed) { @@ -55,7 +79,10 @@ export function ChipInput({ } return ( -
+
{values.map((value, index) => ( { refreshNodeInternals(); requestAnimationFrame(() => { - fitView({ duration: 250 }); + fitView({ + duration: 250, + nodes: getFitNodeIdsIgnoringNotes(getNodes()), + }); }); }); - }, [fitView, onLayout, refreshNodeInternals]); + }, [fitView, getNodes, onLayout, refreshNodeInternals]); const handleToggleDirection = useCallback(() => { onToggleDirection(); requestAnimationFrame(() => { - refreshNodeInternals(); + onLayout(); requestAnimationFrame(() => { refreshNodeInternals(); + requestAnimationFrame(() => { + fitView({ + duration: 250, + nodes: getFitNodeIdsIgnoringNotes(getNodes()), + }); + }); }); }); - }, [onToggleDirection, refreshNodeInternals]); + }, [fitView, getNodes, onLayout, onToggleDirection, refreshNodeInternals]); return ( diff --git a/studio/frontend/src/features/recipe-studio/components/controls/viewport-controls.tsx b/studio/frontend/src/features/recipe-studio/components/controls/viewport-controls.tsx index 27e225f39e..e104841100 100644 --- a/studio/frontend/src/features/recipe-studio/components/controls/viewport-controls.tsx +++ b/studio/frontend/src/features/recipe-studio/components/controls/viewport-controls.tsx @@ -2,6 +2,7 @@ import { type ReactElement, useCallback } from "react"; import { Lock, LockOpen, Maximize2, Minus, Plus } from "lucide-react"; import { Panel, useReactFlow } from "@xyflow/react"; import { Button } from "@/components/ui/button"; +import { getFitNodeIdsIgnoringNotes } from "../../utils/graph/fit-view"; import { RECIPE_FLOATING_ICON_BUTTON_CLASS } from "../recipe-floating-icon-button-class"; type ViewportControlsProps = { @@ -13,7 +14,7 @@ export function ViewportControls({ interactive, onToggleInteractive, }: ViewportControlsProps): ReactElement { - const { zoomIn, zoomOut, fitView } = useReactFlow(); + const { zoomIn, zoomOut, fitView, getNodes } = useReactFlow(); const handleZoomIn = useCallback(() => { zoomIn({ duration: 150 }); @@ -24,8 +25,11 @@ export function ViewportControls({ }, [zoomOut]); const handleFitView = useCallback(() => { - fitView({ duration: 250 }); - }, [fitView]); + fitView({ + duration: 250, + nodes: getFitNodeIdsIgnoringNotes(getNodes()), + }); + }, [fitView, getNodes]); return ( diff --git a/studio/frontend/src/features/recipe-studio/components/inline/inline-expression.tsx b/studio/frontend/src/features/recipe-studio/components/inline/inline-expression.tsx index 94c5470f49..206a068a78 100644 --- a/studio/frontend/src/features/recipe-studio/components/inline/inline-expression.tsx +++ b/studio/frontend/src/features/recipe-studio/components/inline/inline-expression.tsx @@ -1,4 +1,3 @@ -import { Badge } from "@/components/ui/badge"; import { Input } from "@/components/ui/input"; import { Select, @@ -10,7 +9,9 @@ import { import type { ReactElement } from "react"; import { useRecipeStudioStore } from "../../stores/recipe-studio"; import type { ExpressionConfig, ExpressionDtype } from "../../types"; +import { findInvalidJinjaReferences } from "../../utils/refs"; import { getAvailableVariableEntries } from "../../utils/variables"; +import { AvailableReferencesInline } from "../shared/available-references-inline"; import { InlineField } from "./inline-field"; type InlineExpressionProps = { @@ -26,6 +27,10 @@ export function InlineExpression({ }: InlineExpressionProps): ReactElement { const configs = useRecipeStudioStore((state) => state.configs); const vars = getAvailableVariableEntries(configs, config.id); + const invalidRefs = findInvalidJinjaReferences( + config.expr, + vars.map((entry) => entry.name), + ); return (
@@ -52,32 +57,14 @@ export function InlineExpression({ 0} placeholder="{{ column_name }}" value={config.expr} onChange={(event) => onUpdate({ expr: event.target.value })} />
- {vars.length > 0 && ( -
-

Available references

-
- {vars.map((v) => ( - - {v.name} - - ))} -
-
- )} +
); } diff --git a/studio/frontend/src/features/recipe-studio/components/inline/inline-model.tsx b/studio/frontend/src/features/recipe-studio/components/inline/inline-model.tsx index 5316b0f297..2c2c2ffd41 100644 --- a/studio/frontend/src/features/recipe-studio/components/inline/inline-model.tsx +++ b/studio/frontend/src/features/recipe-studio/components/inline/inline-model.tsx @@ -14,19 +14,6 @@ export function InlineModel(props: InlineModelProps): ReactElement { if (props.config.kind === "model_provider") { return (
- - - props.onUpdate({ - // biome-ignore lint/style/useNamingConvention: api schema - provider_type: event.target.value, - }) - } - /> - props.onUpdate({ endpoint: event.target.value })} /> + + + props.onUpdate({ + // biome-ignore lint/style/useNamingConvention: api schema + api_key: event.target.value, + }) + } + /> +
); } diff --git a/studio/frontend/src/features/recipe-studio/components/inline/inline-policy.ts b/studio/frontend/src/features/recipe-studio/components/inline/inline-policy.ts index 97f441edce..17f22b6557 100644 --- a/studio/frontend/src/features/recipe-studio/components/inline/inline-policy.ts +++ b/studio/frontend/src/features/recipe-studio/components/inline/inline-policy.ts @@ -27,6 +27,9 @@ export function getConfigUiMode( } return "dialog"; } + if (config.kind === "seed") { + return "inline"; + } if (config.kind === "expression") { return "inline"; } diff --git a/studio/frontend/src/features/recipe-studio/components/inline/inline-seed.tsx b/studio/frontend/src/features/recipe-studio/components/inline/inline-seed.tsx new file mode 100644 index 0000000000..619faa7704 --- /dev/null +++ b/studio/frontend/src/features/recipe-studio/components/inline/inline-seed.tsx @@ -0,0 +1,66 @@ +import { DocumentAttachmentIcon, DocumentCodeIcon, Plant01Icon } from "@hugeicons/core-free-icons"; +import { HugeiconsIcon } from "@hugeicons/react"; +import type { ReactElement } from "react"; +import type { SeedConfig } from "../../types"; +import { HfDatasetCombobox } from "../shared/hf-dataset-combobox"; +import { InlineField } from "./inline-field"; + +type InlineSeedProps = { + config: SeedConfig; + onUpdate: (patch: Partial) => void; +}; + +export function InlineSeed({ config, onUpdate }: InlineSeedProps): ReactElement { + const mode = config.seed_source_type ?? "hf"; + + if (mode === "hf") { + return ( +
+ + + onUpdate({ + hf_repo_id: next, + hf_path: "", + seed_columns: [], + seed_drop_columns: [], + seed_preview_rows: [], + }) + } + placeholder="org/repo" + /> + +

+ Load columns in dialog. +

+
+ ); + } + + const isLocal = mode === "local"; + const fileName = isLocal + ? config.local_file_name?.trim() + : config.unstructured_file_name?.trim(); + + return ( +
+
+ +
+
+

+ {fileName || "No file selected"} +

+

+ {isLocal ? "Structured file" : "Unstructured document"} · configure in dialog +

+
+ +
+ ); +} diff --git a/studio/frontend/src/features/recipe-studio/components/recipe-graph-aux-node.tsx b/studio/frontend/src/features/recipe-studio/components/recipe-graph-aux-node.tsx index c7f14ebbc4..bbb5dead41 100644 --- a/studio/frontend/src/features/recipe-studio/components/recipe-graph-aux-node.tsx +++ b/studio/frontend/src/features/recipe-studio/components/recipe-graph-aux-node.tsx @@ -1,24 +1,21 @@ -import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Input } from "@/components/ui/input"; import { Textarea } from "@/components/ui/textarea"; import { Handle, - NodeResizer, + Position, type Node, type NodeProps, useUpdateNodeInternals, } from "@xyflow/react"; import { memo, type ReactElement, useEffect } from "react"; -import { MAX_NODE_WIDTH, MIN_NODE_WIDTH } from "../constants"; import { useRecipeStudioStore } from "../stores/recipe-studio"; -import type { LayoutDirection, LlmConfig, Score, ScoreOption } from "../types"; -import { - AUX_HANDLE_CLASS, - getAuxSourceHandlePosition, -} from "../utils/handle-layout"; +import type { LlmConfig, Score, ScoreOption } from "../types"; +import { AUX_HANDLE_CLASS } from "../utils/handle-layout"; import { HANDLE_IDS } from "../utils/handles"; +import { findInvalidJinjaReferences } from "../utils/refs"; import { getAvailableVariableEntries } from "../utils/variables"; +import { AvailableReferencesInline } from "./shared/available-references-inline"; import { BaseNode, BaseNodeContent, BaseNodeHeader, BaseNodeHeaderTitle } from "./rf-ui/base-node"; type PromptField = "prompt" | "system_prompt"; @@ -28,14 +25,12 @@ type PromptInputNodeData = { llmId: string; field: PromptField; title: string; - layoutDirection: LayoutDirection; }; type JudgeScoreNodeData = { kind: "llm-judge-score"; llmId: string; scoreIndex: number; - layoutDirection: LayoutDirection; }; export type RecipeGraphAuxNodeData = PromptInputNodeData | JudgeScoreNodeData; @@ -62,30 +57,12 @@ function updateOptionAt( ); } -function AuxVariableBadges({ llmId }: { llmId: string }): ReactElement | null { - const configs = useRecipeStudioStore((state) => state.configs); - const vars = getAvailableVariableEntries(configs, llmId); - if (vars.length === 0) return null; - return ( -
-

Available references

-
- {vars.map((v) => ( - - {v.name} - - ))} -
-
- ); +function AuxVariableBadges({ + entries, +}: { + entries: ReturnType; +}): ReactElement | null { + return ; } function AuxNodeBase({ @@ -93,6 +70,7 @@ function AuxNodeBase({ data, }: NodeProps): ReactElement | null { const config = useRecipeStudioStore((state) => state.configs[data.llmId]); + const configs = useRecipeStudioStore((state) => state.configs); const updateConfig = useRecipeStudioStore((state) => state.updateConfig); const updateNodeInternals = useUpdateNodeInternals(); @@ -104,30 +82,58 @@ function AuxNodeBase({ return null; } - const sourcePosition = getAuxSourceHandlePosition(data.layoutDirection); + const sourceHandles = ( + <> + + + + + + ); if (data.kind === "llm-prompt-input") { const value = data.field === "prompt" ? config.prompt : config.system_prompt; + const variableEntries = getAvailableVariableEntries(configs, data.llmId); + const availableRefs = variableEntries.map((entry) => entry.name); + const hasInvalidRefs = + findInvalidJinjaReferences(value, availableRefs).length > 0; return ( - {data.title}