feat(studio): complete Houdini workflow editing and domestic OODA

backend/workflow/hda_templates.py

@@ -8,7 +8,6 @@
 from pathlib import Path
 from typing import Any
 
-from backend.workflow.tool_capabilities import resolve_workflow_tool_capability
 from backend.schemas.workflow import (
     WorkflowAdapterBinding,
     WorkflowPackageInternals,
@@ -17,6 +16,7 @@
     WorkflowProjectNode,
     WorkflowTopicCollapse,
 )
+from backend.workflow.tool_capabilities import resolve_workflow_tool_capability
 
 OPENCLI_MULTI_SOURCE_TEMPLATE = "opencli-multi-source"
 OPENCLI_SOURCE_POOL_CATALOG_ID = "intelligence.source.pool"
@@ -66,6 +66,9 @@ def _materialize_node(node: WorkflowProjectNode) -> WorkflowProjectNode:
         sources = _source_slots(node.params.get("sources"))
         if sources:
             expose_raw_source_items = node.params.get("exposeRawSourceItems") is True
+            failure_mode = _read_string(
+                _read_dict(node.params.get("execution")).get("failureMode")
+            )
             internals = _opencli_multi_source_internals(
                 sources,
                 expose_raw_source_items=expose_raw_source_items,
@@ -78,6 +81,7 @@ def _materialize_node(node: WorkflowProjectNode) -> WorkflowProjectNode:
                 "lockedInternals": node.params.get("lockedInternals", True),
                 "execution": {
                     "fanout": "parallel",
+                    **({"failureMode": failure_mode} if failure_mode else {}),
                 },
             }
             ui = {**(node.ui or {}), "catalogId": OPENCLI_HDA_CATALOG_ID}

backend/workflow/opencli_hda_tracer.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import uuid
 from dataclasses import dataclass
 from datetime import UTC, datetime
@@ -72,17 +71,17 @@
     JoyAIVLExecutionError,
     execute_joyai_vl_interaction,
 )
-from backend.workflow.native_node_runtime import (
-    NATIVE_BINDING_IDS,
-    NativeNodeValidationError,
-    execute_native_node,
-)
 from backend.workflow.last30days_provider import Last30DaysProviderError
 from backend.workflow.native_intelligence_executor import (
     NATIVE_INTELLIGENCE_ACTION_BY_TOOL_ID,
     NATIVE_INTELLIGENCE_EXECUTOR,
     execute_native_intelligence_action,
 )
