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

refactor: extract and consolidate execution runtime and tracking logic

Browse source
This commit is contained in:
Shine1i 2026-02-20 14:41:38 +01:00
commit 0d2f81ab3d
4 changed files with 534 additions and 451 deletions
Download patch file Download diff file Expand all files Collapse all files

22
studio/frontend/src/features/recipe-studio/executions/hydration.ts Normal file
View file

@ -0,0 +1,22 @@
import { listRecipeExecutions } from "../data/executions-db";
import type { RecipeExecutionRecord } from "../execution-types";
import {
isExecutionInProgress,
sortExecutions,
withExecutionDefaults,
} from "./execution-helpers";
export async function loadSortedRecipeExecutions(
recipeId: string,
): Promise<RecipeExecutionRecord[]> {
const records = await listRecipeExecutions(recipeId);
return sortExecutions(records.map(withExecutionDefaults));
}
export function findResumableExecution(
records: RecipeExecutionRecord[],
): RecipeExecutionRecord | null {
return (
records.find((record) => record.jobId && isExecutionInProgress(record.status)) ?? null
);
}

128
studio/frontend/src/features/recipe-studio/executions/runtime.ts Normal file
View file

@ -0,0 +1,128 @@
import type { JobEvent, JobStatusResponse } from "../api";
import type { RecipeExecutionKind, RecipeExecutionRecord } from "../execution-types";
import {
DATASET_PAGE_SIZE,
mapJobStatus,
normalizeObject,
} from "./execution-helpers";
const MAX_LOG_LINES = 1500;
function formatEventTime(ts: unknown): string {
if (typeof ts !== "number" || !Number.isFinite(ts)) {
return new Date().toLocaleTimeString();
}
const ms = ts > 10_000_000_000 ? ts : ts * 1000;
return new Date(ms).toLocaleTimeString();
}
export function appendExecutionLogLine(lines: string[], nextLine: string): string[] {
const next = [...lines, nextLine];
if (next.length <= MAX_LOG_LINES) {
return next;
}
return next.slice(next.length - MAX_LOG_LINES);
}
export function toExecutionLogLine(event: JobEvent): string | null {
const eventType =
typeof event.payload.type === "string" ? event.payload.type : event.event;
const ts = formatEventTime(event.payload.ts);
if (eventType === "log") {
const message =
typeof event.payload.message === "string" ? event.payload.message.trim() : "";
if (!message) {
return null;
}
const level =
typeof event.payload.level === "string" && event.payload.level.length > 0
? event.payload.level.toUpperCase()
: "INFO";
return `[${ts}] [${level}] ${message}`;
}
if (eventType === "job.started") {
return `[${ts}] [INFO] Job started`;
}
if (eventType === "job.completed") {
return `[${ts}] [INFO] Job completed`;
}
if (eventType === "job.cancelling") {
return `[${ts}] [WARN] Cancellation requested`;
}
if (eventType === "job.cancelled") {
return `[${ts}] [WARN] Job cancelled`;
}
if (eventType === "job.error") {
const error =
typeof event.payload.error === "string" && event.payload.error.length > 0
? event.payload.error
: "Job failed";
return `[${ts}] [ERROR] ${error}`;
}
return null;
}
export function applyExecutionStatusSnapshot(
execution: RecipeExecutionRecord,
status: JobStatusResponse,
): RecipeExecutionRecord {
const mappedStatus = mapJobStatus(status.status);
return {
...execution,
status: mappedStatus,
rows: status.rows ?? execution.rows,
stage: status.stage ?? execution.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 ?? execution.artifact_path,
error: status.error ?? null,
finishedAt:
mappedStatus === "completed" ||
mappedStatus === "error" ||
mappedStatus === "cancelled"
? Date.now()
: null,
};
}
export function createBaseExecutionRecord(input: {
recipeId: string;
kind: RecipeExecutionKind;
rows: number;
currentSignature: string;
}): RecipeExecutionRecord {
const createdAt = Date.now();
return {
id: crypto.randomUUID(),
recipeId: input.recipeId,
jobId: null,
kind: input.kind,
status: "pending",
rows: input.rows,
createdAt,
finishedAt: null,
recipeSignature: input.currentSignature,
stage: "pending",
current_column: null,
progress: null,
column_progress: null,
model_usage: null,
lastEventId: null,
artifact_path: null,
log_lines: [],
dataset: [],
datasetTotal: 0,
datasetPage: 1,
datasetPageSize: DATASET_PAGE_SIZE,
analysis: null,
processor_artifacts: null,
error: null,
};
}

