refactor(recipe-studio): simplify runtime graph flow + guard stale active execution lock p2

studio/backend/core/data_recipe/jobs/parse.py

@@ -249,5 +249,5 @@ def _compute_overall_progress(job: Job, column_progress: Progress) -> Progress:
 
 
 def coerce_event(obj: Any) -> dict:
-    # worker sends dict already
+    """Normalize worker payload into event dict."""
     return obj if isinstance(obj, dict) else {"type": "log", "message": str(obj)}

studio/frontend/src/features/recipe-studio/components/recipe-graph-node.tsx

@@ -422,15 +422,15 @@ function RecipeGraphNodeBase({
               {llmAuxVisible ? "Hide inputs" : "Show inputs"}
             </Button>
           )}
-            <Button
+          <Button
             type="button"
             size="xs"
             variant="ghost"
-              className="nodrag"
-              disabled={executionLocked}
-              onClick={(event) => {
-                event.preventDefault();
-                event.stopPropagation();
+            className="nodrag"
+            disabled={executionLocked}
+            onClick={(event) => {
+              event.preventDefault();
+              event.stopPropagation();
               openConfig(id);
             }}
           >

studio/frontend/src/features/recipe-studio/recipe-studio-page.tsx

@@ -572,6 +572,7 @@ export function RecipeStudioPage({
   );
   const executionLocked = runtimeVisualState.executionLocked;
   const canvasInteractive = interactive && !executionLocked;
+  const runBusy = previewLoading || fullLoading || executionLocked;
   const currentColumnConfig = useMemo(() => {
     const columnName = activeExecution?.current_column?.trim();
     if (!columnName) {
@@ -782,12 +783,10 @@ export function RecipeStudioPage({
                       type="button"
                       className="h-11 px-5"
                       onClick={() => openRunDialog(runDialogKind)}
-                      disabled={previewLoading || fullLoading || executionLocked}
+                      disabled={runBusy}
                     >
                       <HugeiconsIcon icon={CookBookIcon} className="size-4" />
-                      {previewLoading || fullLoading || executionLocked
-                        ? "Running..."
-                        : "Run"}
+                      {runBusy ? "Running..." : "Run"}
                     </Button>
                     <Button
                       type="button"

studio/frontend/src/features/recipe-studio/utils/graph/derive-display-graph.ts

@@ -37,6 +37,10 @@ export type DisplayGraph = {
   edges: Edge[];
 };
 
+function isAuxEdge(edge: Edge): boolean {
+  return edge.source.startsWith("aux-") || edge.target.startsWith("aux-");
+}
+
 function normalizeEdge(
   edge: Edge,
   configs: Record<string, NodeConfig>,
@@ -44,7 +48,7 @@ function normalizeEdge(
   activeEdgeIds: Set<string>,
 ): Edge {
   const isActiveEdge = activeEdgeIds.has(edge.id);
-  const isAux = edge.source.startsWith("aux-") || edge.target.startsWith("aux-");
+  const isAux = isAuxEdge(edge);
   if (isAux) {
     return {
       ...edge,
@@ -56,7 +60,9 @@ function normalizeEdge(
 
   const source = configs[edge.source];
   const target = configs[edge.target];
-  const semantic = Boolean(source && target) && isSemanticRelation(source, target);
+  const semantic =
+    edge.type === "semantic" ||
+    (Boolean(source && target) && isSemanticRelation(source, target));
   const sourceHandleNormalized = normalizeRecipeHandleId(edge.sourceHandle);
   const targetHandleNormalized = normalizeRecipeHandleId(edge.targetHandle);
   const semanticSourceDefault =
@@ -237,7 +243,7 @@ function pickAuxTargetHandle(
 ): string {
   const occupied = new Set<HandleSide>();
   for (const edge of edges) {
-    if (edge.source.startsWith("aux-") || edge.target.startsWith("aux-")) {
+    if (isAuxEdge(edge)) {
       continue;
     }
     if (edge.target === llmId) {

studio/frontend/src/features/recipe-studio/utils/graph/runtime-visual-state.ts

@@ -12,6 +12,16 @@ const ACTIVE_STATUSES: ReadonlySet<RecipeExecutionStatus> = new Set([
   "active",
   "cancelling",
 ]);
+const FRESH_PENDING_WINDOW_MS = 60_000;
+
+const DONE_UPSTREAM_KINDS: ReadonlySet<NodeConfig["kind"]> = new Set([
+  "sampler",
+  "seed",
+  "expression",
+  "llm",
+  "model_config",
+  "model_provider",
+]);
 
 export type GraphRuntimeVisualState = {
   executionLocked: boolean;
@@ -21,13 +31,50 @@ export type GraphRuntimeVisualState = {
   batch: RecipeExecutionBatch | null;
 };
 
+function isAuxEdge(edge: Edge): boolean {
+  return edge.source.startsWith("aux-") || edge.target.startsWith("aux-");
+}
+
+function hasLiveExecutionSignal(execution: RecipeExecutionRecord): boolean {
+  if (execution.lastEventId !== null) {
+    return true;
+  }
+  if (execution.current_column !== null) {
+    return true;
+  }
+  if (execution.progress !== null || execution.column_progress !== null) {
+    return true;
+  }
+  return Boolean(execution.batch?.idx ?? execution.batch?.total);
+}
+
 export function pickLatestActiveExecution(
   executions: RecipeExecutionRecord[],
 ): RecipeExecutionRecord | null {
+  const now = Date.now();
   for (const execution of executions) {
-    if (ACTIVE_STATUSES.has(execution.status)) {
-      return execution;
+    if (!ACTIVE_STATUSES.has(execution.status)) {
+      continue;
     }
+    if (!execution.jobId) {
+      continue;
+    }
+    if (execution.finishedAt !== null) {
+      continue;
+    }
+
+    const liveSignal = hasLiveExecutionSignal(execution);
+    if (!liveSignal && execution.status === "pending") {
+      const ageMs = Math.max(0, now - execution.createdAt);
+      if (ageMs > FRESH_PENDING_WINDOW_MS) {
+        continue;
+      }
+    }
+    if (!liveSignal && execution.status !== "pending") {
+      continue;
+    }
+
+    return execution;
   }
   return null;
 }
@@ -85,7 +132,7 @@ export function deriveGraphRuntimeVisualState(input: {
       if (edge.target !== runningNodeId) {
         continue;
       }
-      if (edge.source.startsWith("aux-") || edge.target.startsWith("aux-")) {
+      if (isAuxEdge(edge)) {
         continue;
       }
       activeEdgeIds.add(edge.id);
@@ -114,17 +161,9 @@ function collectUpstreamDoneNodeIds(input: {
   configs: Record<string, NodeConfig>;
 }): Set<string> {
   const { rootNodeId, edges, configs } = input;
-  const doneKinds = new Set<NodeConfig["kind"]>([
-    "sampler",
-    "seed",
-    "expression",
-    "llm",
-    "model_config",
-    "model_provider",
-  ]);
   const incoming = new Map<string, string[]>();
   for (const edge of edges) {
-    if (edge.source.startsWith("aux-") || edge.target.startsWith("aux-")) {
+    if (isAuxEdge(edge)) {
       continue;
     }
     const list = incoming.get(edge.target) ?? [];
@@ -134,9 +173,11 @@ function collectUpstreamDoneNodeIds(input: {
 
   const visited = new Set<string>();
   const queue = [rootNodeId];
+  let queueIndex = 0;
   const doneNodeIds = new Set<string>();
-  while (queue.length > 0) {
-    const current = queue.shift();
+  while (queueIndex < queue.length) {
+    const current = queue[queueIndex];
+    queueIndex += 1;
     if (!current || visited.has(current)) {
       continue;
     }
@@ -147,7 +188,7 @@ function collectUpstreamDoneNodeIds(input: {
         queue.push(sourceId);
       }
       const config = configs[sourceId];
-      if (config && doneKinds.has(config.kind)) {
+      if (config && DONE_UPSTREAM_KINDS.has(config.kind)) {
         doneNodeIds.add(sourceId);
       }
     }