+from backend.workflow.native_node_runtime import (
+    NATIVE_BINDING_IDS,
+    NativeNodeValidationError,
+    execute_native_node,
+)
 from backend.workflow.realtime_market_executor import (
     OKX_MARKET_TICKER_SNAPSHOT_EXECUTOR,
     RealtimeMarketExecutionError,
@@ -98,8 +97,8 @@
     execute_workflow_rss_source,
 )
 from backend.workflow.runtime_registry import (
-    DATA_OPERATOR_CATALOG_BINDINGS,
     COLLECTION_OUTPUT_BINDING_ID,
+    DATA_OPERATOR_CATALOG_BINDINGS,
     DEDUPE_BINDING_ID,
     DIFY_GRAPHON_BINDING_ID,
     EXTERNAL_TOOL_BINDING_ID,
@@ -1253,11 +1252,67 @@ async def start_workflow_run(
 
         internal_reasons = blocked_by_package.get(package_node.id, [])
         if internal_reasons:
+            descendant_ids = {
+                node.id
+                for node in runtime_nodes
+                if package_node.id in _package_ancestor_ids(node)
+            }
+            source_node_ids = {
+                node.id
+                for node in runtime_nodes
+                if node.id in descendant_ids
+                and (
+                    _read_string(node.params.get("sourceGroup"))
+                    or _read_string(node.params.get("source_group"))
+                )
+            }
+            terminal_events_by_node: dict[str, WorkflowNodeRunEvent] = {}
+            for event in emitter.events:
+                if (
+                    event.nodeId in descendant_ids
+                    and event.eventType in {"completed", "failed", "blocked"}
+                ):
+                    terminal_events_by_node[event.nodeId] = event
+            source_terminal_events = [
+                event
+                for event in terminal_events_by_node.values()
+                if event.nodeId in source_node_ids
+            ]
+            has_successful_source = any(
+                event.eventType == "completed" for event in source_terminal_events
+            )
+            has_source_failure = any(
+                event.eventType in {"failed", "blocked"}
+                for event in source_terminal_events
+            )
+            has_non_source_failure = any(
+                event.eventType in {"failed", "blocked"}
+                and event.nodeId not in source_node_ids
+                for event in terminal_events_by_node.values()
+            )
+            tolerates_internal_reasons = (
+                _collects_per_source_failures(package_node)
+                and has_successful_source
+                and has_source_failure
+                and not has_non_source_failure
+            )
             emitter.emit(
                 package_node,
                 "partial",
-                message="Package produced partial source results before an internal block",
+                message=(
+                    "Package collected available source results with per-source failures"
+                    if tolerates_internal_reasons
+                    else "Package produced partial source results before an internal block"
+                ),
             )
+            if tolerates_internal_reasons:
+                emitter.emit(
+                    package_node,
+                    "completed",
+                    message="Package completed with per-source failures preserved in the trace",
+                    details={"sourceFailureCount": len(internal_reasons)},
+                )
+                continue
             emitter.emit(
                 package_node,
                 "blocked",
@@ -2065,6 +2120,38 @@ def _run_status(
     if not valid:
         return "failed"
     statuses = {state.status for state in node_states}
+    collect_per_source_package_ids_by_state = (
+        {
+            state.nodeId: _collect_per_source_package_ids(state, runtime_nodes)
+            for state in node_states
+        }
+        if runtime_nodes
+        else {}
+    )
+    successful_source_package_ids = {
+        package_id
+        for state in node_states
+        if state.status == "completed"
+        for package_id in collect_per_source_package_ids_by_state.get(state.nodeId, set())
+    }
+    tolerated_source_failure_ids = (
+        {
+            state.nodeId
+            for state in node_states
+            if state.status in {"failed", "blocked"}
+            and (
+                collect_per_source_package_ids_by_state.get(state.nodeId, set())
+                & successful_source_package_ids
+            )
+        }
+        if runtime_nodes
+        else set()
+    )
+    effective_statuses = {
+        state.status
+        for state in node_states
+        if state.nodeId not in tolerated_source_failure_ids
+    }
     if runtime_nodes:
         terminal_ids = {node.id for node in runtime_nodes if _is_builder_output(node)}
         terminal_statuses = {state.status for state in node_states if state.nodeId in terminal_ids}
@@ -2076,17 +2163,53 @@ def _run_status(
             and terminal_statuses.intersection({"failed", "blocked"})
         ):
             return "partial_success"
-    if "failed" in statuses:
+    if "failed" in effective_statuses:
         return "failed"
-    if "blocked" in statuses:
+    if "blocked" in effective_statuses:
         return "blocked"
-    if "running" in statuses or "partial" in statuses:
+    if "running" in effective_statuses or "partial" in effective_statuses:
         return "partial"
+    if tolerated_source_failure_ids and effective_statuses and effective_statuses <= {"completed"}:
+        return "partial_success"
     if statuses and statuses <= {"completed"}:
         return "completed"
     return "queued"
 
 
+def _collects_per_source_failures(node: CompiledWorkflowNode) -> bool:
+    execution = _read_dict(node.params.get("execution"))
+    return _read_string(execution.get("failureMode")) == "collect-per-source"
+
+
+def _collect_per_source_package_ids(
+    state: WorkflowRunNodeState,
+    runtime_nodes: list[CompiledWorkflowNode],
+) -> set[str]:
+    source_groups = getattr(state, "sourceGroups", [])
+    node_path = getattr(state, "nodePath", [])
+    runtime_node = next(
+        (node for node in runtime_nodes if node.id == state.nodeId),
+        None,
+    )
+    if (
+        not source_groups
+        or runtime_node is None
+        or not (
+            _read_string(runtime_node.params.get("sourceGroup"))
+            or _read_string(runtime_node.params.get("source_group"))
+        )
+    ):
+        return set()
+    tolerant_package_ids = {
+        node.id for node in runtime_nodes if _collects_per_source_failures(node)
+    }
+    return {
+        INTERNAL_ID_SEPARATOR.join(node_path[:depth])
+        for depth in range(1, len(node_path))
+        if INTERNAL_ID_SEPARATOR.join(node_path[:depth]) in tolerant_package_ids
+    }
+
+
 def _is_builder_output(node: CompiledWorkflowNode) -> bool:
     builder = _read_dict(node.params.get("builder"))
     return _read_string(builder.get("nodeType")) in {

frontend/components/flow/command-palette.tsx

@@ -46,7 +46,11 @@ import { workflowNodeDepthFromNetworkStack, workflowNodeLayerAtDepth } from "@/l
 import { localizeNodeText, type WorkflowLanguage } from "@/lib/workflow/node-i18n"
 import { groupPrimitivesForNodeMenu } from "@/lib/workflow/node-menu"
 import { getNodeContractByCatalogId } from "@/lib/workflow/node-contracts"
-import { getWorkflowPrimitives, type WorkflowPrimitive } from "@/lib/workflow/node-primitives"
+import {
+  getDifyCommonWorkflowPrimitives,
+  getWorkflowPrimitives,
+  type WorkflowPrimitive,
+} from "@/lib/workflow/node-primitives"
 import { openCLIAdapterNodeToCatalogItem } from "@/lib/workflow/opencli-adapter-catalog"
 import { useWorkflowCapabilities } from "@/lib/workflow/use-workflow-capabilities"
 import { cn } from "@/lib/utils"
@@ -642,6 +646,7 @@ export function CommandPalette({
       workflowCatalogPluginProvenance(item) === null &&
       item.runtimeCapability?.source !== "backend.workflow.tool_capabilities",
   )
+  const catalogOperatorIds = new Set(catalogOperators.map((item) => item.id))
   const pluginTools = allCatalogItems.filter(
     (item) =>
       catalogAcceptsConnection(item, compatiblePort) &&
@@ -679,7 +684,8 @@ export function CommandPalette({
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [capabilities, catalogItems, compatiblePort, inNodeNetwork, language, queryText, workflowProfile])
   const primitiveGroups = groupPrimitivesForNodeMenu(
-    (inNodeNetwork ? getWorkflowPrimitives() : []).filter((item) => {
+    (inNodeNetwork ? getWorkflowPrimitives() : getDifyCommonWorkflowPrimitives()).filter((item) => {
+      if (catalogOperatorIds.has(item.id)) return false
       if (!primitiveAcceptsConnection(item, compatiblePort)) return false
       if (!queryText) return true
       const text = localizeNodeText(item.id, { label: item.label, description: item.description }, language)
@@ -748,9 +754,23 @@ export function CommandPalette({
             <div className="min-w-0"><div className="truncate text-sm font-medium">{selectedPresentation.label}</div><div className="truncate text-xs text-muted-foreground">{copy.requiredBeforeAdd}</div></div>
           </div>
           <div className="grid max-h-[52vh] gap-3 overflow-y-auto p-4">
-            {selectedOpenCLI.args.filter((arg) => arg.required).map((arg) => (
-              <label key={arg.name} className="grid gap-1.5 text-xs"><span>{arg.name}<span className="ml-1 text-destructive">*</span></span><input value={requiredValues[arg.name] ?? ""} onChange={(event) => setRequiredValues((current) => ({ ...current, [arg.name]: event.target.value }))} placeholder={arg.help ?? `${copy.input} ${arg.name}`} className="min-h-11 rounded-md border bg-background px-3 text-sm outline-none focus:ring-2 focus:ring-ring/50" autoFocus={selectedOpenCLI.requiredArgs[0] === arg.name} /></label>
-            ))}
+            {selectedOpenCLI.args.filter((arg) => arg.required).map((arg) => {
+              const value = requiredValues[arg.name] ?? (arg.default == null ? "" : String(arg.default))
+              const onChange = (next: string) => setRequiredValues((current) => ({ ...current, [arg.name]: next }))
+              return (
+                <label key={arg.name} className="grid gap-1.5 text-xs">
+                  <span>{arg.name}<span className="ml-1 text-destructive">*</span></span>
+                  {arg.choices.length > 0 ? (
+                    <select value={value} onChange={(event) => onChange(event.target.value)} className="min-h-11 rounded-md border bg-background px-3 text-sm outline-none focus:ring-2 focus:ring-ring/50" autoFocus={selectedOpenCLI.requiredArgs[0] === arg.name}>
+                      <option value="">{arg.help ?? `${copy.input} ${arg.name}`}</option>
+                      {arg.choices.map((choice) => <option key={String(choice)} value={String(choice)}>{String(choice)}</option>)}
+                    </select>
+                  ) : (
+                    <input type={arg.type?.toLowerCase().includes("int") || arg.type?.toLowerCase().includes("float") ? "number" : "text"} value={value} onChange={(event) => onChange(event.target.value)} placeholder={arg.help ?? `${copy.input} ${arg.name}`} className="min-h-11 rounded-md border bg-background px-3 text-sm outline-none focus:ring-2 focus:ring-ring/50" autoFocus={selectedOpenCLI.requiredArgs[0] === arg.name} />
+                  )}
+                </label>
+              )
+            })}
           </div>
           <div className="flex justify-end gap-2 border-t p-4"><button type="button" className="min-h-10 rounded-md border px-4 text-xs" onClick={() => setSelectedOpenCLI(null)}>{copy.cancel}</button><button type="submit" className="min-h-10 rounded-md bg-primary px-4 text-xs text-primary-foreground disabled:opacity-50" disabled={missingRequired.length > 0}>{copy.addSource}</button></div>
         </form>

frontend/components/flow/inspector-shell.tsx

@@ -4,15 +4,15 @@ import { useRef, useState, type KeyboardEvent, type PointerEvent, type ReactNode
 import { LocateFixed, Pin, X } from "lucide-react"
 import { cn } from "@/lib/utils"
 
-const stateText: Record<string, string> = {
+export const workflowStatusText: Record<string, string> = {
   idle: "Idle",
   running: "Running",
   success: "Done",
   partial_success: "Partial success",
   error: "Error",
 }
 
-const stateDotClass: Record<string, string> = {
+export const workflowStatusDotClass: Record<string, string> = {
   idle: "border-muted-foreground/50 bg-transparent",
   running: "border-info bg-info",
   success: "border-success bg-success",
@@ -54,10 +54,10 @@ function PanelStatus({ status }: { status?: string }) {
   return (
     <span
       className="inline-flex shrink-0 items-center gap-1.5 text-muted-foreground"
-      title={`Status: ${stateText[status] ?? status}`}
+      title={`Status: ${workflowStatusText[status] ?? status}`}
     >
-      <span className={cn("size-1.5 rounded-full border", stateDotClass[status] ?? stateDotClass.idle)} />
-      <span>{stateText[status] ?? status}</span>
+      <span className={cn("size-1.5 rounded-full border", workflowStatusDotClass[status] ?? workflowStatusDotClass.idle)} />
+      <span>{workflowStatusText[status] ?? status}</span>
     </span>
   )
 }