ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add support for full recipe executions with detailed progress and analysis

Browse source 
- Introduced "Full Run" support in execution logic, including progress tracking, cancellation, and job status updates.
- Extended backend to manage full execution jobs, handle dataset previews, and return detailed analysis and artifacts.
- Updated frontend components to support full runs, with execution sorting, live updates, and detailed execution views.
- Enhanced `ExecutionsView` with progress indicators, status filtering, and dataset preview capabilities.
- Added IndexedDB schema migration to track additional execution metadata.
This commit is contained in:
Shine1i 2026-02-20 12:05:42 +01:00
commit 1259b75d15
10 changed files with 1037 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

59
studio/backend/core/data_recipe/jobs/manager.py
View file

@ -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()

3
studio/backend/core/data_recipe/jobs/types.py
View file

@ -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

81
studio/backend/core/data_recipe/jobs/worker.py
View file

@ -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(
{

11
studio/backend/routes/data_recipe.py
View file

@ -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()

182
studio/frontend/src/features/recipe-studio/api/index.ts
View file

@ -10,6 +10,59 @@ export type PreviewResponse = {
analysis?: Record<string, unknown>;
};
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<string, unknown>;
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<string, unknown>;
};
export type ValidateError = {
message: string;
path?: string | null;
@ -62,6 +115,49 @@ async function postJson<T>(path: string, payload: unknown): Promise<T> {
return response.json();
}
async function getJson<T>(path: string): Promise<T> {
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<string, unknown>;
return {
event: eventName,
id,
payload,
};
}
export async function previewRecipe(payload: unknown): Promise<PreviewResponse> {
return postJson<PreviewResponse>("/preview", payload);
}
@ -72,4 +168,90 @@ export async function validateRecipe(
return postJson<ValidateResponse>("/validate", payload);
}
export async function createRecipeJob(payload: unknown): Promise<JobCreateResponse> {
return postJson<JobCreateResponse>("/jobs", payload);
}
export async function getRecipeJobStatus(jobId: string): Promise<JobStatusResponse> {
return getJson<JobStatusResponse>(`/jobs/${jobId}/status`);
}
export async function getRecipeJobAnalysis(
jobId: string,
): Promise<Record<string, unknown>> {
return getJson<Record<string, unknown>>(`/jobs/${jobId}/analysis`);
}
export async function getRecipeJobDataset(
jobId: string,
limit = 20,
): Promise<JobDatasetResponse> {
return getJson<JobDatasetResponse>(`/jobs/${jobId}/dataset?limit=${limit}`);
}
export async function cancelRecipeJob(jobId: string): Promise<JobStatusResponse> {
return postJson<JobStatusResponse>(`/jobs/${jobId}/cancel`, {});
}
export async function streamRecipeJobEvents(options: {
jobId: string;
signal: AbortSignal;
lastEventId?: number | null;
onOpen?: () => void;
onEvent: (event: JobEvent) => void;
}): Promise<void> {
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.

441
studio/frontend/src/features/recipe-studio/components/executions/executions-view.tsx
View file

@ -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<string, unknown>;
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<ColumnDef<Record<string, unknown>>[]>(() => {
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<string, number>();
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 (
<div className="flex h-full min-h-0">
<aside className="w-72 shrink-0 border-r">
@ -98,15 +232,25 @@ export function ExecutionsView({
<p className="text-xs font-semibold uppercase text-muted-foreground">
Executions
</p>
<Button
type="button"
size="sm"
variant="outline"
onClick={onRunPreview}
disabled={previewLoading}
>
{previewLoading ? "Running..." : "Run preview"}
</Button>
<div className="flex items-center gap-1">
<Button
type="button"
size="sm"
variant="outline"
onClick={onRunPreview}
disabled={previewLoading}
>
{previewLoading ? "Running..." : "Preview"}
</Button>
<Button
type="button"
size="sm"
onClick={onRunFull}
disabled={fullLoading}
>
{fullLoading ? "Starting..." : "Full run"}
</Button>
</div>
</div>
<div className="h-[calc(100%-45px)] overflow-auto p-2">
{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),
)}
>
<div className="mb-2 flex items-center justify-between gap-2">
@ -134,7 +279,7 @@ export function ExecutionsView({
variant="secondary"
className={cn("capitalize", statusTone(execution.status))}
>
{execution.status}
{formatStatus(execution.status)}
</Badge>
</div>
<p className="text-xs text-muted-foreground">
@ -156,26 +301,102 @@ export function ExecutionsView({
) : (
<div className="space-y-4">
<div className="rounded-xl border p-3">
<div className="mb-2 flex flex-wrap items-center gap-2">
<p className="text-sm font-semibold capitalize">
{selectedExecution.kind} execution
</p>
<Badge
variant="secondary"
className={cn("capitalize", statusTone(selectedExecution.status))}
>
{selectedExecution.status}
</Badge>
{isStale && (
<Badge variant="outline">Recipe changed since this run</Badge>
)}
<div className="mb-2 flex flex-wrap items-center justify-between gap-2">
<div className="flex flex-wrap items-center gap-2">
<p className="text-sm font-semibold capitalize">
{selectedExecution.kind} execution
</p>
<Badge
variant="secondary"
className={cn("capitalize", statusTone(selectedExecution.status))}
>
{formatStatus(selectedExecution.status)}
</Badge>
{isStale && (
<Badge variant="outline">Recipe changed since this run</Badge>
)}
</div>
<div className="flex items-center gap-2">
{canCancel && (
<Button
type="button"
size="sm"
variant="outline"
onClick={() => onCancelExecution(selectedExecution.id)}
>
Cancel
</Button>
)}
<Button
type="button"
size="sm"
variant="ghost"
onClick={() => setShowRaw((value) => !value)}
>
{showRaw ? "Hide raw" : "View raw"}
</Button>
</div>
</div>
<p className="text-xs text-muted-foreground">
Started {formatTimestamp(selectedExecution.createdAt)} |{" "}
{selectedExecution.rows} rows
{selectedExecution.rows} rows | Duration{" "}
{formatDuration(
selectedExecution.createdAt,
selectedExecution.finishedAt,
)}
</p>
{selectedExecution.stage && (
<p className="mt-1 text-xs text-muted-foreground">
Stage: {selectedExecution.stage}
{selectedExecution.current_column
? ` | Column: ${selectedExecution.current_column}`
: ""}
</p>
)}
</div>
{isInProgress(selectedExecution.status) && (
<div className="space-y-3 rounded-xl border border-amber-200 bg-amber-50/50 p-3">
<div className="flex items-center justify-between">
<p className="text-sm font-semibold text-amber-900">
Run in progress
</p>
<p className="text-xs text-amber-800">
{formatPercent(selectedExecution.progress?.percent)}
</p>
</div>
<Progress value={selectedExecution.progress?.percent ?? 0} />
<div className="grid gap-2 text-xs text-amber-900 md:grid-cols-4">
<p>
Done: {selectedExecution.progress?.done ?? "--"}
</p>
<p>
Total: {selectedExecution.progress?.total ?? "--"}
</p>
<p>
Rate: {selectedExecution.progress?.rate ?? "--"} rec/s
</p>
<p>
ETA: {selectedExecution.progress?.eta_sec ?? "--"} s
</p>
</div>
</div>
)}
{(selectedExecution.status === "error" ||
selectedExecution.status === "cancelled") && (
<div className="rounded-xl border border-destructive/40 bg-destructive/5 p-3">
<p className="text-sm font-semibold text-destructive">
{selectedExecution.status === "cancelled"
? "Execution cancelled"
: "Execution failed"}
</p>
<p className="text-xs text-destructive">
{selectedExecution.error ?? "Unknown error."}
</p>
</div>
)}
{selectedExecution.status === "running" && (
<div className="space-y-2 rounded-xl border p-3">
<Skeleton className="h-5 w-44" />
@ -183,37 +404,137 @@ export function ExecutionsView({
</div>
)}
{selectedExecution.status === "error" && (
<div className="rounded-xl border border-destructive/40 bg-destructive/5 p-3">
<p className="text-sm font-semibold text-destructive">Preview failed</p>
<p className="text-xs text-destructive">
{selectedExecution.error ?? "Unknown error."}
</p>
</div>
)}
{selectedExecution.status === "completed" && (
<>
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Analysis (full)</p>
<pre className="max-h-80 overflow-auto rounded-md bg-muted/40 p-3 text-xs">
{JSON.stringify(selectedExecution.analysis ?? {}, null, 2)}
</pre>
</div>
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Preview data</p>
{selectedExecution.dataset.length === 0 ? (
<p className="text-xs text-muted-foreground">No rows returned.</p>
) : (
<div className="max-h-[55vh] overflow-auto">
<DataTable
columns={tableColumns}
data={selectedExecution.dataset}
/>
{(selectedExecution.status === "completed" ||
isInProgress(selectedExecution.status)) && (
<Tabs value={detailTab} onValueChange={setDetailTab}>
<TabsList>
<TabsTrigger value="overview">Overview</TabsTrigger>
<TabsTrigger value="columns">Columns</TabsTrigger>
<TabsTrigger value="data">Data</TabsTrigger>
{showRaw && <TabsTrigger value="raw">Raw</TabsTrigger>}
</TabsList>
<TabsContent value="overview" className="mt-3 space-y-3">
<div className="grid gap-3 md:grid-cols-4">
<div className="rounded-xl border p-3">
<p className="text-xs text-muted-foreground">Records</p>
<p className="text-lg font-semibold">
{selectedExecution.analysis?.num_records ?? "--"}
</p>
</div>
)}
</div>
</>
<div className="rounded-xl border p-3">
<p className="text-xs text-muted-foreground">Target</p>
<p className="text-lg font-semibold">
{selectedExecution.analysis?.target_num_records ?? "--"}
</p>
</div>
<div className="rounded-xl border p-3">
<p className="text-xs text-muted-foreground">Completion</p>
<p className="text-lg font-semibold">
{formatPercent(
selectedExecution.analysis?.num_records &&
selectedExecution.analysis?.target_num_records
? (selectedExecution.analysis.num_records /
selectedExecution.analysis.target_num_records) *
100
: null,
)}
</p>
</div>
<div className="rounded-xl border p-3">
<p className="text-xs text-muted-foreground">Columns</p>
<p className="text-lg font-semibold">
{analysisColumns.length > 0 ? analysisColumns.length : "--"}
</p>
</div>
</div>
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Column type breakdown</p>
{columnTypeCounts.length === 0 ? (
<p className="text-xs text-muted-foreground">No analysis yet.</p>
) : (
<div className="flex flex-wrap gap-2">
{columnTypeCounts.map(([type, count]) => (
<Badge key={type} variant="secondary">
{type}: {count}
</Badge>
))}
</div>
)}
</div>
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Side-effect columns</p>
{sideEffects.length === 0 ? (
<p className="text-xs text-muted-foreground">None.</p>
) : (
<div className="flex flex-wrap gap-2">
{sideEffects.map((name) => (
<Badge key={name} variant="outline">
{name}
</Badge>
))}
</div>
)}
</div>
</TabsContent>
<TabsContent value="columns" className="mt-3">
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Column statistics</p>
{analysisColumns.length === 0 ? (
<p className="text-xs text-muted-foreground">
No column statistics yet.
</p>
) : (
<Table>
<TableHeader>
<TableRow>
<TableHead>Column</TableHead>
<TableHead>Type</TableHead>
<TableHead>Data type</TableHead>
<TableHead>Unique</TableHead>
<TableHead>Nulls</TableHead>
</TableRow>
</TableHeader>
<TableBody>
{analysisColumns.map((column) => (
<TableRow key={column.column_name}>
<TableCell>{column.column_name}</TableCell>
<TableCell>{column.column_type}</TableCell>
<TableCell>{column.simple_dtype}</TableCell>
<TableCell>{column.num_unique ?? "--"}</TableCell>
<TableCell>{column.num_null ?? "--"}</TableCell>
</TableRow>
))}
</TableBody>
</Table>
)}
</div>
</TabsContent>
<TabsContent value="data" className="mt-3">
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Dataset sample</p>
{selectedExecution.dataset.length === 0 ? (
<p className="text-xs text-muted-foreground">No rows returned.</p>
) : (
<div className="max-h-[55vh] overflow-auto">
<DataTable
columns={tableColumns}
data={selectedExecution.dataset}
/>
</div>
)}
</div>
</TabsContent>
{showRaw && (
<TabsContent value="raw" className="mt-3">
<div className="rounded-xl border p-3">
<p className="mb-2 text-sm font-semibold">Raw execution</p>
<pre className="max-h-96 overflow-auto rounded-md bg-muted/40 p-3 text-xs">
{JSON.stringify(selectedExecution, null, 2)}
</pre>
</div>
</TabsContent>
)}
</Tabs>
)}
</div>
)}

4
studio/frontend/src/features/recipe-studio/data/executions-db.ts
View file

@ -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<RecipeExecutionRecord[]> {

46
studio/frontend/src/features/recipe-studio/execution-types.ts
View file

@ -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<string, unknown>[];
// biome-ignore lint/style/useNamingConvention: backend schema
side_effect_column_names?: string[] | null;
// biome-ignore lint/style/useNamingConvention: backend schema
column_profiles?: Record<string, unknown>[] | null;
} & Record<string, unknown>;
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<string, unknown> | null;
// biome-ignore lint/style/useNamingConvention: backend schema
lastEventId: number | null;
// biome-ignore lint/style/useNamingConvention: backend schema
artifact_path: string | null;
dataset: Record<string, unknown>[];
analysis: Record<string, unknown> | null;
analysis: RecipeExecutionAnalysis | null;
// biome-ignore lint/style/useNamingConvention: api schema
processor_artifacts: Record<string, unknown> | null;
error: string | null;

286
studio/frontend/src/features/recipe-studio/hooks/use-recipe-studio-actions.ts
View file

@ -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<void>;
openPreviewDialog: () => void;
runPreview: () => Promise<boolean>;
runFull: () => Promise<boolean>;
cancelExecution: (id: string) => Promise<void>;
copyRecipe: () => Promise<void>;
importRecipe: (value: string) => string | null;
};
@ -96,6 +112,64 @@ function normalizeObject(value: unknown): Record<string, unknown> | null {
return value as Record<string, unknown>;
}
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<void> {
return new Promise((resolve) => {
window.setTimeout(resolve, ms);
});
}
async function copyTextToClipboard(text: string): Promise<boolean> {
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<string[]>([]);
const [previewLoading, setPreviewLoading] = useState(false);
const [fullLoading, setFullLoading] = useState(false);
const [executions, setExecutions] = useState<RecipeExecutionRecord[]>([]);
const [selectedExecutionId, setSelectedExecutionId] = useState<string | null>(
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<boolean> => {
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<void> => {
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,
};

13
studio/frontend/src/features/recipe-studio/recipe-studio-page.tsx
View file

@ -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);
}}
/>
)}
</div>