320
studio/frontend/src/features/recipe-studio/executions/tracker.ts Normal file
View file

@ -0,0 +1,320 @@
import { toastError, toastSuccess } from "@/shared/toast";
import {
getRecipeJobAnalysis,
getRecipeJobDataset,
getRecipeJobStatus,
streamRecipeJobEvents,
} from "../api";
import type {
RecipeExecutionKind,
RecipeExecutionProgress,
RecipeExecutionRecord,
RecipeExecutionStatus,
} from "../execution-types";
import {
DATASET_PAGE_SIZE,
delay,
mapJobStatus,
normalizeAnalysis,
normalizeDatasetRows,
toErrorMessage,
} from "./execution-helpers";
import {
appendExecutionLogLine,
applyExecutionStatusSnapshot,
toExecutionLogLine,
} from "./runtime";
type TrackRecipeExecutionParams = {
label: string;
kind: RecipeExecutionKind;
rows: number;
jobId: string;
initialExecution: RecipeExecutionRecord;
notify: boolean;
onUpsert: (record: RecipeExecutionRecord) => void;
onSetPreviewErrors: (errors: string[]) => void;
onPreviewSuccess?: () => void;
};
function isTerminalStatus(status: RecipeExecutionStatus): boolean {
return status === "completed" || status === "error" || status === "cancelled";
}
function normalizeCompletedProgress(input: {
latestExecution: RecipeExecutionRecord;
rows: number;
}): {
progress: RecipeExecutionProgress;
columnProgress: RecipeExecutionProgress | null;
} {
const { latestExecution, rows } = input;
const progressTotal =
typeof latestExecution.progress?.total === "number" && latestExecution.progress.total > 0
? latestExecution.progress.total
: latestExecution.rows > 0
? latestExecution.rows
: rows;
const progress: RecipeExecutionProgress = {
...(latestExecution.progress ?? {}),
done: progressTotal,
total: progressTotal,
percent: 100,
eta_sec: 0,
};
const columnProgress =
latestExecution.column_progress &&
typeof latestExecution.column_progress.total === "number" &&
latestExecution.column_progress.total > 0
? {
...latestExecution.column_progress,
done: latestExecution.column_progress.total,
percent: 100,
eta_sec: 0,
}
: latestExecution.column_progress;
return { progress, columnProgress };
}
export async function trackRecipeExecution({
label,
kind,
rows,
jobId,
initialExecution,
notify,
onUpsert,
onSetPreviewErrors,
onPreviewSuccess,
}: TrackRecipeExecutionParams): Promise<boolean> {
let done = false;
let lastStatus: RecipeExecutionStatus = initialExecution.status;
let completedEventPayload: Record<string, unknown> | null = null;
let latestExecution: RecipeExecutionRecord = initialExecution;
const eventsAbortController = new AbortController();
void streamRecipeJobEvents({
jobId,
signal: eventsAbortController.signal,
lastEventId: latestExecution.lastEventId,
onEvent: (event) => {
let changed = false;
if (typeof event.id === "number") {
latestExecution = {
...latestExecution,
lastEventId: event.id,
};
changed = true;
}
const logLine = toExecutionLogLine(event);
if (logLine) {
latestExecution = {
...latestExecution,
log_lines: appendExecutionLogLine(latestExecution.log_lines, logLine),
};
changed = true;
}
const eventType =
typeof event.payload.type === "string" ? event.payload.type : event.event;
if (eventType === "job.started") {
latestExecution = {
...latestExecution,
status: "active",
};
onUpsert(latestExecution);
return;
}
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,
};
onUpsert(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.`,
};
onUpsert(latestExecution);
return;
}
if (eventType === "job.cancelling") {
latestExecution = {
...latestExecution,
status: "cancelling",
};
onUpsert(latestExecution);
return;
}
if (changed) {
onUpsert(latestExecution);
}
},
}).catch(() => {
// polling is fallback source of truth
});
try {
while (!done) {
const status = await getRecipeJobStatus(jobId);
const mappedStatus = mapJobStatus(status.status);
lastStatus = mappedStatus;
latestExecution = applyExecutionStatusSnapshot(latestExecution, status);
onUpsert(latestExecution);
done = isTerminalStatus(mappedStatus);
if (!done) {
await delay(1200);
}
}
} catch (error) {
const message = toErrorMessage(error, `${label} failed.`);
latestExecution = {
...latestExecution,
status: "error",
error: message,
finishedAt: Date.now(),
};
onUpsert(latestExecution);
if (notify) {
toastError(`${label} failed`, message);
}
return false;
} finally {
eventsAbortController.abort();
}
if (lastStatus === "completed") {
for (let attempt = 0; attempt < 3; attempt += 1) {
try {
const finalStatus = await getRecipeJobStatus(jobId);
latestExecution = applyExecutionStatusSnapshot(latestExecution, finalStatus);
} catch {
break;
}
if (attempt < 2) {
await delay(250);
}
}
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([
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"
? normalizeAnalysis(analysisResult.value)
: latestExecution.analysis;
const datasetResponse =
datasetResult.status === "fulfilled"
? datasetResult.value
: null;
const dataset = datasetResponse
? normalizeDatasetRows(datasetResponse.dataset)
: latestExecution.dataset;
const datasetTotal =
datasetResponse && typeof datasetResponse.total === "number"
? datasetResponse.total
: latestExecution.datasetTotal;
const completedProgress = normalizeCompletedProgress({ latestExecution, rows });
latestExecution = {
...latestExecution,
status: "completed",
progress: completedProgress.progress,
column_progress: completedProgress.columnProgress,
analysis,
dataset,
datasetTotal,
datasetPage: 1,
datasetPageSize: DATASET_PAGE_SIZE,
error: null,
finishedAt: latestExecution.finishedAt ?? Date.now(),
};
onUpsert(latestExecution);
if (notify) {
if (kind === "preview") {
onSetPreviewErrors([]);
onPreviewSuccess?.();
toastSuccess(`Preview generated (${rows} rows).`);
} else {
toastSuccess("Full run completed.");
}
}
return true;
}
if (lastStatus === "cancelled") {
latestExecution = {
...latestExecution,
status: "cancelled",
error: latestExecution.error ?? "Run cancelled.",
finishedAt: latestExecution.finishedAt ?? Date.now(),
};
onUpsert(latestExecution);
if (notify) {
toastError(`${label} cancelled`, "The execution was cancelled.");
}
return false;
}
latestExecution = {
...latestExecution,
status: "error",
error: latestExecution.error ?? `${label} failed.`,
finishedAt: latestExecution.finishedAt ?? Date.now(),
};
onUpsert(latestExecution);
if (notify) {
toastError(`${label} failed`, latestExecution.error ?? "Execution failed.");
}
return false;
}

515
studio/frontend/src/features/recipe-studio/hooks/use-recipe-executions.ts
View file

@ -1,35 +1,27 @@
import { useCallback, useEffect } from "react";
import { useShallow } from "zustand/react/shallow";
import { toastError, toastSuccess } from "@/shared/toast";
import { toastError } from "@/shared/toast";
import {
cancelRecipeJob,
createRecipeJob,
getRecipeJobAnalysis,
getRecipeJobDataset,
getRecipeJobStatus,
streamRecipeJobEvents,
validateRecipe,
type JobEvent,
type JobStatusResponse,
} from "../api";
import { listRecipeExecutions, saveRecipeExecution } from "../data/executions-db";
import type {
RecipeExecutionRecord,
RecipeExecutionStatus,
} from "../execution-types";
import { saveRecipeExecution } from "../data/executions-db";
import type { RecipeExecutionRecord } from "../execution-types";
import {
DATASET_PAGE_SIZE,
delay,
executionLabel,
isExecutionInProgress,
mapJobStatus,
normalizeAnalysis,
normalizeDatasetRows,
normalizeObject,
sortExecutions,
toErrorMessage,
withExecutionDefaults,
} from "../executions/execution-helpers";
import {
findResumableExecution,
loadSortedRecipeExecutions,
} from "../executions/hydration";
import { createBaseExecutionRecord } from "../executions/runtime";
import { trackRecipeExecution } from "../executions/tracker";
import { useRecipeExecutionsStore } from "../stores/recipe-executions";
import type { RecipePayload, RecipePayloadResult } from "../utils/payload/types";
@ -59,127 +51,6 @@ type UseRecipeExecutionsResult = {
loadExecutionDatasetPage: (id: string, page: number) => Promise<void>;
};
const MAX_LOG_LINES = 1500;
function formatEventTime(ts: unknown): string {
if (typeof ts !== "number" || !Number.isFinite(ts)) {
return new Date().toLocaleTimeString();
}
const ms = ts > 10_000_000_000 ? ts : ts * 1000;
return new Date(ms).toLocaleTimeString();
}
function appendLogLine(lines: string[], nextLine: string): string[] {
const next = [...lines, nextLine];
if (next.length <= MAX_LOG_LINES) {
return next;
}
return next.slice(next.length - MAX_LOG_LINES);
}
function toLogLine(event: JobEvent): string | null {
const eventType =
typeof event.payload.type === "string" ? event.payload.type : event.event;
const ts = formatEventTime(event.payload.ts);
if (eventType === "log") {
const message =
typeof event.payload.message === "string" ? event.payload.message.trim() : "";
if (!message) {
return null;
}
const level =
typeof event.payload.level === "string" && event.payload.level.length > 0
? event.payload.level.toUpperCase()
: "INFO";
return `[${ts}] [${level}] ${message}`;
}
if (eventType === "job.started") {
return `[${ts}] [INFO] Job started`;
}
if (eventType === "job.completed") {
return `[${ts}] [INFO] Job completed`;
}
if (eventType === "job.cancelling") {
return `[${ts}] [WARN] Cancellation requested`;
}
if (eventType === "job.cancelled") {
return `[${ts}] [WARN] Job cancelled`;
}
if (eventType === "job.error") {
const error =
typeof event.payload.error === "string" && event.payload.error.length > 0
? event.payload.error
: "Job failed";
return `[${ts}] [ERROR] ${error}`;
}
return null;
}
function applyStatusSnapshot(
execution: RecipeExecutionRecord,
status: JobStatusResponse,
): RecipeExecutionRecord {
const mappedStatus = mapJobStatus(status.status);
return {
...execution,
status: mappedStatus,
rows: status.rows ?? execution.rows,
stage: status.stage ?? execution.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 ?? execution.artifact_path,
error: status.error ?? null,
finishedAt:
mappedStatus === "completed" ||
mappedStatus === "error" ||
mappedStatus === "cancelled"
? Date.now()
: null,
};
}
function createBaseExecution(input: {
recipeId: string;
kind: "preview" | "full";
rows: number;
currentSignature: string;
}): RecipeExecutionRecord {
const createdAt = Date.now();
return {
id: crypto.randomUUID(),
recipeId: input.recipeId,
jobId: null,
kind: input.kind,
status: "pending",
rows: input.rows,
createdAt,
finishedAt: null,
recipeSignature: input.currentSignature,
stage: "pending",
current_column: null,
progress: null,
column_progress: null,
model_usage: null,
lastEventId: null,
artifact_path: null,
log_lines: [],
dataset: [],
datasetTotal: 0,
datasetPage: 1,
datasetPageSize: DATASET_PAGE_SIZE,
analysis: null,
processor_artifacts: null,
error: null,
};
}
export function useRecipeExecutions({
recipeId,
currentSignature,
@ -224,328 +95,66 @@ export function useRecipeExecutions({
resetForRecipe: state.resetForRecipe,
})),
);
const payloadErrorMessage = payloadResult.errors[0] ?? "Invalid payload.";
const upsertAndPersist = useCallback(
(record: RecipeExecutionRecord): void => {
const normalized = withExecutionDefaults(record);
upsertExecution(normalized);
void saveRecipeExecution(normalized).catch((error) => {
const normalizedRecord = withExecutionDefaults(record);
upsertExecution(normalizedRecord);
void saveRecipeExecution(normalizedRecord).catch((error) => {
console.error("Save recipe execution failed:", error);
});
},
[upsertExecution],
);
const trackExecution = useCallback(
async (input: {
label: string;
kind: "preview" | "full";
rows: number;
jobId: string;
initialExecution: RecipeExecutionRecord;
notify: boolean;
}): Promise<boolean> => {
const { label, kind, rows, jobId, notify } = input;
let done = false;
let lastStatus: RecipeExecutionStatus = input.initialExecution.status;
let completedEventPayload: Record<string, unknown> | null = null;
let latestExecution: RecipeExecutionRecord = input.initialExecution;
const eventsAbortController = new AbortController();
void streamRecipeJobEvents({
jobId,
signal: eventsAbortController.signal,
lastEventId: latestExecution.lastEventId,
onEvent: (event) => {
let changed = false;
if (typeof event.id === "number") {
latestExecution = {
...latestExecution,
lastEventId: event.id,
};
changed = true;
}
const logLine = toLogLine(event);
if (logLine) {
latestExecution = {
...latestExecution,
log_lines: appendLogLine(latestExecution.log_lines, logLine),
};
changed = true;
}
const eventType =
typeof event.payload.type === "string" ? event.payload.type : event.event;
if (eventType === "job.started") {
latestExecution = {
...latestExecution,
status: "active",
};
upsertAndPersist(latestExecution);
return;
}
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,
};
upsertAndPersist(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.`,
};
upsertAndPersist(latestExecution);
return;
}
if (eventType === "job.cancelling") {
latestExecution = {
...latestExecution,
status: "cancelling",
};
upsertAndPersist(latestExecution);
return;
}
if (changed) {
upsertAndPersist(latestExecution);
}
},
}).catch(() => {
// polling is fallback source of truth
});
try {
while (!done) {
const status = await getRecipeJobStatus(jobId);
const mappedStatus = mapJobStatus(status.status);
lastStatus = mappedStatus;
latestExecution = applyStatusSnapshot(latestExecution, status);
upsertAndPersist(latestExecution);
done =
mappedStatus === "completed" ||
mappedStatus === "error" ||
mappedStatus === "cancelled";
if (!done) {
await delay(1200);
}
}
} catch (error) {
const message = toErrorMessage(error, `${label} failed.`);
latestExecution = {
...latestExecution,
status: "error",
error: message,
finishedAt: Date.now(),
};
upsertAndPersist(latestExecution);
if (notify) {
toastError(`${label} failed`, message);
}
return false;
} finally {
eventsAbortController.abort();
}
if (lastStatus === "completed") {
for (let attempt = 0; attempt < 3; attempt += 1) {
try {
const finalStatus = await getRecipeJobStatus(jobId);
latestExecution = applyStatusSnapshot(latestExecution, finalStatus);
} catch {
break;
}
if (attempt < 2) {
await delay(250);
}
}
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([
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"
? normalizeAnalysis(analysisResult.value)
: latestExecution.analysis;
const datasetResponse =
datasetResult.status === "fulfilled"
? datasetResult.value
: null;
const dataset = datasetResponse
? normalizeDatasetRows(datasetResponse.dataset)
: latestExecution.dataset;
const datasetTotal =
datasetResponse && typeof datasetResponse.total === "number"
? datasetResponse.total
: latestExecution.datasetTotal;
const progressTotal =
typeof latestExecution.progress?.total === "number" &&
latestExecution.progress.total > 0
? latestExecution.progress.total
: latestExecution.rows > 0
? latestExecution.rows
: rows;
const completedProgress: RecipeExecutionRecord["progress"] = {
...(latestExecution.progress ?? {}),
done: progressTotal,
total: progressTotal,
percent: 100,
eta_sec: 0,
};
const completedColumnProgress: RecipeExecutionRecord["column_progress"] =
latestExecution.column_progress &&
typeof latestExecution.column_progress.total === "number" &&
latestExecution.column_progress.total > 0
? {
...latestExecution.column_progress,
done: latestExecution.column_progress.total,
percent: 100,
eta_sec: 0,
}
: latestExecution.column_progress;
latestExecution = {
...latestExecution,
status: "completed",
progress: completedProgress,
column_progress: completedColumnProgress,
analysis,
dataset,
datasetTotal,
datasetPage: 1,
datasetPageSize: DATASET_PAGE_SIZE,
error: null,
finishedAt: latestExecution.finishedAt ?? Date.now(),
};
upsertAndPersist(latestExecution);
if (notify) {
if (kind === "preview") {
setPreviewErrors([]);
onPreviewSuccess?.();
toastSuccess(`Preview generated (${rows} rows).`);
} else {
toastSuccess("Full run completed.");
}
}
return true;
}
if (lastStatus === "cancelled") {
latestExecution = {
...latestExecution,
status: "cancelled",
error: latestExecution.error ?? "Run cancelled.",
finishedAt: latestExecution.finishedAt ?? Date.now(),
};
upsertAndPersist(latestExecution);
if (notify) {
toastError(`${label} cancelled`, "The execution was cancelled.");
}
return false;
}
latestExecution = {
...latestExecution,
status: "error",
error: latestExecution.error ?? `${label} failed.`,
finishedAt: latestExecution.finishedAt ?? Date.now(),
};
upsertAndPersist(latestExecution);
if (notify) {
toastError(`${label} failed`, latestExecution.error ?? "Execution failed.");
}
return false;
},
[onPreviewSuccess, setPreviewErrors, upsertAndPersist],
);
useEffect(() => {
let cancelled = false;
resetForRecipe();
async function loadExecutions(): Promise<void> {
async function hydrate(): Promise<void> {
try {
const records = await listRecipeExecutions(recipeId);
const records = await loadSortedRecipeExecutions(recipeId);
if (cancelled) {
return;
}
const sortedRecords = sortExecutions(records.map(withExecutionDefaults));
setExecutions(sortedRecords);
const activeExecution = sortedRecords.find(
(record) => record.jobId && isExecutionInProgress(record.status),
);
if (activeExecution?.jobId) {
void trackExecution({
label: executionLabel(activeExecution.kind),
kind: activeExecution.kind,
rows: activeExecution.rows,
jobId: activeExecution.jobId,
initialExecution: activeExecution,
notify: false,
});
setExecutions(records);
const resumable = findResumableExecution(records);
if (!resumable?.jobId) {
return;
}
void trackRecipeExecution({
label: executionLabel(resumable.kind),
kind: resumable.kind,
rows: resumable.rows,
jobId: resumable.jobId,
initialExecution: resumable,
notify: false,
onUpsert: upsertAndPersist,
onSetPreviewErrors: setPreviewErrors,
onPreviewSuccess,
});
} catch (error) {
console.error("Load recipe executions failed:", error);
}
}
void loadExecutions();
void hydrate();
return () => {
cancelled = true;
};
}, [recipeId, resetForRecipe, setExecutions, trackExecution]);
}, [
onPreviewSuccess,
recipeId,
resetForRecipe,
setExecutions,
setPreviewErrors,
upsertAndPersist,
]);
const readPayload = useCallback((): RecipePayload | null => {
if (payloadResult.errors.length === 0) {
@ -553,6 +162,16 @@ export function useRecipeExecutions({
}
return null;
}, [payloadResult.errors.length, payloadResult.payload]);
const readExecutablePayload = useCallback((): RecipePayload | null => {
const payload = readPayload();
if (payload) {
return payload;
}
setPreviewErrors(payloadResult.errors);
toastError("Invalid recipe payload", payloadErrorMessage);
return null;
}, [payloadErrorMessage, payloadResult.errors, readPayload, setPreviewErrors]);
const openPreviewDialog = useCallback((): void => {
setPreviewErrors([]);
@ -570,7 +189,7 @@ export function useRecipeExecutions({
const label = executionLabel(kind);
setLoading(true);
const baseExecution = createBaseExecution({
const baseExecution = createBaseExecutionRecord({
recipeId,
kind,
rows,
@ -594,19 +213,22 @@ export function useRecipeExecutions({
},
};
const createdJob = await createRecipeJob(jobPayload);
const latestExecution = {
const executionWithJob = {
...baseExecution,
jobId: createdJob.job_id,
};
upsertAndPersist(latestExecution);
upsertAndPersist(executionWithJob);
return await trackExecution({
return await trackRecipeExecution({
label,
kind,
rows,
jobId: createdJob.job_id,
initialExecution: latestExecution,
initialExecution: executionWithJob,
notify: true,
onUpsert: upsertAndPersist,
onSetPreviewErrors: setPreviewErrors,
onPreviewSuccess,
});
} catch (error) {
const message = toErrorMessage(error, `${label} request failed.`);
@ -628,22 +250,19 @@ export function useRecipeExecutions({
[
currentSignature,
onExecutionStart,
onPreviewSuccess,
recipeId,
setFullLoading,
setPreviewDialogOpen,
setPreviewErrors,
setPreviewLoading,
trackExecution,
upsertAndPersist,
],
);
const runPreview = useCallback(async (): Promise<boolean> => {
const payload = readPayload();
const payloadErrorMessage = payloadResult.errors[0] ?? "Invalid payload.";
const payload = readExecutablePayload();
if (!payload) {
setPreviewErrors(payloadResult.errors);
toastError("Invalid recipe payload", payloadErrorMessage);
return false;
}
@ -676,14 +295,11 @@ export function useRecipeExecutions({
payload,
rows: previewRows,
});
}, [payloadResult.errors, previewRows, readPayload, runExecution, setPreviewErrors]);
}, [previewRows, readExecutablePayload, runExecution, setPreviewErrors]);
const runFull = useCallback(async (): Promise<boolean> => {
const payload = readPayload();
const payloadErrorMessage = payloadResult.errors[0] ?? "Invalid payload.";
const payload = readExecutablePayload();
if (!payload) {
setPreviewErrors(payloadResult.errors);
toastError("Invalid recipe payload", payloadErrorMessage);
return false;
}
@ -698,7 +314,7 @@ export function useRecipeExecutions({
payload,
rows,
});
}, [payloadResult.errors, readPayload, runExecution, setPreviewErrors]);
}, [readExecutablePayload, runExecution]);
const cancelExecution = useCallback(
async (id: string): Promise<void> => {
@ -723,10 +339,7 @@ export function useRecipeExecutions({
const loadExecutionDatasetPage = useCallback(
async (id: string, page: number): Promise<void> => {
const execution = executions.find((entry) => entry.id === id);
if (!execution || execution.kind !== "full" || !execution.jobId) {
return;
}
if (page < 1) {
if (!execution || execution.kind !== "full" || !execution.jobId || page < 1) {
return;
}