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

feat: refactor and extend recipe execution logic

Browse source 
- Extracted shared execution utilities into `execution-helpers.ts` for reusability across features.
- Replaced deprecated `/preview` endpoint and its logic with unified job execution handling.
- Consolidated job execution flows ("Preview" and "Full Run") into shared `runJobExecution` logic.
- Enhanced execution progress tracking with support for column-level progress reporting.
- Added support for handling execution job events and improved error reporting from the backend.
- Updated backend to better manage dataset access errors and provide more informative error messages.
- Cleaned up redundant code in `use-recipe-studio-actions` and streamlined execution APIs.
This commit is contained in:
Shine1i 2026-02-20 12:47:51 +01:00
commit 13e153e448
11 changed files with 516 additions and 361 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -147,6 +147,15 @@ class JobManager:
"ok": job.progress.ok,
"failed": job.progress.failed,
},
"column_progress": {
"done": job.column_progress.done,
"total": job.column_progress.total,
"percent": job.column_progress.percent,
"eta_sec": job.column_progress.eta_sec,
"rate": job.column_progress.rate,
"ok": job.column_progress.ok,
"failed": job.column_progress.failed,
},
"model_usage": {
name: {
"model": usage.model,
@ -207,31 +216,31 @@ class JobManager:
return None
in_memory_dataset = self._job.dataset
artifact_path = self._job.artifact_path
job_status = self._job.status
if in_memory_dataset is not None:
total = len(in_memory_dataset)
rows = in_memory_dataset[offset:offset + limit]
return {"dataset": rows, "total": total}
if not artifact_path:
return None
try:
from data_designer.engine.dataset_builders.artifact_storage import ArtifactStorage
except Exception:
if job_status in {"completed", "error", "cancelled"}:
return {"error": "artifact path missing"}
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()
parquet_dir = base_dataset_path / "parquet-files"
if not parquet_dir.exists():
return {"error": f"dataset path missing: {parquet_dir}"}
from data_designer.config.utils.io_helpers import read_parquet_dataset
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}
except Exception:
return None
except Exception as exc:
return {"error": f"dataset load failed: {exc}"}
def subscribe(self, job_id: str, *, after_seq: int | None = None) -> Subscription | None:
"""SSE subscribe: get replay buffer + live events stream."""
@ -347,6 +356,9 @@ class JobManager:
self._job.artifact_path = event.get("artifact_path")
self._job.dataset = event.get("dataset")
self._job.processor_artifacts = event.get("processor_artifacts")
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":
self._job.status = "error"
self._job.finished_at = time.time()

47
studio/backend/core/data_recipe/jobs/parse.py
View file

@ -132,12 +132,15 @@ 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:
job._seen_generation_columns.append(update.current_column)
if update.rows is not None:
job.rows = update.rows
if update.cols is not None:
job.cols = update.cols
if update.progress is not None:
job.progress = update.progress
job.column_progress = update.progress
job.progress = _compute_overall_progress(job, update.progress)
if update.batch_idx is not None:
job.batch.idx = update.batch_idx
if update.batch_total is not None:
@ -194,6 +197,48 @@ def apply_update(job: Job, update: ParsedUpdate) -> None:
usage.rpm = update.usage_rpm
def _compute_overall_progress(job: Job, column_progress: Progress) -> Progress:
if not job.rows or not job.current_column:
return column_progress
total_rows = max(1, int(job.rows))
total_cols = max(
1,
len(job._seen_generation_columns),
int(job.cols or 0),
)
current_done = 0 if column_progress.done is None else int(column_progress.done)
current_done = max(0, min(current_done, total_rows))
try:
col_index = job._seen_generation_columns.index(job.current_column)
except ValueError:
col_index = max(0, len(job._seen_generation_columns) - 1)
col_index = max(0, min(col_index, total_cols - 1))
total = total_rows * total_cols
done = min(total, (col_index * total_rows) + current_done)
prev_done = int(job.progress.done or 0)
if done < prev_done:
done = prev_done
if done > total:
done = total
percent = (done / total) * 100 if total > 0 else 100.0
prev_percent = float(job.progress.percent or 0.0)
if percent < prev_percent:
percent = prev_percent
return Progress(
done=done,
total=total,
percent=percent,
eta_sec=column_progress.eta_sec,
rate=column_progress.rate,
ok=column_progress.ok,
failed=column_progress.failed,
)
def coerce_event(obj: Any) -> dict:
# worker sends dict already
return obj if isinstance(obj, dict) else {"type": "log", "message": str(obj)}

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

@ -52,6 +52,7 @@ class Job:
stage: str | None = None
current_column: str | None = None
progress: Progress = field(default_factory=Progress)
column_progress: Progress = field(default_factory=Progress)
batch: BatchProgress = field(default_factory=BatchProgress)
rows: int | None = None
cols: int | None = None
@ -66,3 +67,4 @@ class Job:
model_usage: dict[str, ModelUsage] = field(default_factory=dict)
_current_usage_model: str | None = None
_in_usage_summary: bool = False
_seen_generation_columns: list[str] = field(default_factory=list)

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

@ -81,10 +81,14 @@ def run_job_process(
rows = int(run.get("rows") or 1000)
dataset_name = str(run.get("dataset_name") or "dataset")
artifact_path_raw = run.get("artifact_path")
artifact_path = None
if isinstance(artifact_path_raw, str) and artifact_path_raw.strip():
artifact_path = artifact_path_raw.strip()
run_config_raw = run.get("run_config") or {}
builder = build_config_builder(recipe)
designer = create_data_designer(recipe)
designer = create_data_designer(recipe, artifact_path=artifact_path)
if run_config_raw:
designer.set_run_config(RunConfig.model_validate(run_config_raw))

7
studio/backend/core/data_recipe/service.py
View file

@ -110,10 +110,15 @@ def build_config_builder(recipe: dict[str, Any]):
return DataDesignerConfigBuilder.from_config({"data_designer": recipe_core})
def create_data_designer(recipe: dict[str, Any]):
def create_data_designer(
recipe: dict[str, Any],
*,
artifact_path: str | None = None,
):
from data_designer.interface.data_designer import DataDesigner
return DataDesigner(
artifact_path=artifact_path,
model_providers=build_model_providers(recipe),
mcp_providers=build_mcp_providers(recipe),
)

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

@ -17,8 +17,8 @@ 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 preview_recipe, validate_recipe
from models.data_recipe import JobCreateResponse, PreviewResponse, RecipePayload, ValidateError, ValidateResponse
from core.data_recipe.service import validate_recipe
from models.data_recipe import JobCreateResponse, RecipePayload, ValidateError, ValidateResponse
router = APIRouter()
@ -47,25 +47,6 @@ def validate(payload: RecipePayload) -> ValidateResponse:
return ValidateResponse(valid=True)
@router.post("/preview", response_model=PreviewResponse)
def preview(payload: RecipePayload) -> PreviewResponse:
recipe = payload.recipe
if not recipe.get("columns"):
raise HTTPException(status_code=400, detail="Recipe must include columns.")
run = payload.run or {}
num_records = int(run.get("rows") or 5)
try:
dataset, artifacts, analysis = preview_recipe(recipe, num_records)
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return PreviewResponse(dataset=dataset, processor_artifacts=artifacts, analysis=analysis)
@router.post("/jobs", response_class=JSONResponse, response_model=JobCreateResponse)
def create_job(payload: RecipePayload):
recipe = payload.recipe
@ -73,6 +54,10 @@ def create_job(payload: RecipePayload):
raise HTTPException(status_code=400, detail="Recipe must include columns.")
run: dict[str, Any] = payload.run or {}
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:
@ -139,6 +124,8 @@ def job_dataset(
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"],

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

@ -3,13 +3,6 @@ const DEFAULT_BASE = "/api/data-recipe";
export const DATA_DESIGNER_API_BASE =
import.meta.env.VITE_DATA_DESIGNER_API ?? DEFAULT_BASE;
export type PreviewResponse = {
dataset?: unknown[];
// biome-ignore lint/style/useNamingConvention: api schema
processor_artifacts?: Record<string, unknown>;
analysis?: Record<string, unknown>;
};
export type JobCreateResponse = {
// biome-ignore lint/style/useNamingConvention: api schema
job_id: string;
@ -37,6 +30,17 @@ export type JobStatusResponse = {
failed?: number | null;
};
// biome-ignore lint/style/useNamingConvention: api schema
column_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;
@ -161,10 +165,6 @@ function parseJobEvent(rawEvent: string): JobEvent | null {
};
}
export async function previewRecipe(payload: unknown): Promise<PreviewResponse> {
return postJson<PreviewResponse>("/preview", payload);
}
export async function validateRecipe(
payload: unknown,
): Promise<ValidateResponse> {

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

@ -388,6 +388,14 @@ export function ExecutionsView({
ETA: {selectedExecution.progress?.eta_sec ?? "--"} s
</p>
</div>
{selectedExecution.current_column && selectedExecution.column_progress && (
<p className="text-xs text-amber-900">
Column {selectedExecution.current_column}:{" "}
{selectedExecution.column_progress.done ?? "--"}/
{selectedExecution.column_progress.total ?? "--"} (
{formatPercent(selectedExecution.column_progress.percent)})
</p>
)}
</div>
)}

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

@ -49,6 +49,8 @@ export type RecipeExecutionRecord = {
current_column: string | null;
progress: RecipeExecutionProgress | null;
// biome-ignore lint/style/useNamingConvention: backend schema
column_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;

163
studio/frontend/src/features/recipe-studio/executions/execution-helpers.ts Normal file
View file

@ -0,0 +1,163 @@
import type {
RecipeExecutionAnalysis,
RecipeExecutionRecord,
RecipeExecutionStatus,
} from "../execution-types";
import type { RecipePayload } from "../utils/payload/types";
export const DATASET_PAGE_SIZE = 20;
export function buildSignature(name: string, payload: RecipePayload): string {
return JSON.stringify({ name, payload });
}
export function formatSavedLabel(savedAt: number | null): string {
if (!savedAt) {
return "Not saved yet";
}
const time = new Date(savedAt).toLocaleTimeString([], {
hour: "numeric",
minute: "2-digit",
});
return `Saved ${time}`;
}
export function toErrorMessage(error: unknown, fallback: string): string {
if (error instanceof Error) {
return error.message;
}
return fallback;
}
export function normalizeDatasetRows(value: unknown): Record<string, unknown>[] {
if (!Array.isArray(value)) {
return [];
}
return value.filter(
(row): row is Record<string, unknown> =>
typeof row === "object" && row !== null && !Array.isArray(row),
);
}
export function normalizeObject(value: unknown): Record<string, unknown> | null {
if (typeof value !== "object" || value === null || Array.isArray(value)) {
return null;
}
return value as Record<string, unknown>;
}
export function normalizeAnalysis(value: unknown): RecipeExecutionAnalysis | null {
const normalized = normalizeObject(value);
if (!normalized) {
return null;
}
return normalized as RecipeExecutionAnalysis;
}
export 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";
}
export function executionLabel(kind: "preview" | "full"): string {
return kind === "preview" ? "Preview" : "Full run";
}
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;
}
export 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;
}
export function withExecutionDefaults(
record: RecipeExecutionRecord,
): RecipeExecutionRecord {
const dataset = Array.isArray(record.dataset) ? record.dataset : [];
const datasetPageSize =
typeof record.datasetPageSize === "number" && record.datasetPageSize > 0
? record.datasetPageSize
: DATASET_PAGE_SIZE;
const datasetPage =
typeof record.datasetPage === "number" && record.datasetPage > 0
? record.datasetPage
: 1;
const datasetTotal =
typeof record.datasetTotal === "number" && record.datasetTotal >= 0
? record.datasetTotal
: dataset.length;
return {
...record,
dataset,
datasetTotal,
datasetPage,
datasetPageSize,
column_progress: record.column_progress ?? null,
};
}
export function delay(ms: number): Promise<void> {
return new Promise((resolve) => {
window.setTimeout(resolve, ms);
});
}
export async function copyTextToClipboard(text: string): Promise<boolean> {
try {
if (navigator.clipboard?.writeText) {
await navigator.clipboard.writeText(text);
return true;
}
} catch {
// fallthrough to legacy path
}
try {
const textarea = document.createElement("textarea");
textarea.value = text;
textarea.setAttribute("readonly", "");
textarea.style.position = "fixed";
textarea.style.top = "0";
textarea.style.left = "-9999px";
document.body.appendChild(textarea);
textarea.select();
const ok = document.execCommand("copy");
document.body.removeChild(textarea);
return ok;
} catch {
return false;
}
}

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

@ -7,12 +7,26 @@ import {
getRecipeJobAnalysis,
getRecipeJobDataset,
getRecipeJobStatus,
previewRecipe,
streamRecipeJobEvents,
validateRecipe,
} from "../api";
import { listRecipeExecutions, saveRecipeExecution } from "../data/executions-db";
import {
buildSignature,
copyTextToClipboard,
DATASET_PAGE_SIZE,
delay,
executionLabel,
formatSavedLabel,
mapJobStatus,
normalizeAnalysis,
normalizeDatasetRows,
normalizeObject,
sortExecutions,
toErrorMessage,
withExecutionDefaults,
} from "../executions/execution-helpers";
import type {
RecipeExecutionAnalysis,
RecipeExecutionRecord,
RecipeExecutionStatus,
} from "../execution-types";
@ -74,157 +88,14 @@ type UseRecipeStudioActionsResult = {
importRecipe: (value: string) => string | null;
};
const DATASET_PAGE_SIZE = 20;
function buildSignature(name: string, payload: RecipePayload): string {
return JSON.stringify({ name, payload });
}
function formatSavedLabel(savedAt: number | null): string {
if (!savedAt) {
return "Not saved yet";
}
const time = new Date(savedAt).toLocaleTimeString([], {
hour: "numeric",
minute: "2-digit",
});
return `Saved ${time}`;
}
function toErrorMessage(error: unknown, fallback: string): string {
if (error instanceof Error) {
return error.message;
}
return fallback;
}
function normalizeDatasetRows(value: unknown): Record<string, unknown>[] {
if (!Array.isArray(value)) {
return [];
}
return value.filter(
(row): row is Record<string, unknown> =>
typeof row === "object" && row !== null && !Array.isArray(row),
);
}
function normalizeObject(value: unknown): Record<string, unknown> | null {
if (typeof value !== "object" || value === null || Array.isArray(value)) {
return 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 withExecutionDefaults(
record: RecipeExecutionRecord,
): RecipeExecutionRecord {
const dataset = Array.isArray(record.dataset) ? record.dataset : [];
const datasetPageSize =
typeof record.datasetPageSize === "number" && record.datasetPageSize > 0
? record.datasetPageSize
: DATASET_PAGE_SIZE;
const datasetPage =
typeof record.datasetPage === "number" && record.datasetPage > 0
? record.datasetPage
: 1;
const datasetTotal =
typeof record.datasetTotal === "number" && record.datasetTotal >= 0
? record.datasetTotal
: dataset.length;
return {
...record,
dataset,
datasetTotal,
datasetPage,
datasetPageSize,
};
}
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) {
await navigator.clipboard.writeText(text);
return true;
}
} catch {
// fallthrough to legacy path
}
try {
const textarea = document.createElement("textarea");
textarea.value = text;
textarea.setAttribute("readonly", "");
textarea.style.position = "fixed";
textarea.style.top = "0";
textarea.style.left = "-9999px";
document.body.appendChild(textarea);
textarea.select();
const ok = document.execCommand("copy");
document.body.removeChild(textarea);
return ok;
} catch {
return false;
}
}
type JobCompletedEventPayload = {
analysis?: unknown;
dataset?: unknown;
// biome-ignore lint/style/useNamingConvention: backend schema
artifact_path?: unknown;
error?: unknown;
type?: unknown;
};
export function useRecipeStudioActions({
recipeId,
@ -383,131 +254,22 @@ export function useRecipeStudioActions({
setPreviewDialogOpen(true);
}
const runPreview = useCallback(async (): Promise<boolean> => {
const payload = readPayload();
if (!payload) {
setPreviewErrors(payloadResult.errors);
toastError("Invalid recipe payload", payloadErrorMessage);
return false;
}
setPreviewLoading(true);
const runJobExecution = useCallback(async (input: {
kind: "preview" | "full";
payload: RecipePayload;
rows: number;
}): Promise<boolean> => {
const { kind, payload, rows } = input;
const setLoading = kind === "preview" ? setPreviewLoading : setFullLoading;
const label = executionLabel(kind);
setLoading(true);
const createdAt = Date.now();
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: [],
datasetTotal: 0,
datasetPage: 1,
datasetPageSize: DATASET_PAGE_SIZE,
analysis: null,
processor_artifacts: null,
error: null,
};
upsertExecution(baseExecution);
onExecutionStart?.();
setPreviewDialogOpen(false);
const previewPayload = {
...payload,
run: {
...payload.run,
rows: previewRows,
},
};
try {
const validation = await validateRecipe(previewPayload);
if (!validation.valid) {
const errors = validation.errors.map((item) => item.message);
const fallback = validation.raw_detail ?? "Validation failed.";
const nextErrors = errors.length > 0 ? errors : [fallback];
upsertExecution({
...baseExecution,
status: "error",
error: nextErrors[0],
});
setPreviewErrors(nextErrors);
toastError("Validation failed", nextErrors[0]);
return false;
}
const result = await previewRecipe(previewPayload);
const dataset = normalizeDatasetRows(result.dataset);
upsertExecution({
...baseExecution,
status: "completed",
finishedAt: Date.now(),
dataset,
datasetTotal: dataset.length,
datasetPage: 1,
datasetPageSize: DATASET_PAGE_SIZE,
analysis: normalizeAnalysis(result.analysis),
processor_artifacts: normalizeObject(result.processor_artifacts),
error: null,
});
setPreviewErrors([]);
toastSuccess(`Preview generated (${previewRows} rows).`);
onPreviewSuccess?.();
return true;
} catch (error) {
console.error("Preview failed:", error);
const message = toErrorMessage(error, "Preview request failed.");
upsertExecution({
...baseExecution,
status: "error",
finishedAt: Date.now(),
error: message,
});
setPreviewErrors([message]);
toastError("Preview failed", message);
return false;
} finally {
setPreviewLoading(false);
}
}, [
currentSignature,
onExecutionStart,
onPreviewSuccess,
payloadErrorMessage,
payloadResult.errors,
previewRows,
readPayload,
recipeId,
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",
kind,
status: "pending",
rows,
createdAt,
@ -516,6 +278,7 @@ export function useRecipeStudioActions({
stage: "pending",
current_column: null,
progress: null,
column_progress: null,
model_usage: null,
lastEventId: null,
artifact_path: null,
@ -530,65 +293,151 @@ export function useRecipeStudioActions({
upsertExecution(baseExecution);
onExecutionStart?.();
setFullLoading(true);
if (kind === "preview") {
setPreviewDialogOpen(false);
}
try {
const fullPayload = {
const jobPayload = {
...payload,
run: {
...payload.run,
rows,
// biome-ignore lint/style/useNamingConvention: backend schema
execution_type: "full",
execution_type: kind,
},
};
const createdJob = await createRecipeJob(fullPayload);
const createdJob = await createRecipeJob(jobPayload);
const jobId = createdJob.job_id;
let done = false;
let lastStatus: RecipeExecutionStatus = "pending";
let completedEventPayload: JobCompletedEventPayload | null = null;
let latestExecution: RecipeExecutionRecord = {
...baseExecution,
jobId,
};
upsertExecution(latestExecution);
while (!done) {
const status = await getRecipeJobStatus(jobId);
const mappedStatus = mapJobStatus(status.status);
lastStatus = mappedStatus;
const eventsAbortController = new AbortController();
void streamRecipeJobEvents({
jobId,
signal: eventsAbortController.signal,
onEvent: (event) => {
if (typeof event.id === "number") {
latestExecution = {
...latestExecution,
lastEventId: event.id,
};
}
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:
const eventType =
typeof event.payload.type === "string" ? event.payload.type : event.event;
if (eventType === "job.completed") {
lastStatus = "completed";
completedEventPayload = event.payload;
done = true;
latestExecution = {
...latestExecution,
status: "completed",
finishedAt: Date.now(),
artifact_path:
typeof event.payload.artifact_path === "string"
? event.payload.artifact_path
: latestExecution.artifact_path,
error: null,
};
upsertExecution(latestExecution);
return;
}
if (eventType === "job.error") {
lastStatus = "error";
done = true;
latestExecution = {
...latestExecution,
status: "error",
finishedAt: Date.now(),
error:
typeof event.payload.error === "string"
? event.payload.error
: latestExecution.error ?? `${label} failed.`,
};
upsertExecution(latestExecution);
return;
}
if (eventType === "job.cancelling") {
latestExecution = {
...latestExecution,
status: "cancelling",
};
upsertExecution(latestExecution);
}
},
}).catch(() => {
// polling remains fallback source of truth
});
try {
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,
column_progress:
(normalizeObject(status.column_progress) as RecipeExecutionRecord["column_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"
? Date.now()
: null,
};
upsertExecution(latestExecution);
done =
mappedStatus === "completed" ||
mappedStatus === "error" ||
mappedStatus === "cancelled";
if (!done) {
await delay(1200);
mappedStatus === "cancelled";
if (!done) {
await delay(1200);
}
}
} finally {
eventsAbortController.abort();
}
if (lastStatus === "completed") {
const eventAnalysis = completedEventPayload
? completedEventPayload["analysis"]
: null;
const eventDataset = completedEventPayload
? completedEventPayload["dataset"]
: null;
const shouldFetchPreviewDataset =
kind === "preview" && !Array.isArray(eventDataset);
const shouldFetchAnalysis =
!completedEventPayload ||
typeof eventAnalysis !== "object" ||
eventAnalysis === null ||
kind === "full";
const [analysisResult, datasetResult] = await Promise.allSettled([
getRecipeJobAnalysis(jobId),
getRecipeJobDataset(jobId, { limit: DATASET_PAGE_SIZE, offset: 0 }),
shouldFetchAnalysis
? getRecipeJobAnalysis(jobId)
: Promise.resolve(eventAnalysis),
shouldFetchPreviewDataset || kind === "full"
? getRecipeJobDataset(jobId, { limit: DATASET_PAGE_SIZE, offset: 0 })
: Promise.resolve({ dataset: eventDataset ?? [], total: rows }),
]);
const analysis =
analysisResult.status === "fulfilled"
@ -617,7 +466,14 @@ export function useRecipeStudioActions({
error: null,
finishedAt: latestExecution.finishedAt ?? Date.now(),
});
toastSuccess("Full run completed.");
if (kind === "preview") {
setPreviewErrors([]);
onPreviewSuccess?.();
toastSuccess(`Preview generated (${rows} rows).`);
} else {
toastSuccess("Full run completed.");
}
return true;
}
@ -628,39 +484,110 @@ export function useRecipeStudioActions({
error: latestExecution.error ?? "Run cancelled.",
finishedAt: latestExecution.finishedAt ?? Date.now(),
});
toastError("Full run cancelled", "The execution was cancelled.");
toastError(`${label} cancelled`, "The execution was cancelled.");
return false;
}
upsertExecution({
...latestExecution,
status: "error",
error: latestExecution.error ?? "Full run failed.",
error: latestExecution.error ?? `${label} failed.`,
finishedAt: latestExecution.finishedAt ?? Date.now(),
});
toastError("Full run failed", latestExecution.error ?? "Execution failed.");
toastError(`${label} failed`, latestExecution.error ?? "Execution failed.");
return false;
} catch (error) {
const message = toErrorMessage(error, "Full run request failed.");
const message = toErrorMessage(error, `${label} request failed.`);
upsertExecution({
...baseExecution,
status: "error",
error: message,
finishedAt: Date.now(),
});
toastError("Full run failed", message);
if (kind === "preview") {
setPreviewErrors([message]);
}
toastError(`${label} failed`, message);
return false;
} finally {
setFullLoading(false);
setLoading(false);
}
}, [
currentSignature,
onExecutionStart,
onPreviewSuccess,
recipeId,
upsertExecution,
]);
const runPreview = useCallback(async (): Promise<boolean> => {
const payload = readPayload();
if (!payload) {
setPreviewErrors(payloadResult.errors);
toastError("Invalid recipe payload", payloadErrorMessage);
return false;
}
const previewPayload = {
...payload,
run: {
...payload.run,
rows: previewRows,
},
};
try {
const validation = await validateRecipe(previewPayload);
if (!validation.valid) {
const errors = validation.errors.map((item) => item.message);
const fallback = validation.raw_detail ?? "Validation failed.";
const nextErrors = errors.length > 0 ? errors : [fallback];
setPreviewErrors(nextErrors);
toastError("Validation failed", nextErrors[0]);
return false;
}
} catch (error) {
const message = toErrorMessage(error, "Validation failed.");
setPreviewErrors([message]);
toastError("Validation failed", message);
return false;
}
return runJobExecution({
kind: "preview",
payload,
rows: previewRows,
});
}, [
payloadErrorMessage,
payloadResult.errors,
previewRows,
readPayload,
runJobExecution,
]);
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;
return runJobExecution({
kind: "full",
payload,
rows,
});
}, [
payloadErrorMessage,
payloadResult.errors,
readPayload,
recipeId,
upsertExecution,
runJobExecution,
]);
const cancelExecution = useCallback(async (id: string): Promise<void> => {