diff --git a/studio/backend/core/data_recipe/jobs/manager.py b/studio/backend/core/data_recipe/jobs/manager.py index 6f86196bcc..4170f28674 100644 --- a/studio/backend/core/data_recipe/jobs/manager.py +++ b/studio/backend/core/data_recipe/jobs/manager.py @@ -6,6 +6,7 @@ import queue import threading import time import uuid +from pathlib import Path from collections import deque from dataclasses import dataclass from typing import Any @@ -19,6 +20,29 @@ 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: @@ -144,6 +168,9 @@ class JobManager: "rows": job.rows, "cols": job.cols, "error": job.error, + "has_analysis": job.analysis is not None, + "dataset_rows": None if job.dataset is None else len(job.dataset), + "artifact_path": job.artifact_path, "started_at": job.started_at, "finished_at": job.finished_at, } @@ -167,6 +194,36 @@ class JobManager: return None return self._job.analysis + def get_dataset(self, job_id: str, *, limit: int) -> list[dict[str, Any]] | None: + """Load job dataset rows for UI previews (limited head).""" + with self._lock: + if self._job is None or self._job.job_id != job_id: + return None + in_memory_dataset = self._job.dataset + artifact_path = self._job.artifact_path + + if in_memory_dataset is not None: + return in_memory_dataset[:limit] + if not artifact_path: + return None + + try: + from data_designer.engine.dataset_builders.artifact_storage import ArtifactStorage + except Exception: + return None + + try: + base_dataset_path = Path(artifact_path) + storage = ArtifactStorage( + artifact_path=str(base_dataset_path.parent), + dataset_name=base_dataset_path.name, + ) + dataframe = storage.load_dataset() + rows = dataframe.head(limit).to_dict(orient="records") + return _to_jsonable(rows) + except Exception: + return None + def subscribe(self, job_id: str, *, after_seq: int | None = None) -> Subscription | None: """SSE subscribe: get replay buffer + live events stream.""" with self._lock: @@ -279,6 +336,8 @@ class JobManager: self._job.finished_at = time.time() self._job.analysis = event.get("analysis") self._job.artifact_path = event.get("artifact_path") + self._job.dataset = event.get("dataset") + self._job.processor_artifacts = event.get("processor_artifacts") if et == "job.error": self._job.status = "error" self._job.finished_at = time.time() diff --git a/studio/backend/core/data_recipe/jobs/types.py b/studio/backend/core/data_recipe/jobs/types.py index 8e203f2add..d475b928e1 100644 --- a/studio/backend/core/data_recipe/jobs/types.py +++ b/studio/backend/core/data_recipe/jobs/types.py @@ -61,7 +61,8 @@ class Job: analysis: dict[str, Any] | None = None artifact_path: str | None = None + dataset: list[dict[str, Any]] | None = None + processor_artifacts: dict[str, Any] | None = None model_usage: dict[str, ModelUsage] = field(default_factory=dict) _current_usage_model: str | None = None _in_usage_summary: bool = False - diff --git a/studio/backend/core/data_recipe/jobs/worker.py b/studio/backend/core/data_recipe/jobs/worker.py index f13e3a8b73..d3c0da79a0 100644 --- a/studio/backend/core/data_recipe/jobs/worker.py +++ b/studio/backend/core/data_recipe/jobs/worker.py @@ -27,6 +27,32 @@ class _QueueLogHandler(logging.Handler): 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, @@ -63,19 +89,48 @@ def run_job_process( if run_config_raw: designer.set_run_config(RunConfig.model_validate(run_config_raw)) - results = designer.create(builder, num_records=rows, dataset_name=dataset_name) - - analysis = results.load_analysis().model_dump(mode="json") - artifact_path = str(results.artifact_storage.base_dataset_path) - - event_queue.put( - { - "type": "job.completed", - "ts": time.time(), - "analysis": analysis, - "artifact_path": artifact_path, - } - ) + execution_type = str(run.get("execution_type") or "full").strip().lower() + if execution_type == "preview": + results = designer.preview(builder, num_records=rows) + analysis = ( + None + if results.analysis is None + else _to_jsonable(results.analysis.model_dump(mode="json")) + ) + dataset = ( + [] + if results.dataset is None + else _to_jsonable(results.dataset.to_dict(orient="records")) + ) + processor_artifacts = ( + None + if results.processor_artifacts is None + else _to_jsonable(results.processor_artifacts) + ) + event_queue.put( + { + "type": "job.completed", + "ts": time.time(), + "analysis": analysis, + "dataset": dataset, + "processor_artifacts": processor_artifacts, + "artifact_path": None, + "execution_type": execution_type, + } + ) + else: + results = designer.create(builder, num_records=rows, dataset_name=dataset_name) + analysis = _to_jsonable(results.load_analysis().model_dump(mode="json")) + artifact_path = str(results.artifact_storage.base_dataset_path) + event_queue.put( + { + "type": "job.completed", + "ts": time.time(), + "analysis": analysis, + "artifact_path": artifact_path, + "execution_type": execution_type, + } + ) except Exception as exc: event_queue.put( { diff --git a/studio/backend/routes/data_recipe.py b/studio/backend/routes/data_recipe.py index fd73022a9f..825552d65a 100644 --- a/studio/backend/routes/data_recipe.py +++ b/studio/backend/routes/data_recipe.py @@ -8,7 +8,7 @@ import sys from pathlib import Path from typing import Any -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import JSONResponse, StreamingResponse # same thing as other files do @@ -129,6 +129,15 @@ def job_analysis(job_id: str): return analysis +@router.get("/jobs/{job_id}/dataset") +def job_dataset(job_id: str, limit: int = Query(default=20, ge=1, le=500)): + mgr = get_job_manager() + dataset = mgr.get_dataset(job_id, limit=limit) + if dataset is None: + raise HTTPException(status_code=404, detail="dataset not ready") + return {"dataset": dataset} + + @router.get("/jobs/{job_id}/events") async def job_events(request: Request, job_id: str): mgr = get_job_manager() diff --git a/studio/frontend/src/features/recipe-studio/api/index.ts b/studio/frontend/src/features/recipe-studio/api/index.ts index 10087b33be..f75901e7a2 100644 --- a/studio/frontend/src/features/recipe-studio/api/index.ts +++ b/studio/frontend/src/features/recipe-studio/api/index.ts @@ -10,6 +10,59 @@ export type PreviewResponse = { analysis?: Record; }; +export type JobCreateResponse = { + // biome-ignore lint/style/useNamingConvention: api schema + job_id: string; +}; + +export type JobStatusResponse = { + // biome-ignore lint/style/useNamingConvention: api schema + job_id: string; + status: string; + stage?: string | null; + // biome-ignore lint/style/useNamingConvention: api schema + current_column?: string | null; + batch?: { + idx?: number | null; + total?: number | null; + }; + progress?: { + done?: number | null; + total?: number | null; + percent?: number | null; + // biome-ignore lint/style/useNamingConvention: api schema + eta_sec?: number | null; + rate?: number | null; + ok?: number | null; + failed?: number | null; + }; + // biome-ignore lint/style/useNamingConvention: api schema + model_usage?: Record; + rows?: number | null; + cols?: number | null; + error?: string | null; + // biome-ignore lint/style/useNamingConvention: api schema + has_analysis?: boolean; + // biome-ignore lint/style/useNamingConvention: api schema + dataset_rows?: number | null; + // biome-ignore lint/style/useNamingConvention: api schema + artifact_path?: string | null; + // biome-ignore lint/style/useNamingConvention: api schema + started_at?: number | null; + // biome-ignore lint/style/useNamingConvention: api schema + finished_at?: number | null; +}; + +export type JobDatasetResponse = { + dataset?: unknown[]; +}; + +export type JobEvent = { + event: string; + id: number | null; + payload: Record; +}; + export type ValidateError = { message: string; path?: string | null; @@ -62,6 +115,49 @@ async function postJson(path: string, payload: unknown): Promise { return response.json(); } +async function getJson(path: string): Promise { + const response = await fetch(`${DATA_DESIGNER_API_BASE}${path}`); + if (!response.ok) { + throw new Error(await parseErrorResponse(response)); + } + return response.json(); +} + +function parseJobEvent(rawEvent: string): JobEvent | null { + const lines = rawEvent.split(/\r?\n/); + let eventName = "message"; + let id: number | null = null; + const dataLines: string[] = []; + + for (const line of lines) { + if (!line) { + continue; + } + if (line.startsWith("event:")) { + eventName = line.slice(6).trim() || "message"; + continue; + } + if (line.startsWith("id:")) { + const value = Number(line.slice(3).trim()); + id = Number.isFinite(value) ? value : null; + continue; + } + if (line.startsWith("data:")) { + dataLines.push(line.slice(5).trimStart()); + } + } + + if (dataLines.length === 0) { + return null; + } + const payload = JSON.parse(dataLines.join("\n")) as Record; + return { + event: eventName, + id, + payload, + }; +} + export async function previewRecipe(payload: unknown): Promise { return postJson("/preview", payload); } @@ -72,4 +168,90 @@ export async function validateRecipe( return postJson("/validate", payload); } +export async function createRecipeJob(payload: unknown): Promise { + return postJson("/jobs", payload); +} + +export async function getRecipeJobStatus(jobId: string): Promise { + return getJson(`/jobs/${jobId}/status`); +} + +export async function getRecipeJobAnalysis( + jobId: string, +): Promise> { + return getJson>(`/jobs/${jobId}/analysis`); +} + +export async function getRecipeJobDataset( + jobId: string, + limit = 20, +): Promise { + return getJson(`/jobs/${jobId}/dataset?limit=${limit}`); +} + +export async function cancelRecipeJob(jobId: string): Promise { + return postJson(`/jobs/${jobId}/cancel`, {}); +} + +export async function streamRecipeJobEvents(options: { + jobId: string; + signal: AbortSignal; + lastEventId?: number | null; + onOpen?: () => void; + onEvent: (event: JobEvent) => void; +}): Promise { + const headers = new Headers(); + let query = ""; + if (typeof options.lastEventId === "number") { + headers.set("Last-Event-ID", String(options.lastEventId)); + query = `?after=${options.lastEventId}`; + } + + const response = await fetch( + `${DATA_DESIGNER_API_BASE}/jobs/${options.jobId}/events${query}`, + { + method: "GET", + headers, + signal: options.signal, + }, + ); + if (!response.ok) { + throw new Error(await parseErrorResponse(response)); + } + if (!response.body) { + throw new Error("Job stream unavailable."); + } + + options.onOpen?.(); + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + while (true) { + const { value, done } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let separatorIndex = buffer.search(/\r?\n\r?\n/); + while (separatorIndex >= 0) { + const rawEvent = buffer.slice(0, separatorIndex); + const separatorLength = buffer[separatorIndex] === "\r" ? 4 : 2; + buffer = buffer.slice(separatorIndex + separatorLength); + + if (rawEvent.startsWith("retry:")) { + separatorIndex = buffer.search(/\r?\n\r?\n/); + continue; + } + + const parsed = parseJobEvent(rawEvent); + if (parsed) { + options.onEvent(parsed); + } + separatorIndex = buffer.search(/\r?\n\r?\n/); + } + } +} + // NOTE: tools + seed inspect/preview endpoints removed from harness. diff --git a/studio/frontend/src/features/recipe-studio/components/executions/executions-view.tsx b/studio/frontend/src/features/recipe-studio/components/executions/executions-view.tsx index 51dc812575..03fc805a57 100644 --- a/studio/frontend/src/features/recipe-studio/components/executions/executions-view.tsx +++ b/studio/frontend/src/features/recipe-studio/components/executions/executions-view.tsx @@ -1,19 +1,44 @@ -import { useMemo, type ReactElement } from "react"; +import { useEffect, useMemo, useState, type ReactElement } from "react"; import type { ColumnDef } from "@tanstack/react-table"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { DataTable } from "@/components/ui/data-table"; +import { Progress } from "@/components/ui/progress"; import { Skeleton } from "@/components/ui/skeleton"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table"; +import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { cn } from "@/lib/utils"; -import type { RecipeExecutionRecord } from "../../execution-types"; +import type { + RecipeExecutionAnalysis, + RecipeExecutionRecord, + RecipeExecutionStatus, +} from "../../execution-types"; type ExecutionsViewProps = { executions: RecipeExecutionRecord[]; selectedExecutionId: string | null; currentSignature: string; previewLoading: boolean; + fullLoading: boolean; onSelectExecution: (id: string) => void; onRunPreview: () => void; + onRunFull: () => void; + onCancelExecution: (id: string) => void; +}; + +type AnalysisColumnStat = { + column_name: string; + column_type: string; + simple_dtype: string; + num_unique: number | null; + num_null: number | null; }; function formatTimestamp(value: number): string { @@ -37,14 +62,90 @@ function formatCellValue(value: unknown): string { } } -function statusTone(status: RecipeExecutionRecord["status"]): string { +function parseNumber(value: unknown): number | null { + return typeof value === "number" && Number.isFinite(value) ? value : null; +} + +function parseString(value: unknown): string { + return typeof value === "string" && value.length > 0 ? value : "--"; +} + +function parseAnalysisColumns(analysis: RecipeExecutionAnalysis | null): AnalysisColumnStat[] { + const items = Array.isArray(analysis?.column_statistics) + ? analysis.column_statistics + : []; + return items + .map((item) => { + if (!item || typeof item !== "object" || Array.isArray(item)) { + return null; + } + const row = item as Record; + return { + column_name: parseString(row.column_name), + column_type: parseString(row.column_type), + simple_dtype: parseString(row.simple_dtype), + num_unique: parseNumber(row.num_unique), + num_null: parseNumber(row.num_null), + }; + }) + .filter((item): item is AnalysisColumnStat => item !== null); +} + +function isInProgress(status: RecipeExecutionStatus): boolean { + return ( + status === "running" || + status === "active" || + status === "pending" || + status === "cancelling" + ); +} + +function statusTone(status: RecipeExecutionStatus): string { if (status === "completed") { return "bg-emerald-100 text-emerald-700"; } - if (status === "error") { + if (status === "error" || status === "cancelled") { return "bg-red-100 text-red-700"; } - return "bg-amber-100 text-amber-700"; + if (isInProgress(status)) { + return "bg-amber-100 text-amber-700"; + } + return "bg-muted text-muted-foreground"; +} + +function statusRightBorder(status: RecipeExecutionStatus): string { + if (status === "completed") { + return "border-r-emerald-500"; + } + if (status === "error" || status === "cancelled") { + return "border-r-red-500"; + } + if (isInProgress(status)) { + return "border-r-amber-500"; + } + return "border-r-border"; +} + +function formatStatus(status: RecipeExecutionStatus): string { + if (status === "cancelled") { + return "cancelled"; + } + return status; +} + +function formatPercent(value: number | null | undefined): string { + if (typeof value !== "number" || Number.isNaN(value)) { + return "--"; + } + return `${value.toFixed(1)}%`; +} + +function formatDuration(startedAt: number, finishedAt: number | null): string { + if (!finishedAt || finishedAt <= startedAt) { + return "--"; + } + const seconds = Math.round((finishedAt - startedAt) / 1000); + return `${seconds}s`; } export function ExecutionsView({ @@ -52,9 +153,14 @@ export function ExecutionsView({ selectedExecutionId, currentSignature, previewLoading, + fullLoading, onSelectExecution, onRunPreview, + onRunFull, + onCancelExecution, }: ExecutionsViewProps): ReactElement { + const [detailTab, setDetailTab] = useState("overview"); + const [showRaw, setShowRaw] = useState(false); const selectedExecution = useMemo( () => executions.find((execution) => execution.id === selectedExecutionId) ?? @@ -67,6 +173,12 @@ export function ExecutionsView({ selectedExecution.recipeSignature !== currentSignature, ); + useEffect(() => { + if (!showRaw && detailTab === "raw") { + setDetailTab("overview"); + } + }, [detailTab, showRaw]); + const tableColumns = useMemo>[]>(() => { if (!selectedExecution) { return []; @@ -91,6 +203,28 @@ export function ExecutionsView({ })); }, [selectedExecution]); + const analysisColumns = useMemo( + () => parseAnalysisColumns(selectedExecution?.analysis ?? null), + [selectedExecution?.analysis], + ); + const columnTypeCounts = useMemo(() => { + const map = new Map(); + for (const column of analysisColumns) { + map.set(column.column_type, (map.get(column.column_type) ?? 0) + 1); + } + return Array.from(map.entries()); + }, [analysisColumns]); + const sideEffects = useMemo(() => { + const values = selectedExecution?.analysis?.side_effect_column_names; + return Array.isArray(values) + ? values.filter((value): value is string => typeof value === "string") + : []; + }, [selectedExecution?.analysis?.side_effect_column_names]); + + const canCancel = Boolean( + selectedExecution?.jobId && isInProgress(selectedExecution.status), + ); + return (
{executions.length === 0 ? ( @@ -120,10 +264,11 @@ export function ExecutionsView({ type="button" onClick={() => onSelectExecution(execution.id)} className={cn( - "mb-2 w-full rounded-xl corner-squircle border p-3 text-left", + "mb-2 w-full rounded-xl corner-squircle border border-r-4 p-3 text-left", selectedExecutionId === execution.id ? "border-primary/50 bg-primary/5" : "hover:bg-muted/40", + statusRightBorder(execution.status), )} >
@@ -134,7 +279,7 @@ export function ExecutionsView({ variant="secondary" className={cn("capitalize", statusTone(execution.status))} > - {execution.status} + {formatStatus(execution.status)}

@@ -156,26 +301,102 @@ export function ExecutionsView({ ) : (

-
-

- {selectedExecution.kind} execution -

- - {selectedExecution.status} - - {isStale && ( - Recipe changed since this run - )} +
+
+

+ {selectedExecution.kind} execution +

+ + {formatStatus(selectedExecution.status)} + + {isStale && ( + Recipe changed since this run + )} +
+
+ {canCancel && ( + + )} + +

Started {formatTimestamp(selectedExecution.createdAt)} |{" "} - {selectedExecution.rows} rows + {selectedExecution.rows} rows | Duration{" "} + {formatDuration( + selectedExecution.createdAt, + selectedExecution.finishedAt, + )}

+ {selectedExecution.stage && ( +

+ Stage: {selectedExecution.stage} + {selectedExecution.current_column + ? ` | Column: ${selectedExecution.current_column}` + : ""} +

+ )}
+ {isInProgress(selectedExecution.status) && ( +
+
+

+ Run in progress +

+

+ {formatPercent(selectedExecution.progress?.percent)} +

+
+ +
+

+ Done: {selectedExecution.progress?.done ?? "--"} +

+

+ Total: {selectedExecution.progress?.total ?? "--"} +

+

+ Rate: {selectedExecution.progress?.rate ?? "--"} rec/s +

+

+ ETA: {selectedExecution.progress?.eta_sec ?? "--"} s +

+
+
+ )} + + {(selectedExecution.status === "error" || + selectedExecution.status === "cancelled") && ( +
+

+ {selectedExecution.status === "cancelled" + ? "Execution cancelled" + : "Execution failed"} +

+

+ {selectedExecution.error ?? "Unknown error."} +

+
+ )} + {selectedExecution.status === "running" && (
@@ -183,37 +404,137 @@ export function ExecutionsView({
)} - {selectedExecution.status === "error" && ( -
-

Preview failed

-

- {selectedExecution.error ?? "Unknown error."} -

-
- )} - - {selectedExecution.status === "completed" && ( - <> -
-

Analysis (full)

- 
-                    {JSON.stringify(selectedExecution.analysis ?? {}, null, 2)}
-
-
-
-

Preview data

- {selectedExecution.dataset.length === 0 ? ( -

No rows returned.

- ) : ( -
- + {(selectedExecution.status === "completed" || + isInProgress(selectedExecution.status)) && ( + + + Overview + Columns + Data + {showRaw && Raw} + + +
+
+

Records

+

+ {selectedExecution.analysis?.num_records ?? "--"} +

- )} -
- +
+

Target

+

+ {selectedExecution.analysis?.target_num_records ?? "--"} +

+
+
+

Completion

+

+ {formatPercent( + selectedExecution.analysis?.num_records && + selectedExecution.analysis?.target_num_records + ? (selectedExecution.analysis.num_records / + selectedExecution.analysis.target_num_records) * + 100 + : null, + )} +

+
+
+

Columns

+

+ {analysisColumns.length > 0 ? analysisColumns.length : "--"} +

+
+
+
+

Column type breakdown

+ {columnTypeCounts.length === 0 ? ( +

No analysis yet.

+ ) : ( +
+ {columnTypeCounts.map(([type, count]) => ( + + {type}: {count} + + ))} +
+ )} +
+
+

Side-effect columns

+ {sideEffects.length === 0 ? ( +

None.

+ ) : ( +
+ {sideEffects.map((name) => ( + + {name} + + ))} +
+ )} +
+ + +
+

Column statistics

+ {analysisColumns.length === 0 ? ( +

+ No column statistics yet. +

+ ) : ( + + + + Column + Type + Data type + Unique + Nulls + + + + {analysisColumns.map((column) => ( + + {column.column_name} + {column.column_type} + {column.simple_dtype} + {column.num_unique ?? "--"} + {column.num_null ?? "--"} + + ))} + +
+ )} +
+ + +
+

Dataset sample

+ {selectedExecution.dataset.length === 0 ? ( +

No rows returned.

+ ) : ( +
+ +
+ )} +
+ + {showRaw && ( + +
+

Raw execution

+ 
+                        {JSON.stringify(selectedExecution, null, 2)}
+
+
+ + )} + )}
)} diff --git a/studio/frontend/src/features/recipe-studio/data/executions-db.ts b/studio/frontend/src/features/recipe-studio/data/executions-db.ts index b0b3fb104d..e09824a664 100644 --- a/studio/frontend/src/features/recipe-studio/data/executions-db.ts +++ b/studio/frontend/src/features/recipe-studio/data/executions-db.ts @@ -9,6 +9,10 @@ db.version(1).stores({ executions: "id, recipeId, kind, status, createdAt", }); +db.version(2).stores({ + executions: "id, recipeId, kind, status, createdAt, finishedAt, jobId", +}); + export async function listRecipeExecutions( recipeId: string, ): Promise { diff --git a/studio/frontend/src/features/recipe-studio/execution-types.ts b/studio/frontend/src/features/recipe-studio/execution-types.ts index 245488bbd1..bba9ec96ad 100644 --- a/studio/frontend/src/features/recipe-studio/execution-types.ts +++ b/studio/frontend/src/features/recipe-studio/execution-types.ts @@ -2,18 +2,60 @@ export type RecipeStudioView = "editor" | "executions"; export type RecipeExecutionKind = "preview" | "full"; -export type RecipeExecutionStatus = "running" | "completed" | "error"; +export type RecipeExecutionStatus = + | "pending" + | "running" + | "active" + | "cancelling" + | "cancelled" + | "completed" + | "error"; + +export type RecipeExecutionProgress = { + done?: number | null; + total?: number | null; + percent?: number | null; + // biome-ignore lint/style/useNamingConvention: backend schema + eta_sec?: number | null; + rate?: number | null; + ok?: number | null; + failed?: number | null; +}; + +export type RecipeExecutionAnalysis = { + num_records?: number; + target_num_records?: number; + // biome-ignore lint/style/useNamingConvention: backend schema + column_statistics?: Record[]; + // biome-ignore lint/style/useNamingConvention: backend schema + side_effect_column_names?: string[] | null; + // biome-ignore lint/style/useNamingConvention: backend schema + column_profiles?: Record[] | null; +} & Record; export type RecipeExecutionRecord = { id: string; recipeId: string; + // biome-ignore lint/style/useNamingConvention: backend schema + jobId: string | null; kind: RecipeExecutionKind; status: RecipeExecutionStatus; rows: number; createdAt: number; + finishedAt: number | null; recipeSignature: string; + stage: string | null; + // biome-ignore lint/style/useNamingConvention: backend schema + current_column: string | null; + progress: RecipeExecutionProgress | null; + // biome-ignore lint/style/useNamingConvention: backend schema + model_usage: Record | null; + // biome-ignore lint/style/useNamingConvention: backend schema + lastEventId: number | null; + // biome-ignore lint/style/useNamingConvention: backend schema + artifact_path: string | null; dataset: Record[]; - analysis: Record | null; + analysis: RecipeExecutionAnalysis | null; // biome-ignore lint/style/useNamingConvention: api schema processor_artifacts: Record | null; error: string | null; diff --git a/studio/frontend/src/features/recipe-studio/hooks/use-recipe-studio-actions.ts b/studio/frontend/src/features/recipe-studio/hooks/use-recipe-studio-actions.ts index 72eadde4f1..0cad7fe2e5 100644 --- a/studio/frontend/src/features/recipe-studio/hooks/use-recipe-studio-actions.ts +++ b/studio/frontend/src/features/recipe-studio/hooks/use-recipe-studio-actions.ts @@ -1,9 +1,21 @@ import { useCallback, useEffect, useMemo, useState } from "react"; import { toastError, toastSuccess } from "@/shared/toast"; import { normalizeNonEmptyName } from "@/utils"; -import { previewRecipe, validateRecipe } from "../api"; +import { + cancelRecipeJob, + createRecipeJob, + getRecipeJobAnalysis, + getRecipeJobDataset, + getRecipeJobStatus, + previewRecipe, + validateRecipe, +} from "../api"; import { listRecipeExecutions, saveRecipeExecution } from "../data/executions-db"; -import type { RecipeExecutionRecord } from "../execution-types"; +import type { + RecipeExecutionAnalysis, + RecipeExecutionRecord, + RecipeExecutionStatus, +} from "../execution-types"; import { importRecipePayload, type RecipeSnapshot } from "../utils/import"; import type { RecipePayload, RecipePayloadResult } from "../utils/payload/types"; @@ -28,6 +40,7 @@ type UseRecipeStudioActionsParams = { resetRecipe: () => void; loadRecipe: (snapshot: RecipeSnapshot) => void; getCurrentPayloadFromStore: () => RecipePayload; + onExecutionStart?: () => void; onPreviewSuccess?: () => void; }; @@ -46,6 +59,7 @@ type UseRecipeStudioActionsResult = { setPreviewRows: (rows: number) => void; previewErrors: string[]; previewLoading: boolean; + fullLoading: boolean; currentSignature: string; executions: RecipeExecutionRecord[]; selectedExecutionId: string | null; @@ -53,6 +67,8 @@ type UseRecipeStudioActionsResult = { persistRecipe: () => Promise; openPreviewDialog: () => void; runPreview: () => Promise; + runFull: () => Promise; + cancelExecution: (id: string) => Promise; copyRecipe: () => Promise; importRecipe: (value: string) => string | null; }; @@ -96,6 +112,64 @@ function normalizeObject(value: unknown): Record | null { return value as Record; } +function normalizeAnalysis(value: unknown): RecipeExecutionAnalysis | null { + const normalized = normalizeObject(value); + if (!normalized) { + return null; + } + return normalized as RecipeExecutionAnalysis; +} + +function mapJobStatus(status: string): RecipeExecutionStatus { + if (status === "active") { + return "active"; + } + if (status === "pending") { + return "pending"; + } + if (status === "cancelling") { + return "cancelling"; + } + if (status === "cancelled") { + return "cancelled"; + } + if (status === "completed") { + return "completed"; + } + if (status === "error") { + return "error"; + } + return "running"; +} + +function executionSortWeight(status: RecipeExecutionStatus): number { + if (status === "running" || status === "active" || status === "pending" || status === "cancelling") { + return 0; + } + if (status === "error" || status === "cancelled") { + return 2; + } + return 1; +} + +function sortExecutions(records: RecipeExecutionRecord[]): RecipeExecutionRecord[] { + const next = [...records]; + next.sort((a, b) => { + const statusDelta = executionSortWeight(a.status) - executionSortWeight(b.status); + if (statusDelta !== 0) { + return statusDelta; + } + return b.createdAt - a.createdAt; + }); + return next; +} + +function delay(ms: number): Promise { + return new Promise((resolve) => { + window.setTimeout(resolve, ms); + }); +} + async function copyTextToClipboard(text: string): Promise { try { if (navigator.clipboard?.writeText) { @@ -133,6 +207,7 @@ export function useRecipeStudioActions({ resetRecipe, loadRecipe, getCurrentPayloadFromStore, + onExecutionStart, onPreviewSuccess, }: UseRecipeStudioActionsParams): UseRecipeStudioActionsResult { const [workflowName, setWorkflowName] = useState("Unnamed"); @@ -145,6 +220,7 @@ export function useRecipeStudioActions({ const [previewRows, setPreviewRows] = useState(5); const [previewErrors, setPreviewErrors] = useState([]); const [previewLoading, setPreviewLoading] = useState(false); + const [fullLoading, setFullLoading] = useState(false); const [executions, setExecutions] = useState([]); const [selectedExecutionId, setSelectedExecutionId] = useState( null, @@ -204,8 +280,9 @@ export function useRecipeStudioActions({ if (cancelled) { return; } - setExecutions(records); - setSelectedExecutionId(records[0]?.id ?? null); + const sortedRecords = sortExecutions(records); + setExecutions(sortedRecords); + setSelectedExecutionId(sortedRecords[0]?.id ?? null); } catch (error) { console.error("Load recipe executions failed:", error); } @@ -220,9 +297,8 @@ export function useRecipeStudioActions({ const upsertExecution = useCallback((record: RecipeExecutionRecord): void => { setExecutions((current) => { - const next = current.filter((item) => item.id !== record.id); - next.unshift(record); - return next; + const withoutCurrent = current.filter((item) => item.id !== record.id); + return sortExecutions([record, ...withoutCurrent]); }); setSelectedExecutionId(record.id); void saveRecipeExecution(record).catch((error) => { @@ -290,17 +366,27 @@ export function useRecipeStudioActions({ const baseExecution: RecipeExecutionRecord = { id: crypto.randomUUID(), recipeId, + jobId: null, kind: "preview", status: "running", rows: previewRows, createdAt, + finishedAt: null, recipeSignature: currentSignature, + stage: "preview", + current_column: null, + progress: null, + model_usage: null, + lastEventId: null, + artifact_path: null, dataset: [], analysis: null, processor_artifacts: null, error: null, }; upsertExecution(baseExecution); + onExecutionStart?.(); + setPreviewDialogOpen(false); const previewPayload = { ...payload, @@ -330,12 +416,12 @@ export function useRecipeStudioActions({ upsertExecution({ ...baseExecution, status: "completed", + finishedAt: Date.now(), dataset: normalizeDatasetRows(result.dataset), - analysis: normalizeObject(result.analysis), + analysis: normalizeAnalysis(result.analysis), processor_artifacts: normalizeObject(result.processor_artifacts), error: null, }); - setPreviewDialogOpen(false); setPreviewErrors([]); toastSuccess(`Preview generated (${previewRows} rows).`); onPreviewSuccess?.(); @@ -346,6 +432,7 @@ export function useRecipeStudioActions({ upsertExecution({ ...baseExecution, status: "error", + finishedAt: Date.now(), error: message, }); setPreviewErrors([message]); @@ -356,6 +443,7 @@ export function useRecipeStudioActions({ } }, [ currentSignature, + onExecutionStart, onPreviewSuccess, payloadErrorMessage, payloadResult.errors, @@ -365,6 +453,183 @@ export function useRecipeStudioActions({ upsertExecution, ]); + const runFull = useCallback(async (): Promise => { + const payload = readPayload(); + if (!payload) { + setPreviewErrors(payloadResult.errors); + toastError("Invalid recipe payload", payloadErrorMessage); + return false; + } + + const requestedRows = Number(payload.run?.rows); + const rows = Number.isFinite(requestedRows) && requestedRows > 0 + ? Math.floor(requestedRows) + : 1000; + const createdAt = Date.now(); + const baseExecution: RecipeExecutionRecord = { + id: crypto.randomUUID(), + recipeId, + jobId: null, + kind: "full", + status: "pending", + rows, + createdAt, + finishedAt: null, + recipeSignature: currentSignature, + stage: "pending", + current_column: null, + progress: null, + model_usage: null, + lastEventId: null, + artifact_path: null, + dataset: [], + analysis: null, + processor_artifacts: null, + error: null, + }; + + upsertExecution(baseExecution); + onExecutionStart?.(); + setFullLoading(true); + + try { + const fullPayload = { + ...payload, + run: { + ...payload.run, + rows, + // biome-ignore lint/style/useNamingConvention: backend schema + execution_type: "full", + }, + }; + const createdJob = await createRecipeJob(fullPayload); + const jobId = createdJob.job_id; + let done = false; + let lastStatus: RecipeExecutionStatus = "pending"; + let latestExecution: RecipeExecutionRecord = { + ...baseExecution, + jobId, + }; + upsertExecution(latestExecution); + + while (!done) { + const status = await getRecipeJobStatus(jobId); + const mappedStatus = mapJobStatus(status.status); + lastStatus = mappedStatus; + + latestExecution = { + ...latestExecution, + status: mappedStatus, + rows: status.rows ?? latestExecution.rows, + stage: status.stage ?? latestExecution.stage, + current_column: status.current_column ?? null, + progress: (normalizeObject(status.progress) as RecipeExecutionRecord["progress"]) ?? null, + model_usage: normalizeObject(status.model_usage), + artifact_path: status.artifact_path ?? latestExecution.artifact_path, + error: status.error ?? null, + finishedAt: + mappedStatus === "completed" || + mappedStatus === "error" || + mappedStatus === "cancelled" + ? Date.now() + : null, + }; + upsertExecution(latestExecution); + + done = + mappedStatus === "completed" || + mappedStatus === "error" || + mappedStatus === "cancelled"; + if (!done) { + await delay(1200); + } + } + + if (lastStatus === "completed") { + const [analysisResult, datasetResult] = await Promise.allSettled([ + getRecipeJobAnalysis(jobId), + getRecipeJobDataset(jobId, 20), + ]); + const analysis = + analysisResult.status === "fulfilled" + ? normalizeAnalysis(analysisResult.value) + : latestExecution.analysis; + const dataset = + datasetResult.status === "fulfilled" + ? normalizeDatasetRows(datasetResult.value.dataset) + : latestExecution.dataset; + + upsertExecution({ + ...latestExecution, + status: "completed", + analysis, + dataset, + error: null, + finishedAt: latestExecution.finishedAt ?? Date.now(), + }); + toastSuccess("Full run completed."); + return true; + } + + if (lastStatus === "cancelled") { + upsertExecution({ + ...latestExecution, + status: "cancelled", + error: latestExecution.error ?? "Run cancelled.", + finishedAt: latestExecution.finishedAt ?? Date.now(), + }); + toastError("Full run cancelled", "The execution was cancelled."); + return false; + } + + upsertExecution({ + ...latestExecution, + status: "error", + error: latestExecution.error ?? "Full run failed.", + finishedAt: latestExecution.finishedAt ?? Date.now(), + }); + toastError("Full run failed", latestExecution.error ?? "Execution failed."); + return false; + } catch (error) { + const message = toErrorMessage(error, "Full run request failed."); + upsertExecution({ + ...baseExecution, + status: "error", + error: message, + finishedAt: Date.now(), + }); + toastError("Full run failed", message); + return false; + } finally { + setFullLoading(false); + } + }, [ + currentSignature, + onExecutionStart, + payloadErrorMessage, + payloadResult.errors, + readPayload, + recipeId, + upsertExecution, + ]); + + const cancelExecution = useCallback(async (id: string): Promise => { + const execution = executions.find((entry) => entry.id === id); + if (!execution?.jobId) { + return; + } + try { + await cancelRecipeJob(execution.jobId); + upsertExecution({ + ...execution, + status: "cancelling", + }); + } catch (error) { + const message = toErrorMessage(error, "Could not cancel execution."); + toastError("Cancel failed", message); + } + }, [executions, upsertExecution]); + const selectExecution = useCallback((id: string): void => { setSelectedExecutionId(id); }, []); @@ -418,6 +683,7 @@ export function useRecipeStudioActions({ setPreviewRows, previewErrors, previewLoading, + fullLoading, currentSignature, executions, selectedExecutionId, @@ -425,6 +691,8 @@ export function useRecipeStudioActions({ persistRecipe, openPreviewDialog, runPreview, + runFull, + cancelExecution, copyRecipe, importRecipe, }; diff --git a/studio/frontend/src/features/recipe-studio/recipe-studio-page.tsx b/studio/frontend/src/features/recipe-studio/recipe-studio-page.tsx index c915e0adf8..08c814dc91 100644 --- a/studio/frontend/src/features/recipe-studio/recipe-studio-page.tsx +++ b/studio/frontend/src/features/recipe-studio/recipe-studio-page.tsx @@ -272,6 +272,7 @@ export function RecipeStudioPage({ setPreviewRows, previewErrors, previewLoading, + fullLoading, currentSignature, executions, selectedExecutionId, @@ -279,6 +280,8 @@ export function RecipeStudioPage({ persistRecipe, openPreviewDialog, runPreview, + runFull, + cancelExecution, copyRecipe, importRecipe, } = useRecipeStudioActions({ @@ -291,6 +294,9 @@ export function RecipeStudioPage({ resetRecipe, loadRecipe, getCurrentPayloadFromStore, + onExecutionStart: () => { + setActiveView("executions"); + }, onPreviewSuccess: () => { setActiveView("executions"); }, @@ -390,8 +396,15 @@ export function RecipeStudioPage({ selectedExecutionId={selectedExecutionId} currentSignature={currentSignature} previewLoading={previewLoading} + fullLoading={fullLoading} onSelectExecution={setSelectedExecutionId} onRunPreview={openPreviewDialog} + onRunFull={() => { + void runFull(); + }} + onCancelExecution={(executionId) => { + void cancelExecution(executionId); + }} /> )}