starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 08:09:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: workflow history (#13588)

Browse Source 
This enhances the workflow history , stores all runs instead of just
failed ones and also some UX improvements on FE

---------

Signed-off-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: sai nikhil kethe <nikhil@openobserve.ai>
This commit is contained in:
Yashodhan Joshi
2026-08-04 08:22:01 +00:00
committed by GitHub
co-authored by Yashodhan Joshi sai nikhil kethe
parent cc759bb6e0
commit 6f0b2dcb07
30 changed files with 1417 additions and 505 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/management/src/request/workflows/mod.rs
+10 -2
View File
@@ -39,6 +39,7 @@ use openobserve_api_common::extractors::Headers;
use openobserve_core::auth::UserEmail;
use search_service::{self as SearchService, query_range::get_settings_max_query_range};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::{
common::{meta::http::HttpResponse as MetaHttpResponse, utils::http::get_or_create_trace_id},
@@ -56,6 +57,7 @@ pub struct WorkflowTestInput {
#[derive(Serialize)]
pub struct WorkflowTestResult {
errors: HashMap<String, NodeErrors>,
inputs: HashMap<String, Vec<Value>>,
}
#[derive(Serialize)]
@@ -387,7 +389,10 @@ pub async fn test_workflow(
workflow.org_id = org_id.clone();
workflow.id = format!("test-{}", config::ider::uuid());
match workflows::test_workflow(&org_id, workflow, inputs.inputs, inputs.from_node).await {
Ok(v) => MetaHttpResponse::json(WorkflowTestResult { errors: v.errors }),
Ok(v) => MetaHttpResponse::json(WorkflowTestResult {
errors: v.errors,
inputs: v.inputs,
}),
Err(e) => MetaHttpResponse::bad_request(e),
}
}
@@ -528,7 +533,10 @@ pub async fn retry_workflow(
Json(details): Json<WorkflowRetryDetails>,
) -> Response {
match workflows::retry_run(&org_id, &workflow_id, &details.run_id, details.from_node).await {
Ok(v) => MetaHttpResponse::json(WorkflowTestResult { errors: v.errors }),
Ok(v) => MetaHttpResponse::json(WorkflowTestResult {
errors: v.errors,
inputs: v.inputs,
}),
Err(e) => MetaHttpResponse::bad_request(e),
}
}
src/core/src/pipeline/batch_execution.rs
+59 -3
View File
@@ -70,6 +70,7 @@ struct BatchBuffer {
pub struct WorkflowResult {
pub stream_details: HashMap<StreamParams, Vec<(usize, Value)>>,
pub errors: HashMap<String, NodeErrors>,
pub inputs: HashMap<String, Vec<Value>>,
}
#[cfg(feature = "enterprise")]
@@ -564,6 +565,7 @@ impl ExecutablePipeline {
child_senders,
result_sender: result_sender_cp,
error_sender: error_sender_cp,
inputs_sender: None, // not applicable for pipelines
};
let task = tokio::spawn(process_node(metadata, node, function_runtime, channels));
node_tasks.push(task);
@@ -775,6 +777,9 @@ impl ExecutablePipeline {
let (error_sender, mut error_receiver) =
channel::<(String, String, String, Option<String>, Option<Value>)>(batch_size);
// inputs_channel
let (inputs_sender, mut inputs_receiver) = channel::<(String, Value)>(batch_size);
let mut node_senders = HashMap::new();
let mut node_receivers = HashMap::new();
@@ -798,6 +803,7 @@ impl ExecutablePipeline {
.collect();
let result_sender_cp = node.children.is_empty().then_some(result_sender.clone());
let error_sender_cp = error_sender.clone();
let inputs_sender_cp = inputs_sender.clone();
let function_runtime: Option<CompiledFunctionRuntime> =
self.function_map.get(node_id).cloned();
let pipeline_name = pipeline_name.clone();
@@ -837,6 +843,7 @@ impl ExecutablePipeline {
child_senders,
result_sender: result_sender_cp,
error_sender: error_sender_cp,
inputs_sender: Some(inputs_sender_cp),
};
let task = tokio::spawn(process_node(metadata, node, function_runtime, channels));
node_tasks.push(task);
@@ -884,6 +891,14 @@ impl ExecutablePipeline {
}
});
let inputs_task = tokio::spawn(async move {
let mut input_map = HashMap::new();
while let Some((node_id, val)) = inputs_receiver.recv().await {
input_map.entry(node_id).or_insert(vec![]).push(val);
}
input_map
});
// Send records to the source node to begin processing
let flattened = {
let source_node = self.node_map.get(&self.source_node_id).unwrap();
@@ -915,6 +930,7 @@ impl ExecutablePipeline {
drop(result_sender);
drop(error_sender);
drop(node_senders);
drop(inputs_sender);
log::debug!(
"[Workflow] {pipeline_name} [inv={inv_id}]: All records send into pipeline for processing"
);
@@ -939,6 +955,13 @@ impl ExecutablePipeline {
anyhow!("[Workflow] error collecting job failed: {}", e)
})?;
let inputs = inputs_task.await.map_err(|e| {
log::error!(
"[Workflow] {pipeline_name} [inv={inv_id}]: input collecting job failed: {e}"
);
anyhow!("[Workflow] input collecting job failed: {}", e)
})?;
let node_errors = errors
.as_ref()
.map(|v| v.node_errors.clone())
@@ -1005,6 +1028,7 @@ impl ExecutablePipeline {
Ok(WorkflowResult {
stream_details: results,
errors: node_errors,
inputs,
})
}
@@ -1152,6 +1176,24 @@ struct ProcessChannels {
child_senders: Vec<Sender<PipelineItem>>,
result_sender: Option<Sender<(usize, StreamParams, Value)>>,
error_sender: Sender<(String, String, String, Option<String>, Option<Value>)>,
inputs_sender: Option<Sender<(String, Value)>>,
}
impl ProcessChannels {
// TODO YJDoc2: eventually implement these fns for other channels as well, so we have
// a common place for error logging and handling, instead of repeating the same code
// in each of the node processing fns
async fn send_input(&mut self, metadata: &ProcessMetadata, id: &str, value: &Value) {
if let Some(ref mut channel) = self.inputs_sender
&& let Err(e) = channel.send((id.to_string(), value.clone())).await
{
log::error!(
"[Pipeline] {} [inv={}] error sending input via input channel for node {id} : {e}",
metadata.pipeline_name,
metadata.inv_id
);
}
}
}
async fn process_node(
@@ -1179,6 +1221,7 @@ async fn process_node(
NodeData::WorkflowTrigger => {
let mut count: usize = 0;
while let Some(item) = channels.receiver.recv().await {
channels.send_input(&metadata, &node.id, &item.record).await;
send_to_children(&mut channels.child_senders, item, "WorkflowTrigger").await;
count += 1;
}
@@ -1709,16 +1752,22 @@ async fn process_function_node(
) -> usize {
let mut count: usize = 0;
let cfg = config::get_config();
let inv_id = metadata.inv_id;
let inv_id = metadata.inv_id.clone();
log::debug!(
"[Pipeline] {} [inv={inv_id}]: func node {} starts processing",
metadata.pipeline_name,
metadata.node_idx
);
let mut vrl_runtime_state = crate::ingestion::init_functions_runtime();
let stream_name = metadata.stream_name.unwrap_or("pipeline".to_string());
let stream_name = metadata
.stream_name
.clone()
.unwrap_or("pipeline".to_string());
let mut result_array_records = Vec::new();
while let Some(pipeline_item) = channels.receiver.recv().await {
channels
.send_input(&metadata, &node.id, &pipeline_item.record)
.await;
let PipelineItem {
idx,
mut record,
@@ -2024,7 +2073,7 @@ async fn process_condition_node(
) -> usize {
let mut count: usize = 0;
let cfg = config::get_config();
let inv_id = metadata.inv_id;
let inv_id = metadata.inv_id.clone();
log::debug!(
"[Pipeline] {} [inv={inv_id}]: cond node {} starts processing",
metadata.pipeline_name,
@@ -2032,6 +2081,9 @@ async fn process_condition_node(
);
let mut total_received: usize = 0;
while let Some(pipeline_item) = channels.receiver.recv().await {
channels
.send_input(&metadata, &node.id, &pipeline_item.record)
.await;
total_received += 1;
let PipelineItem {
idx,
@@ -2344,6 +2396,9 @@ async fn process_destination_node(
let mut data = Vec::new();
while let Some(pipeline_item) = channels.receiver.recv().await {
channels
.send_input(&metadata, &node.id, &pipeline_item.record)
.await;
data.push(std::sync::Arc::new(pipeline_item.record));
}
@@ -2750,6 +2805,7 @@ mod tests {
child_senders: vec![child_tx],
result_sender: None,
error_sender: error_tx,
inputs_sender: None,
},
)
.await;
src/core/src/workflows/mod.rs
+56 -33
View File
@@ -41,8 +41,10 @@ use crate::{
pub mod runtime;
#[derive(Serialize, Deserialize)]
pub struct InputMap {
complete: Vec<Value>,
node_map: HashMap<String, Vec<Value>>,
// even though a bad naming convention, node_map is
error_node_map: HashMap<String, Vec<Value>>,
#[serde(default)]
input_map: HashMap<String, Vec<Value>>,
}
#[derive(Serialize, Deserialize)]
@@ -453,7 +455,6 @@ async fn execute_workflow(
let executable = ExecutablePipeline::new_from_workflow(&workflow).await?;
let now = chrono::Utc::now().timestamp_micros();
let input_copy = inputs.clone();
let res = executable.process_workflow(org_id, inputs, None).await?;
let mut errored_input_map = HashMap::new();
@@ -498,33 +499,41 @@ async fn execute_workflow(
}
}
if !workflow_errors.is_empty() {
let ip_map = InputMap {
complete: input_copy,
node_map: errored_input_map,
};
let ip_map = InputMap {
error_node_map: errored_input_map,
input_map: res.inputs,
};
let errors = WorkflowRunErrors {
org_id: org_id.to_string(),
cluster: config::get_cluster_name(),
id: 0, // will be set directly in db
workflow_id: id.to_string(),
run_id: run_id.to_string(),
ran_at: now,
data: workflow_errors,
input_data: Some(serde_json::to_string(&ip_map).unwrap()),
};
// workflow has already run, so not much point in returning error because
// we couldn't save the errors to db, log and ignore
if let Err(e) = db::workflows::save_workflow_errors(errors).await {
log::error!(
"[Workflows] : error saving workflow run errors for run id {run_id} for workflow {org_id}/{id} in db : {e}"
);
}
return Ok(WorkflowExecutionStatus::Errored);
// if this is not empty, then some node errored
let errored = !workflow_errors.is_empty();
// we hijack the workflow run errors to store both the error as well as execution history map
// as both have essentially the same structure and no point in duplicating tables and adding
// migration true error v/s just run can be distinguished by workflow_errors is empty or
// not.
let errors = WorkflowRunErrors {
org_id: org_id.to_string(),
cluster: config::get_cluster_name(),
id: 0, // will be set directly in db
workflow_id: id.to_string(),
run_id: run_id.to_string(),
ran_at: now,
data: workflow_errors,
input_data: Some(serde_json::to_string(&ip_map).unwrap()),
};
// workflow has already run, so not much point in returning error because
// we couldn't save the errors to db, log and ignore
if let Err(e) = db::workflows::save_workflow_errors(errors).await {
log::error!(
"[Workflows] : error saving workflow run errors for run id {run_id} for workflow {org_id}/{id} in db : {e}"
);
}
Ok(WorkflowExecutionStatus::Success)
if errored {
Ok(WorkflowExecutionStatus::Errored)
} else {
Ok(WorkflowExecutionStatus::Success)
}
}
pub async fn get_workflow_errors(
@@ -545,6 +554,21 @@ pub async fn retry_run(
let workflow = workflows::get_by_org_wid(org_id, wid)
.await?
.ok_or(anyhow::anyhow!("workflow with given id not found"))?;
let mut start_id = None;
for node in &workflow.nodes {
if matches!(node.data, NodeData::WorkflowTrigger) {
start_id = Some(node.id.clone());
break;
}
}
let Some(start_id) = start_id else {
log::error!(
"missing workflow trigger node in workflow {org_id}/{wid} for retry run {run_id}"
);
return Err(anyhow::anyhow!("workflow trigger node missing in workflow"));
};
let executable = ExecutablePipeline::new_from_workflow(&workflow).await?;
let errors = match workflows::list_errors_for_workflow_run(org_id, wid, run_id).await {
@@ -577,12 +601,11 @@ pub async fn retry_run(
anyhow::anyhow!("error deserializing inputs : {e}")
})?;
let inputs = match from_node.as_ref() {
Some(node) => ip_map.node_map.remove(node).ok_or(anyhow::anyhow!(
"node id {node} does not have any associated input data in the stored inputs"
))?,
None => ip_map.complete,
};
let node_id = from_node.as_ref().unwrap_or(&start_id);
let inputs = ip_map.input_map.remove(node_id).ok_or(anyhow::anyhow!(
"node id {node_id} does not have any associated input data in the stored inputs"
))?;
let res = executable
.process_workflow(org_id, inputs, from_node)
web/src/components/flow/forms/DestinationPicker.spec.ts
+10
View File
@@ -50,6 +50,7 @@ function createWrapper(props: Record<string, any> = {}) {
OSwitch: OSwitchStub,
CreateDestinationForm: {
name: "CreateDestinationForm",
props: ["forcedType"],
emits: ["created", "cancel"],
template: '<div class="create-destination-stub" />',
},
@@ -203,6 +204,15 @@ describe("DestinationPicker", () => {
});
});
it("forwards forcedType to the inline create form", async () => {
const wrapper = createWrapper({ forcedType: "custom" });
await flushPromises();
await wrapper.find(".o-switch").trigger("click"); // into create mode
expect(wrapper.findComponent({ name: "CreateDestinationForm" }).props("forcedType")).toBe(
"custom",
);
});
it("refetches when toggling out of create mode", async () => {
const wrapper = createWrapper();
await flushPromises();
web/src/components/flow/forms/DestinationPicker.vue
+5 -1
View File
@@ -41,6 +41,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
<!-- inline create destination form (own save/cancel) -->
<div v-if="createNewDestination" class="w-full">
<CreateDestinationForm
:forced-type="forcedType"
@created="onDestinationCreated"
@cancel="createNewDestination = false"
/>
@@ -76,8 +77,11 @@ import {
type ExternalDestinationForm,
} from "@/components/pipeline/NodeForm/ExternalDestination.schema";
const props = withDefaults(defineProps<{ initialName?: string }>(), {
// `forcedType`, when set, is forwarded to the inline create form to lock its
// destination type and skip the type-selection step (workflows → "custom").
const props = withDefaults(defineProps<{ initialName?: string; forcedType?: string }>(), {
initialName: "",
forcedType: undefined,
});
const emit = defineEmits<{ (e: "expand", value: boolean): void }>();
web/src/components/pipeline/NodeForm/CreateDestinationForm.spec.ts
+50
View File
@@ -541,6 +541,56 @@ describe("CreateDestinationForm", () => {
});
});
// ─────────────────────────────────────────────────────────────────────────────
describe("forcedType (skip step 1, locked type, no going back)", () => {
it("opens directly on step 2, locked to the forced type", () => {
const w = createWrapper({ forcedType: "custom" });
expect(w.vm.step).toBe(2);
expect(w.vm.destinationType).toBe("custom");
w.unmount();
});
it("seeds the create defaults for the forced type", () => {
const w = createWrapper({ forcedType: "custom" });
const v = w.vm.form.state.values;
expect(v.destination_type).toBe("custom");
expect(v.method).toBe("post");
expect(v.output_format).toBe("json");
expect(v.url_endpoint).toBe("");
expect(v.headers).toEqual([{ key: "", value: "" }]);
w.unmount();
});
it("hides the Back button (there's no step 1 to return to)", () => {
const w = createWrapper({ forcedType: "custom" });
expect(w.find('[data-test="step3-back-btn"]').exists()).toBe(false);
w.unmount();
});
it("hides the stepper header (single-step flow)", () => {
const forced = createWrapper({ forcedType: "custom" });
expect(forced.findComponent({ name: "OStepper" }).props("hideHeader")).toBe(true);
forced.unmount();
// ...and the default (no forcedType) keeps the header.
expect(wrapper.findComponent({ name: "OStepper" }).props("hideHeader")).toBe(false);
});
it("resetForm keeps it on step 2 when the type is forced", async () => {
const w = createWrapper({ forcedType: "custom" });
w.vm.resetForm();
await nextTick();
expect(w.vm.step).toBe(2);
expect(w.vm.form.state.values.destination_type).toBe("custom");
w.unmount();
});
it("without forcedType, the Back button and step 1 remain (unchanged default)", () => {
// default wrapper has no forcedType
expect(wrapper.vm.step).toBe(1);
expect(wrapper.find('[data-test="destination-type-card-openobserve"]').exists()).toBe(true);
});
});
// ─────────────────────────────────────────────────────────────────────────────
describe("canProceedStep2 Validation", () => {
it("is true for valid openobserve form data", async () => {
web/src/components/pipeline/NodeForm/CreateDestinationForm.vue
+43 -25
View File
@@ -28,9 +28,18 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
footer below stays full-width so its top border lines up with the
full-width page header (matching the other destination forms). -->
<div class="w-full max-w-[50vw]">
<OStepper v-model="step" ref="stepper" animated>
<!-- Step 1: Choose Destination Type -->
<OStep :name="1" title="Choose Type" icon="edit" :done="step > 1" :navigable="step > 1">
<!-- When the type is forced (step 1 skipped), the stepper is effectively a
single step, so hide its header — the progress bar would just be noise. -->
<OStepper v-model="step" ref="stepper" animated :hide-header="!!forcedType">
<!-- Step 1: Choose Destination Type. When the type is forced, it stays in
the header as a done step but is NOT navigable — there's no going back. -->
<OStep
:name="1"
title="Choose Type"
icon="edit"
:done="step > 1"
:navigable="forcedType ? false : step > 1"
>
<div class="mb-3 text-sm font-medium">
Select Destination Type <span class="text-status-error-text">*</span>
</div>
@@ -382,6 +391,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</div>
<div v-if="step > 1" class="flex gap-2">
<OButton
v-if="!forcedType"
data-test="step3-back-btn"
variant="outline"
size="sm-action"
@@ -438,6 +448,11 @@ import { makeDestinationSchema, type DestinationForm } from "./CreateDestination
// Props
const props = defineProps<{
destination?: DestinationData | null;
// When set, the type-selection step (step 1) is skipped: the form opens directly
// on the Connection step locked to this destination type, and there's no way back
// to step 1. Used by the workflow destination node, which only supports Custom
// HTTP destinations for now.
forcedType?: string;
}>();
const emit = defineEmits(["created", "updated", "cancel"]);
@@ -506,7 +521,9 @@ const destinationTypes = [
},
];
const step = ref(1);
// Skip straight to the Connection step when the type is forced (step 1 is the type
// picker, which is bypassed).
const step = ref(props.forcedType ? 2 : 1);
// A single Headers row. Matches the schema's `headerRowSchema` ({ key, value }).
// No `uuid` — the dynamic array-field keys rows by index and add/remove operate
@@ -588,26 +605,27 @@ const endpointForType = (type: string): string => {
// defaulting) run in a single `{ flush: "sync" }` watch on the form's
// destination_type. Edit-prefill seeds the form via `form.reset(record)`.
// Build the create-mode default record. OpenObserve is the default type, so seed
// the OpenObserve endpoint + default headers.
const buildCreateDefaults = (): DestinationForm => ({
name: "",
url: "",
skip_tls_verify: false,
headers: getDefaultHeaders("openobserve").map((h) => ({
key: h.key,
value: h.value,
})),
metadata: {},
destination_type: "openobserve",
url_endpoint: "/api/default/default/_json", // Default endpoint for OpenObserve
method: "post",
output_format: "json",
esbulk_index: "",
separator: "",
org: "default",
stream: "default",
});
// Build the create-mode default record. Defaults to OpenObserve, unless a
// `forcedType` is supplied (workflow node → "custom"), in which case the record is
// seeded for that type instead — so the skipped step-1 selection is still applied.
const buildCreateDefaults = (): DestinationForm => {
const type = props.forcedType ?? "openobserve";
return {
name: "",
url: "",
skip_tls_verify: false,
headers: getDefaultHeaders(type).map((h) => ({ key: h.key, value: h.value })),
metadata: {},
destination_type: type,
url_endpoint: type === "openobserve" ? "/api/default/default/_json" : endpointForType(type),
method: "post",
output_format: type === "splunk" ? "nestedevent" : type === "elasticsearch" ? "esbulk" : "json",
esbulk_index: type === "elasticsearch" ? "default" : "",
separator: "",
org: "default",
stream: "default",
};
};
const form = useOForm<DestinationForm>({
defaultValues: buildCreateDefaults(),
@@ -1200,7 +1218,7 @@ const deleteApiHeader = (index: number) => {
// field, including the default headers.
const resetForm = () => {
form.reset(buildCreateDefaults());
step.value = 1;
step.value = props.forcedType ? 2 : 1;
};
// Expose functions for testing. `form` is the single source of truth — tests
web/src/components/workflows/WorkflowEditor.spec.ts
+5 -6
View File
@@ -256,7 +256,7 @@ describe("WorkflowEditor", () => {
expect(workflowObj.dialog.show).toBe(false);
});
it("adds the picked trigger to the canvas and opens its panel", async () => {
it("adds the picked trigger to the canvas WITHOUT auto-opening its panel", async () => {
wrapper = mountEditor();
await flushPromises();
@@ -274,17 +274,16 @@ describe("WorkflowEditor", () => {
.vm.$emit("pick", { key: "workflow_trigger", trigger_kind: "alert_fired" });
await nextTick();
// Committed straight onto the canvas — the trigger panel is read-only and
// has no Save, so a staged node could never be committed.
// Committed straight onto the canvas.
expect(wf().nodes).toHaveLength(1);
expect(wf().nodes[0]).toMatchObject({
type: "input",
position: { x: 100, y: 40 },
data: { node_type: "workflow_trigger", trigger_kind: "alert_fired", alert_ids: [] },
});
// ...and its panel opens in EDIT mode, so closing it keeps the node.
expect(workflowObj.isEditNode).toBe(true);
expect(workflowObj.dialog.show).toBe(true);
// ...but its read-only detail panel does NOT auto-open — placing the trigger
// shouldn't interrupt the flow. The user clicks the node to open it.
expect(workflowObj.dialog.show).toBe(false);
expect(workflowObj.stepPicker.show).toBe(false);
});
web/src/components/workflows/WorkflowRuns.spec.ts
+20 -3
View File
@@ -75,7 +75,7 @@ const OButtonStub = {
name: "OButton",
props: ["variant"],
emits: ["click"],
template: `<button class="o-btn" @click="$emit('click')"><slot /></button>`,
template: `<button class="o-btn" v-bind="$attrs" @click="$emit('click')"><slot /></button>`,
};
const RunsPanelStub = {
name: "WorkflowRunsPanel",
@@ -93,6 +93,7 @@ const globalCfg = {
WorkflowRunsPanel: RunsPanelStub,
WorkflowCanvas: stub("WorkflowCanvas"),
WorkflowStepResultDrawer: stub("WorkflowStepResultDrawer"),
WorkflowTestDialog: stub("WorkflowTestDialog"),
},
};
@@ -190,15 +191,31 @@ describe("WorkflowRuns", () => {
expect(panel(wrapper).props("selectedRunId")).toBe("");
});
it("Edit Workflow navigates to the editor (edit stays separate from inspect)", async () => {
it("Edit Workflow navigates to the editor (build stays separate from inspect)", async () => {
const wrapper = await mountRuns();
await wrapper.find(".o-btn").trigger("click");
await wrapper.find('[data-test="workflow-runs-edit"]').trigger("click");
expect(mockRouter.push).toHaveBeenCalledWith({
name: "workflowEditor",
query: { id: "wf-1", name: "my flow", org_identifier: "default" },
});
});
it("Test opens the dry-run dialog here and deselects the historical run", async () => {
const wrapper = await mountRuns();
// Select a run first, so we can prove Test deselects it.
panel(wrapper).vm.$emit("select-run", "run-5");
await flushPromises();
expect(panel(wrapper).props("selectedRunId")).toBe("run-5");
workflowObj.testRun.show = false;
await wrapper.find('[data-test="workflow-runs-test"]').trigger("click");
expect(workflowObj.testRun.show).toBe(true);
expect(panel(wrapper).props("selectedRunId")).toBe("");
// Test never navigates away — it stays on the runs page.
expect(mockRouter.push).not.toHaveBeenCalled();
});
it("back returns to the workflows list", async () => {
const wrapper = await mountRuns();
await wrapper.find(".back-btn").trigger("click");
web/src/components/workflows/WorkflowRuns.vue
+38 -5
View File
@@ -27,8 +27,10 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
└──────────────────────────────┴──────────────────────┘
Selecting a run loads it onto the canvas (read-only) and updates ?run_id, so a
specific run is shareable. "Edit Workflow" is the deliberate switch to the
editor — the two surfaces never blur together, and nothing resizes mid-flow.
specific run is shareable. "Edit Workflow" is the deliberate switch to the build
canvas. Test lives here too — it's a read-only dry-run (never edits the graph),
so there's no reason to force a trip to the editor just to run one; a test simply
swaps the canvas from the selected historical run to the fresh test result.
-->
<template>
<div data-test="workflow-runs-page" class="flex h-full min-h-0 flex-col">
@@ -50,6 +52,11 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</span>
</template>
<template #actions>
<!-- Test is a read-only dry-run (it never edits the workflow), so it's
offered here too — no need to switch to the editor just to test. -->
<OButton variant="outline" data-test="workflow-runs-test" @click="onTest">
{{ t("workflow.test.button") }}
</OButton>
<OButton variant="outline" data-test="workflow-runs-edit" @click="onEditWorkflow">
{{ t("workflow.runs.edit") }}
</OButton>
@@ -76,14 +83,19 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</div>
</div>
<!-- Per-step Input/Output result drawer — opened by clicking an error node's
badge. Read-only (no Replay on a historical run). -->
<!-- Test input popup — a fresh dry-run of the current graph, launched from the
header. Results paint on the read-only canvas (switching it out of the
selected historical run). -->
<WorkflowTestDialog v-if="workflowObj.testRun.show" />
<!-- Per-step Input/Output result drawer — opened by clicking a node's badge.
Read-only for a historical run; editable + replayable for a live test. -->
<WorkflowStepResultDrawer v-if="workflowObj.testRun.resultDrawer.show" />
</div>
</template>
<script setup lang="ts">
import { computed, onBeforeUnmount, onMounted, ref } from "vue";
import { computed, onBeforeUnmount, onMounted, ref, watch } from "vue";
import { useI18n } from "vue-i18n";
import { useRouter } from "vue-router";
import { useStore } from "vuex";
@@ -95,6 +107,7 @@ import { toast } from "@/lib/feedback/Toast/useToast";
import WorkflowCanvas from "@/plugins/workflows/WorkflowCanvas.vue";
import WorkflowStepResultDrawer from "./WorkflowStepResultDrawer.vue";
import WorkflowTestDialog from "./WorkflowTestDialog.vue";
import WorkflowRunsPanel from "./WorkflowRunsPanel.vue";
import useWorkflowCanvas, {
workflowObj,
@@ -125,6 +138,26 @@ const goBack = () => {
router.push({ name: "workflows", query: { org_identifier: orgId.value } });
};
// Dry-run the current graph without leaving to the editor. Deselect the historical
// run first — the canvas is about to show the LIVE test result instead, and a
// lingering row highlight would misrepresent what's on the canvas. Clicking a run
// in the list again reloads history, toggling back.
const onTest = () => {
selectedRunId.value = "";
workflowObj.testRun.show = true;
};
// A live test result (from the header Test button, or a step drawer's "Use as Test
// Input" which drops the user into a fresh test) replaces the historical run on the
// canvas. Deselect the run so the list stops highlighting a row that no longer
// matches what's shown. A history load keeps `mode: "history"`, so it's left alone.
watch(
() => workflowObj.testRun.result,
(result: any) => {
if (result && result.mode !== "history") selectedRunId.value = "";
},
);
// Deliberate switch to the editor — the only bridge between inspect and build.
const onEditWorkflow = () => {
router.push({
web/src/components/workflows/WorkflowRunsPanel.spec.ts
+25 -1
View File
@@ -68,6 +68,22 @@ const stub = (name: string, props: string[] = []) => ({
template: `<div class="${name}" />`,
});
// The real DateTime picker emits `on:date-change` on mount with the initial range —
// the panel's FIRST fetch rides on that emit (NOT an immediate watch, which would
// double-fetch). Mirror it so the mount-fetch fires in tests.
const DateTimeStub = {
name: "DateTime",
emits: ["on:date-change"],
template: `<div class="DateTime" />`,
mounted() {
(this as any).$emit("on:date-change", {
startTime: 1000,
endTime: 2000,
relativeTimePeriod: "15m",
});
},
};
import { describe, it, expect, beforeEach } from "vitest";
import { mount, flushPromises } from "@vue/test-utils";
import i18n from "@/locales";
@@ -82,7 +98,7 @@ const globalCfg = {
OBadge: stub("OBadge", ["variant", "size"]),
OButton: stub("OButton", ["variant", "size", "iconLeft", "loading"]),
OTooltip: stub("OTooltip", ["side", "content"]),
DateTime: stub("DateTime"),
DateTime: DateTimeStub,
WorkflowExecutionTimeline: stub("WorkflowExecutionTimeline", [
"history",
"firingLabel",
@@ -111,6 +127,14 @@ describe("WorkflowRunsPanel", () => {
expect(wrapper.findComponent(OTableStub as any).exists()).toBe(true);
});
it("fetches history EXACTLY ONCE on mount (no immediate-watch double-fetch)", async () => {
mockList.mockResolvedValue({ data: [] });
mountPanel();
await flushPromises();
// Only the DateTime picker's on-mount emit fetches — not a second immediate watch.
expect(mockList).toHaveBeenCalledTimes(1);
});
it("emits select-run with the clicked run's id", async () => {
const wrapper = mountPanel();
await flushPromises();
web/src/components/workflows/WorkflowRunsPanel.vue
+5 -4
View File
@@ -343,13 +343,14 @@ const openRun = (row: any) => {
emit("select-run", row.run_id);
};
// Fetch on mount and whenever the workflow changes.
// Refetch when the workflow changes. The INITIAL fetch is NOT triggered here — the
// DateTime picker emits `@on:date-change` on mount (→ updateDateTime → fetchHistory)
// with the actual displayed range, so an `immediate` watch would double-fetch.
watch(
() => props.workflowId,
(id) => {
if (id) fetchHistory();
(id, prev) => {
if (id && id !== prev) fetchHistory();
},
{ immediate: true },
);
defineExpose({ fetchHistory });
web/src/components/workflows/WorkflowStepResultDrawer.spec.ts
+362 -264
View File
@@ -13,11 +13,13 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
// WorkflowStepResultDrawer — the "Test Step Result" drawer. By design it only
// opens for ERROR nodes: Input = the CENTRAL testRun.input (same payload the
// Test dialog edits), Output = the NodeErrors messages (the backend returns no
// per-node node_io), plus Replay (executeTestRun with fromNode). A history run
// is read-only: per-node captured input, no Replay.
// WorkflowStepResultDrawer — the "Test Step Result" drawer. Opens for ANY node
// after a Test run. Input = the records THIS node received (from the backend
// per-node `inputs` map), editable + replayable. Output = what it emitted, derived
// per outgoing edge (child input == parent output on a single-incoming tree),
// rendered in a read-only editor. Errored nodes show the error AND what they still
// forwarded. Status badge = Passed / Errored / No Records. "Use as Test Input"
// promotes a pane's records to the whole-flow payload. History runs are read-only.
import { describe, it, expect, beforeEach, vi } from "vitest";
import { mount, flushPromises } from "@vue/test-utils";
@@ -25,6 +27,8 @@ import { nextTick } from "vue";
import i18n from "@/locales";
import store from "@/test/unit/helpers/store";
const t = (k: string, p?: any) => i18n.global.t(k, p);
const mockToast = vi.fn();
vi.mock("@/lib/feedback/Toast/useToast", () => ({
toast: (...a: any[]) => mockToast(...a),
@@ -65,7 +69,6 @@ const ODrawerStub = {
};
const OButtonStub = {
name: "OButton",
// `title` is intentionally NOT declared so it falls through to the button.
props: ["variant", "size", "disabled", "loading"],
template: `<button :disabled="disabled" :data-loading="String(!!loading)"><slot /></button>`,
};
@@ -88,7 +91,7 @@ const CodeQueryEditorStub = {
name: "CodeQueryEditor",
props: ["editorId", "language", "query", "readOnly", "showAutoComplete"],
emits: ["update:query"],
template: `<div class="code-editor" :data-readonly="String(!!readOnly)">{{ query }}</div>`,
template: `<div class="code-editor" :data-editor="editorId" :data-readonly="String(!!readOnly)">{{ query }}</div>`,
};
const globalConfig = {
@@ -103,45 +106,50 @@ const globalConfig = {
},
};
// trigger(t1) -> function(f1) -> destination(d1)
const NODES = [
{ id: "t1", data: { node_type: "workflow_trigger" } },
{ id: "f1", data: { node_type: "function", name: "parse_json" } },
{ id: "d1", data: { node_type: "destination", destination_id: "sink-a" } },
];
const EDGES = [
{ source: "t1", target: "f1" },
{ source: "f1", target: "d1" },
];
const VALID_INPUT = '[{"meta":{"alert_name":"a"},"data":[{"x":1}]}]';
// A trigger-shape record (nested { meta, data }).
const rec = (x: number) => ({ meta: { alert_name: "a" }, data: [{ x }] });
const errorResult = (extra: Record<string, any> = {}) => ({
errors: {
f1: { error_count: 2, errors: [["boom", { x: 1 }], ["bad record"]] },
},
// Every node received records; nothing errored.
const okResult = () => ({
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)], d1: [rec(3)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
});
// f1 errored but still forwarded records to d1.
const errorResult = () => ({
errors: { f1: { error_count: 2, errors: [["boom", [{ x: 1 }]], ["bad record"]] } },
inputs: { t1: [rec(1)], f1: [rec(2)], d1: [rec(3)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: ["d1"],
...extra,
});
const setup = (
opts: {
nodeId?: string;
input?: string;
result?: any;
nodes?: any[];
} = {},
opts: { nodeId?: string; input?: string; result?: any; nodes?: any[]; edges?: any[] } = {},
) => {
workflowObj.currentSelectedWorkflow = {
id: "wf1",
name: "wf",
nodes: opts.nodes ?? NODES,
edges: [
{ source: "t1", target: "f1" },
{ source: "f1", target: "d1" },
],
edges: opts.edges ?? EDGES,
} as any;
workflowObj.testRun = {
show: false,
input: opts.input ?? VALID_INPUT,
input: opts.input ?? "",
fromNode: "",
result: opts.result === undefined ? errorResult() : opts.result,
result: opts.result === undefined ? okResult() : opts.result,
resultDrawer: { show: true, nodeId: opts.nodeId ?? "f1" },
} as any;
};
@@ -149,9 +157,17 @@ const setup = (
const mountDrawer = () =>
mount(WorkflowStepResultDrawer, { global: globalConfig, attachTo: document.body });
const editorVm = (w: any) => w.findComponent(CodeQueryEditorStub as any);
const editors = (w: any) => w.findAllComponents(CodeQueryEditorStub as any);
const inputEditor = (w: any) =>
editors(w).find((e: any) => e.props("editorId") === "workflow-step-input");
const outputEditor = (w: any) =>
editors(w).find((e: any) => e.props("editorId") === "workflow-step-output");
const statusBadge = (w: any) => w.find('[data-test="workflow-step-result-status"]');
const replayBtn = (w: any) => w.find('[data-test="workflow-step-replay-btn"]');
const iconButtons = (w: any) => w.findAll(".o-drawer button");
const useInputBtn = (w: any) => w.find('[data-test="workflow-step-use-input-as-test"]');
const useOutputBtn = (w: any) => w.find('[data-test="workflow-step-use-output-as-test"]');
const tooltipFor = (w: any, content: string) =>
w.findAll(".o-tooltip").find((tt: any) => tt.attributes("data-content") === content);
describe("WorkflowStepResultDrawer", () => {
beforeEach(() => {
@@ -160,10 +176,9 @@ describe("WorkflowStepResultDrawer", () => {
});
describe("title", () => {
it("shows '<Type> - <detail>' for the errored node", () => {
const wrapper = mountDrawer();
expect(wrapper.find(".o-drawer").attributes("data-title")).toBe(
`${i18n.global.t("workflow.node.function")} - parse_json`,
it("shows '<Type> - <detail>' for the node", () => {
expect(mountDrawer().find(".o-drawer").attributes("data-title")).toBe(
`${t("workflow.node.function")} - parse_json`,
);
});
@@ -173,212 +188,307 @@ describe("WorkflowStepResultDrawer", () => {
nodes: [
{
id: "f1",
data: {
node_type: "function",
name: "a_really_long_vrl_function_name_here",
},
data: { node_type: "function", name: "a_really_long_vrl_function_name_here" },
},
],
});
const wrapper = mountDrawer();
const title = wrapper.find(".o-drawer").attributes("data-title")!;
expect(title).toHaveLength(31); // 30 chars + the ellipsis
const title = mountDrawer().find(".o-drawer").attributes("data-title")!;
expect(title).toHaveLength(31);
expect(title.endsWith("…")).toBe(true);
});
it("falls back to the raw node_type when it is not in the palette", () => {
setup({ nodeId: "x1", nodes: [{ id: "x1", data: { node_type: "mystery" } }] });
const wrapper = mountDrawer();
expect(wrapper.find(".o-drawer").attributes("data-title")).toBe("mystery");
setup({ nodeId: "x1", nodes: [{ id: "x1", data: { node_type: "mystery" } }], edges: [] });
expect(mountDrawer().find(".o-drawer").attributes("data-title")).toBe("mystery");
});
it("renders an empty title when the node id no longer exists", () => {
setup({ nodeId: "ghost" });
const wrapper = mountDrawer();
expect(wrapper.find(".o-drawer").attributes("data-title")).toBe("");
expect(mountDrawer().find(".o-drawer").attributes("data-title")).toBe("");
});
});
describe("status + output (errors only, by design)", () => {
it("always shows the Errored badge (the drawer only opens for error nodes)", () => {
const wrapper = mountDrawer();
const badge = wrapper.find(".o-badge");
describe("status badge", () => {
it("Passed (success) when records reached the node", () => {
const badge = statusBadge(mountDrawer());
expect(badge.attributes("data-variant")).toBe("success-soft");
expect(badge.text()).toBe(t("workflow.test.stepResult.status.ok"));
});
it("Errored (error) when the node is in the errors map", () => {
setup({ result: errorResult() });
const badge = statusBadge(mountDrawer());
expect(badge.attributes("data-variant")).toBe("error-soft");
expect(badge.text()).toBe(i18n.global.t("workflow.test.stepResult.status.error"));
expect(badge.text()).toBe(t("workflow.test.stepResult.status.error"));
});
it("renders each NodeErrors message as the Output — first tuple element only", () => {
const wrapper = mountDrawer();
const lines = wrapper.findAll("[data-test='workflow-step-result-error-line']");
expect(lines).toHaveLength(2);
expect(lines[0].text()).toBe("boom");
expect(lines[1].text()).toBe("bad record");
});
it("stringifies a non-tuple error entry", () => {
it("No Records (default) when the node ran but got nothing (absent from inputs)", () => {
// d1 reachable but not in inputs -> the condition/function above filtered all.
setup({
nodeId: "d1",
result: {
errors: { f1: { error_count: 1, errors: ["flat message"] } },
ranNodeIds: [],
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
const badge = statusBadge(mountDrawer());
expect(badge.attributes("data-variant")).toBe("default-soft");
expect(badge.text()).toBe(t("workflow.test.stepResult.status.skipped"));
});
});
describe("input pane — the records this node received", () => {
it("shows the node's own input, editable, in the editor", () => {
const editor = inputEditor(mountDrawer());
expect(editor.props("readOnly")).toBe(false);
expect(JSON.parse(editor.props("query"))).toEqual([rec(2)]); // f1's input
});
it("seeds a DIFFERENT node's input (per-node, not a shared sample)", () => {
setup({ nodeId: "t1" });
expect(JSON.parse(inputEditor(mountDrawer()).props("query"))).toEqual([rec(1)]);
});
it("shows an empty state (not a blank editor) when 0 records reached the node", () => {
setup({
nodeId: "d1",
result: {
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
const wrapper = mountDrawer();
expect(wrapper.find("[data-test='workflow-step-result-error-line']").text()).toBe(
"flat message",
expect(inputEditor(wrapper)).toBeUndefined(); // no input editor rendered
expect(wrapper.find('[data-test="workflow-step-result-no-input"]').text()).toBe(
t("workflow.test.stepResult.noInput"),
);
});
it("shows the no-output placeholder when the node has no error entries", () => {
setup({ result: { errors: {}, ranNodeIds: [], blockedNodeIds: [] } });
const wrapper = mountDrawer();
expect(wrapper.find("[data-test='workflow-step-result-error-line']").exists()).toBe(false);
expect(wrapper.find("[data-test='workflow-step-result-no-output']").text()).toBe(
i18n.global.t("workflow.test.stepResult.noOutput"),
);
});
it("tolerates a null result and a malformed errors entry", () => {
setup({ result: null });
expect(mountDrawer().find("[data-test='workflow-step-result-no-output']").exists()).toBe(
true,
);
it("unwraps a JSON-stringified `data` field for readability", () => {
const flat = { meta_alert_name: "a", data: '[{"x":9}]' };
setup({
result: { errors: { f1: { errors: "not-an-array" } } },
nodeId: "d1",
result: { errors: {}, inputs: { d1: [flat] }, ranNodeIds: ["d1"], blockedNodeIds: [] },
nodes: [{ id: "d1", data: { node_type: "destination", destination_id: "sink-a" } }],
edges: [],
});
expect(mountDrawer().find("[data-test='workflow-step-result-no-output']").exists()).toBe(
true,
);
const parsed = JSON.parse(inputEditor(mountDrawer()).props("query"));
expect(parsed[0].data).toEqual([{ x: 9 }]); // parsed, not the escaped string
});
});
describe("input — bound to the central testRun.input (test mode)", () => {
it("shows the central input, editable", () => {
describe("output pane — what the node emitted (derived)", () => {
it("single downstream: the records the child received, in a read-only editor", () => {
const editor = outputEditor(mountDrawer());
expect(editor.props("readOnly")).toBe(true);
expect(JSON.parse(editor.props("query"))).toEqual([rec(3)]); // f1's output == d1's input
});
it("single downstream: NO caption (the target is obvious from the canvas)", () => {
// f1 -> d1 (single). No "→ Destination · sink-a" line, no record count — just
// the records in the editor.
const wrapper = mountDrawer();
expect(editorVm(wrapper).props("query")).toBe(VALID_INPUT);
expect(editorVm(wrapper).props("readOnly")).toBe(false);
});
it("writes edits straight back to testRun.input (stays in sync with the Test dialog)", async () => {
const wrapper = mountDrawer();
editorVm(wrapper).vm.$emit("update:query", '[{"y":2}]');
await nextTick();
expect(workflowObj.testRun.input).toBe('[{"y":2}]');
expect(editorVm(wrapper).props("query")).toBe('[{"y":2}]');
});
it("shows the invalid-JSON hint for a malformed payload", async () => {
setup({ input: "{oops" });
const wrapper = mountDrawer();
expect(wrapper.find("[data-test='workflow-step-result-input-error']").text()).toBe(
i18n.global.t("workflow.test.invalidJson"),
);
});
it("treats a non-array payload as invalid", () => {
setup({ input: '{"a":1}' });
expect(mountDrawer().find("[data-test='workflow-step-result-input-error']").exists()).toBe(
true,
);
});
it("treats an empty payload as invalid (nothing to replay)", () => {
setup({ input: " " });
const wrapper = mountDrawer();
expect(wrapper.find("[data-test='workflow-step-result-input-error']").exists()).toBe(true);
expect(replayBtn(wrapper).attributes("disabled")).toBeDefined();
});
it("hides the invalid-JSON hint for a valid array", () => {
expect(mountDrawer().find("[data-test='workflow-step-result-input-error']").exists()).toBe(
expect(wrapper.find('[data-test="workflow-step-result-output-branch-label"]').exists()).toBe(
false,
);
expect(outputEditor(wrapper)).toBeTruthy();
});
it("terminal (destination): shows the records it sent, with the 'sent externally' caption", () => {
setup({ nodeId: "d1" });
const wrapper = mountDrawer();
expect(JSON.parse(outputEditor(wrapper).props("query"))).toEqual([rec(3)]);
expect(wrapper.text()).toContain(t("workflow.test.stepResult.sentExternally"));
});
it("fan-out: one object keyed by '→ target' so each branch stays labelled", () => {
// t1 -> f1 and t1 -> d1 (broadcast). Inspect t1's output.
setup({
nodeId: "t1",
edges: [
{ source: "t1", target: "f1" },
{ source: "t1", target: "d1" },
],
result: {
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)], d1: [rec(3)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
const wrapper = mountDrawer();
const out = JSON.parse(outputEditor(wrapper).props("query"));
const keys = Object.keys(out);
expect(keys).toHaveLength(2);
expect(keys.every((k) => k.startsWith("→ "))).toBe(true);
// fan-out DOES keep a caption per branch (to tell them apart)
expect(
wrapper.findAll('[data-test="workflow-step-result-output-branch-label"]'),
).toHaveLength(2);
});
it("errored node: shows the error message(s) AND the records it forwarded", () => {
setup({ result: errorResult() });
const wrapper = mountDrawer();
const lines = wrapper.findAll('[data-test="workflow-step-result-error-line"]');
expect(lines.map((l: any) => l.text())).toEqual(["boom", "bad record"]);
// forwarded editor still present (f1 passed records to d1)
expect(JSON.parse(outputEditor(wrapper).props("query"))).toEqual([rec(3)]);
// both section headings shown
expect(wrapper.text()).toContain(t("workflow.test.stepResult.errorHeading"));
expect(wrapper.text()).toContain(t("workflow.test.stepResult.forwardedHeading"));
});
it("errored terminal: shows the error alone (a failed send has no output)", () => {
setup({
nodeId: "d1",
result: {
errors: { d1: { error_count: 1, errors: [["send failed"]] } },
inputs: { d1: [rec(3)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
const wrapper = mountDrawer();
expect(wrapper.find('[data-test="workflow-step-result-error-line"]').text()).toBe(
"send failed",
);
expect(outputEditor(wrapper)).toBeUndefined();
});
it("filtered non-terminal: centered empty state, no editor", () => {
// f1 forwarded nothing to d1 -> f1's output is empty.
setup({
result: {
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
const wrapper = mountDrawer();
expect(outputEditor(wrapper)).toBeUndefined();
expect(wrapper.find('[data-test="workflow-step-result-output-empty"]').exists()).toBe(true);
});
it("condition that matched nothing explains why in the empty state", () => {
setup({
nodeId: "c1",
nodes: [
{ id: "c1", data: { node_type: "condition" } },
{ id: "d1", data: { node_type: "destination", destination_id: "s" } },
],
edges: [{ source: "c1", target: "d1" }],
result: {
errors: {},
inputs: { c1: [rec(1)] },
ranNodeIds: ["c1", "d1"],
blockedNodeIds: [],
},
});
expect(mountDrawer().find('[data-test="workflow-step-result-output-empty"]').text()).toBe(
t("workflow.test.stepResult.conditionNoMatch"),
);
});
it("terminal with 0 records: 'nothing was sent', not a false send", () => {
setup({
nodeId: "d1",
result: {
errors: {},
inputs: { t1: [rec(1)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
expect(mountDrawer().find('[data-test="workflow-step-result-output-empty"]').text()).toBe(
t("workflow.test.stepResult.destinationNoRecords"),
);
});
});
describe("history mode (read-only past run)", () => {
const historyResult = () =>
errorResult({
mode: "history",
runId: "r1",
nodeInputs: { f1: [{ meta: { alert_name: "a" }, data: [{ x: 1 }] }] },
describe("use as test input", () => {
it("Input: promotes the records to the flow payload and opens the Test dialog", async () => {
const wrapper = mountDrawer();
expect(useInputBtn(wrapper).attributes("disabled")).toBeUndefined();
await useInputBtn(wrapper).trigger("click");
expect(JSON.parse(workflowObj.testRun.input)).toEqual([rec(2)]); // f1's input
expect(workflowObj.testRun.show).toBe(true);
expect(workflowObj.testRun.resultDrawer.show).toBe(false);
});
it("Output: promotes what the node emitted", async () => {
const wrapper = mountDrawer();
await useOutputBtn(wrapper).trigger("click");
expect(JSON.parse(workflowObj.testRun.input)).toEqual([rec(3)]); // f1's output
expect(workflowObj.testRun.show).toBe(true);
});
it("disables + explains when the pane has no records", () => {
// d1 got nothing -> both input and output empty.
setup({
nodeId: "d1",
result: {
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
it("shows the per-node captured input, pretty-printed and read-only", () => {
setup({ result: historyResult() });
const wrapper = mountDrawer();
const editor = editorVm(wrapper);
expect(editor.props("readOnly")).toBe(true);
expect(editor.props("query")).toBe(
JSON.stringify([{ meta: { alert_name: "a" }, data: [{ x: 1 }] }], null, 2),
);
expect(useInputBtn(wrapper).attributes("disabled")).toBeDefined();
expect(useOutputBtn(wrapper).attributes("disabled")).toBeDefined();
expect(tooltipFor(wrapper, t("workflow.test.stepResult.useAsTestInputNoInput"))).toBeTruthy();
expect(
tooltipFor(wrapper, t("workflow.test.stepResult.useAsTestInputNoOutput")),
).toBeTruthy();
});
it("shows an empty input when the run captured none for this node", () => {
setup({ result: errorResult({ mode: "history", nodeInputs: {} }) });
expect(editorVm(mountDrawer()).props("query")).toBe("");
});
it("ignores editor writes (the setter no-ops) so the central input is untouched", async () => {
setup({ result: historyResult() });
it("is ALSO available in history mode — it drops the user into a fresh test", async () => {
setup({ result: okResult(), input: "" });
workflowObj.testRun.result = { ...okResult(), mode: "history" } as any;
workflowObj.testRun.show = false;
const wrapper = mountDrawer();
editorVm(wrapper).vm.$emit("update:query", "tampered");
await nextTick();
expect(workflowObj.testRun.input).toBe(VALID_INPUT);
});
expect(useInputBtn(wrapper).exists()).toBe(true);
expect(useOutputBtn(wrapper).exists()).toBe(true);
it("hides Replay and the invalid-JSON hint (a past run is read-only)", () => {
setup({ input: "{oops", result: historyResult() });
const wrapper = mountDrawer();
expect(replayBtn(wrapper).exists()).toBe(false);
expect(wrapper.find("[data-test='workflow-step-result-input-error']").exists()).toBe(false);
});
it("still shows the error output for the past run", () => {
setup({ result: historyResult() });
expect(mountDrawer().findAll("[data-test='workflow-step-result-error-line']")).toHaveLength(
2,
);
await useInputBtn(wrapper).trigger("click");
// seeds the flow payload with the historical node's input and opens the Test dialog
expect(JSON.parse(workflowObj.testRun.input)).toEqual([rec(2)]);
expect(workflowObj.testRun.show).toBe(true);
});
});
describe("replay", () => {
it("re-runs the workflow FROM this node with the edited input, then closes", async () => {
mockTestWorkflow.mockResolvedValue({ data: { errors: {} } });
it("re-runs FROM this node with its (edited) input, then closes", async () => {
mockTestWorkflow.mockResolvedValue({ data: { errors: {}, inputs: {} } });
const wrapper = mountDrawer();
expect(replayBtn(wrapper).attributes("disabled")).toBeUndefined();
await replayBtn(wrapper).trigger("click");
await flushPromises();
// The whole in-memory graph is sent (test-without-saving), not just an id.
expect(mockTestWorkflow).toHaveBeenCalledWith(
expect.objectContaining({
org_identifier: "default",
inputs: JSON.parse(VALID_INPUT),
inputs: [rec(2)], // f1's own input, unwrapped
from_node: "f1",
workflow: expect.objectContaining({ id: "wf1" }),
}),
);
expect(workflowObj.testRun.resultDrawer).toEqual({
show: false,
nodeId: "",
});
expect(workflowObj.testRun.resultDrawer).toEqual({ show: false, nodeId: "" });
});
it("keeps the drawer open and toasts the backend message on failure", async () => {
mockTestWorkflow.mockRejectedValue({
response: { data: { message: "vrl compile error" } },
});
mockTestWorkflow.mockRejectedValue({ response: { data: { message: "vrl compile error" } } });
const wrapper = mountDrawer();
await replayBtn(wrapper).trigger("click");
await flushPromises();
expect(mockToast).toHaveBeenCalledWith({
message: "vrl compile error",
variant: "error",
});
expect(mockToast).toHaveBeenCalledWith({ message: "vrl compile error", variant: "error" });
expect(workflowObj.testRun.resultDrawer.show).toBe(true);
});
@@ -388,98 +498,124 @@ describe("WorkflowStepResultDrawer", () => {
await replayBtn(wrapper).trigger("click");
await flushPromises();
expect(mockToast).toHaveBeenCalledWith({
message: i18n.global.t("workflow.test.runError"),
message: t("workflow.test.runError"),
variant: "error",
});
});
it("is disabled and inert while a replay is in flight", async () => {
it("is disabled + inert while a replay is in flight", async () => {
let resolve!: (v: any) => void;
mockTestWorkflow.mockReturnValue(
new Promise((r) => {
resolve = r;
}),
);
mockTestWorkflow.mockReturnValue(new Promise((r) => (resolve = r)));
const wrapper = mountDrawer();
await replayBtn(wrapper).trigger("click");
await nextTick();
expect(replayBtn(wrapper).attributes("data-loading")).toBe("true");
expect(replayBtn(wrapper).attributes("disabled")).toBeDefined();
// a second click while replaying is a no-op
await replayBtn(wrapper).trigger("click");
expect(mockTestWorkflow).toHaveBeenCalledTimes(1);
resolve({ data: { errors: {} } });
resolve({ data: { errors: {}, inputs: {} } });
await flushPromises();
});
it("does not call the API when the input is not a valid JSON array", async () => {
setup({ input: "not json" });
const wrapper = mountDrawer();
await replayBtn(wrapper).trigger("click");
await flushPromises();
expect(mockTestWorkflow).not.toHaveBeenCalled();
it("is disabled when no records reached the node (nothing to replay)", () => {
setup({
nodeId: "d1",
result: {
errors: {},
inputs: { t1: [rec(1)], f1: [rec(2)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: [],
},
});
expect(replayBtn(mountDrawer()).attributes("disabled")).toBeDefined();
});
it("carries the replay hint tooltip", () => {
expect(mountDrawer().find(".o-tooltip").attributes("data-content")).toBe(
i18n.global.t("workflow.test.stepResult.replayHint"),
expect(tooltipFor(mountDrawer(), t("workflow.test.stepResult.replayHint"))).toBeTruthy();
});
});
describe("history mode (read-only past run)", () => {
// History carries the SAME per-node `inputs` map as a Test run — the drawer
// renders Input/Output for every node, just read-only (no Replay).
const historyResult = () => ({
errors: { f1: { error_count: 1, errors: [["boom"]] } },
inputs: { t1: [rec(1)], f1: [{ meta: { alert_name: "a" }, data: [{ x: 1 }] }], d1: [rec(3)] },
ranNodeIds: ["t1", "f1", "d1"],
blockedNodeIds: ["d1"],
mode: "history",
runId: "r1",
});
it("shows the per-node input, read-only", () => {
setup({ result: historyResult() });
const editor = inputEditor(mountDrawer());
expect(editor.props("readOnly")).toBe(true);
expect(JSON.parse(editor.props("query"))).toEqual([
{ meta: { alert_name: "a" }, data: [{ x: 1 }] },
]);
});
it("derives Output for a NON-error node too (not just error nodes)", () => {
// f1 -> d1: f1's output == d1's input.
setup({ result: historyResult() });
expect(JSON.parse(outputEditor(mountDrawer()).props("query"))).toEqual([rec(3)]);
});
it("hides Replay (read-only) but KEEPS Use-as-input (seed a fresh test)", () => {
setup({ result: historyResult() });
const wrapper = mountDrawer();
expect(replayBtn(wrapper).exists()).toBe(false);
expect(useInputBtn(wrapper).exists()).toBe(true);
});
it("still shows the error output for the past run", () => {
setup({ result: historyResult() });
expect(mountDrawer().find('[data-test="workflow-step-result-error-line"]').text()).toBe(
"boom",
);
});
});
describe("copy", () => {
it("copies the input with the input success message", async () => {
it("copies the input records with the input success message", async () => {
const wrapper = mountDrawer();
const btns = wrapper.findAll(
`[title="${i18n.global.t("workflow.test.stepResult.copyInput")}"]`,
);
await btns[0].trigger("click");
expect(mockCopy).toHaveBeenCalledWith(VALID_INPUT, {
successMessage: i18n.global.t("workflow.test.stepResult.copiedInput"),
const btn = wrapper.find(`[title="${t("workflow.test.stepResult.copyInput")}"]`);
await btn.trigger("click");
expect(mockCopy).toHaveBeenCalledWith(JSON.stringify([rec(2)], null, 2), {
successMessage: t("workflow.test.stepResult.copiedInput"),
});
});
it("copies the output as the joined error messages", async () => {
it("copies the output records", async () => {
const wrapper = mountDrawer();
const btns = wrapper.findAll(
`[title="${i18n.global.t("workflow.test.stepResult.copyOutput")}"]`,
);
await btns[0].trigger("click");
expect(mockCopy).toHaveBeenCalledWith("boom\nbad record", {
successMessage: i18n.global.t("workflow.test.stepResult.copiedOutput"),
const btn = wrapper.find(`[title="${t("workflow.test.stepResult.copyOutput")}"]`);
await btn.trigger("click");
expect(mockCopy).toHaveBeenCalledWith(JSON.stringify([rec(3)], null, 2), {
successMessage: t("workflow.test.stepResult.copiedOutput"),
});
});
it("disables both copy buttons and copies nothing when there is no text", async () => {
setup({ input: "", result: { errors: {} } });
it("copies error + forwarded together for an errored node", async () => {
setup({ result: errorResult() });
const wrapper = mountDrawer();
const copyIn = wrapper.find(
`[title="${i18n.global.t("workflow.test.stepResult.copyInput")}"]`,
);
const copyOut = wrapper.find(
`[title="${i18n.global.t("workflow.test.stepResult.copyOutput")}"]`,
);
expect(copyIn.attributes("disabled")).toBeDefined();
expect(copyOut.attributes("disabled")).toBeDefined();
// even if invoked, an empty payload is a no-op
await copyIn.trigger("click");
await copyOut.trigger("click");
expect(mockCopy).not.toHaveBeenCalled();
const btn = wrapper.find(`[title="${t("workflow.test.stepResult.copyOutput")}"]`);
await btn.trigger("click");
const copied = mockCopy.mock.calls[0][0];
expect(copied).toContain("boom");
expect(copied).toContain('"x": 3'); // forwarded records included
});
});
describe("fullscreen", () => {
const fsButtons = (w: any) =>
w.findAll(`[title="${i18n.global.t("workflow.test.stepResult.enterFullscreen")}"]`);
w.findAll(`[title="${t("workflow.test.stepResult.enterFullscreen")}"]`);
it("toggles the Input+Output container as one unit", async () => {
const wrapper = mountDrawer();
await fsButtons(wrapper)[0].trigger("click");
expect(mockToggleFullscreen).toHaveBeenCalledWith(
wrapper.find("[data-test='workflow-step-io-container']").element,
wrapper.find('[data-test="workflow-step-io-container"]').element,
);
});
@@ -493,34 +629,7 @@ describe("WorkflowStepResultDrawer", () => {
spy.mockRestore();
});
it("flips the icon + title once the document reports fullscreen", async () => {
const wrapper = mountDrawer();
const el = wrapper.find("[data-test='workflow-step-io-container']").element;
expect(wrapper.findAll('[data-icon="fullscreen"]')).toHaveLength(2);
Object.defineProperty(document, "fullscreenElement", {
configurable: true,
value: el,
});
document.dispatchEvent(new Event("fullscreenchange"));
await nextTick();
expect(wrapper.findAll('[data-icon="fullscreen-exit"]')).toHaveLength(2);
expect(
wrapper.findAll(`[title="${i18n.global.t("workflow.test.stepResult.exitFullscreen")}"]`),
).toHaveLength(2);
// leaving fullscreen flips it back
Object.defineProperty(document, "fullscreenElement", {
configurable: true,
value: null,
});
document.dispatchEvent(new Event("fullscreenchange"));
await nextTick();
expect(wrapper.findAll('[data-icon="fullscreen"]')).toHaveLength(2);
});
it("removes its fullscreen listeners on unmount", async () => {
it("removes its fullscreen listeners on unmount", () => {
const remove = vi.spyOn(document, "removeEventListener");
mountDrawer().unmount();
const events = remove.mock.calls.map((c) => c[0]);
@@ -535,12 +644,9 @@ describe("WorkflowStepResultDrawer", () => {
const wrapper = mountDrawer();
const close = wrapper
.findAll(".drawer-footer button")
.find((b: any) => b.text() === i18n.global.t("common.close"))!;
.find((b: any) => b.text() === t("common.close"))!;
await close.trigger("click");
expect(workflowObj.testRun.resultDrawer).toEqual({
show: false,
nodeId: "",
});
expect(workflowObj.testRun.resultDrawer).toEqual({ show: false, nodeId: "" });
});
it("clears the result drawer when the drawer dismisses itself", async () => {
@@ -548,19 +654,11 @@ describe("WorkflowStepResultDrawer", () => {
await wrapper.find(".drawer-dismiss").trigger("click");
expect(workflowObj.testRun.resultDrawer.show).toBe(false);
});
it("keeps it open when the drawer reports open=true", async () => {
const wrapper = mountDrawer();
wrapper.findComponent(ODrawerStub as any).vm.$emit("update:open", true);
await nextTick();
expect(workflowObj.testRun.resultDrawer.show).toBe(true);
});
});
it("mounts fine when there are only icon buttons and no node data at all", () => {
setup({ nodeId: "", nodes: [], result: null });
it("mounts fine with no node data and no result at all", () => {
setup({ nodeId: "", nodes: [], edges: [], result: null });
const wrapper = mountDrawer();
expect(wrapper.find(".o-drawer").exists()).toBe(true);
expect(iconButtons(wrapper).length).toBeGreaterThan(0);
});
});
web/src/components/workflows/WorkflowStepResultDrawer.vue
+292 -47
View File
@@ -20,11 +20,12 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
copy + a shared fullscreen toggle, mirroring the LLM span preview), plus a
Replay button that re-runs the workflow from this node with its input.
Data: the test response is errors-only today, so for an ERROR node the input is
derived from the errored records (`NodeErrors`) and the output is the error
messages; Replay works. For an OK/skipped node there's no captured I/O yet, so
the panels show a placeholder — they fill in automatically once the backend
returns per-node `node_io { input, output }` (already read here when present).
Data: the test response carries a per-node `inputs` map. Input = the records
THIS node received (nodeTestInput). Output = what it emitted, derived per
outgoing edge as the downstream node's input (nodeTestOutputBranches) — the graph
is a single-incoming tree, so a child's input IS this node's output on that edge.
A terminal (destination) shows the records it sent externally; an errored node
shows the error message(s). Opens for ANY node (✓ / grey / error), not just errors.
-->
<template>
<ODrawer
@@ -35,10 +36,11 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
@update:open="onOpenChange"
>
<div class="flex h-full min-h-0 flex-col gap-3 p-4">
<!-- status — the drawer only opens for error nodes, so it's always Errored -->
<!-- status — reflects this node's actual test result (Errored / Passed /
No Records), same as its canvas badge. -->
<div class="flex items-center justify-end">
<OBadge variant="error-soft" size="sm">
{{ t("workflow.test.stepResult.status.error") }}
<OBadge :variant="statusVariant" size="sm" data-test="workflow-step-result-status">
{{ t(`workflow.test.stepResult.status.${stepStatus}`) }}
</OBadge>
</div>
@@ -57,6 +59,17 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
<div class="text-text-body mb-2 flex items-center justify-between text-sm font-bold">
<div>{{ t("workflow.test.stepResult.input") }}</div>
<div class="flex items-center gap-1">
<OTooltip :content="useInputTooltip" :delay="300" side="top">
<OButton
variant="outline"
size="icon"
:disabled="!inputRecordsForTest"
data-test="workflow-step-use-input-as-test"
@click="useAsTestInput(inputRecordsForTest)"
>
<OIcon name="play-circle" size="xs" />
</OButton>
</OTooltip>
<OButton
variant="outline"
size="icon"
@@ -77,11 +90,14 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</div>
</div>
<!-- Test mode: input is EDITABLE and Replay re-runs from this step with
it. History mode: read-only per-node input captured for the run. -->
it. History mode: read-only per-node input captured for the run.
When no records reached this step, show a clear empty state instead
of a blank editor (which reads as "type something here"). -->
<div
class="border-border-default rounded-default bg-code-bg min-h-0 flex-1 overflow-hidden border"
>
<CodeQueryEditor
v-if="receivedRecords"
editor-id="workflow-step-input"
language="json"
:query="inputModel"
@@ -89,9 +105,16 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
:show-auto-complete="false"
@update:query="inputModel = $event"
/>
<div
v-else
data-test="workflow-step-result-no-input"
class="text-text-secondary flex h-full items-center justify-center p-8 text-center text-sm italic"
>
{{ t("workflow.test.stepResult.noInput") }}
</div>
</div>
<div
v-if="!isHistory && editableInput && inputInvalid"
v-if="!isHistory && receivedRecords && editableInput && inputInvalid"
data-test="workflow-step-result-input-error"
class="text-input-error-text mt-1 text-xs leading-snug"
>
@@ -104,6 +127,17 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
<div class="text-text-body mb-2 flex items-center justify-between text-sm font-bold">
<div>{{ t("workflow.test.stepResult.output") }}</div>
<div class="flex items-center gap-1">
<OTooltip :content="useOutputTooltip" :delay="300" side="top">
<OButton
variant="outline"
size="icon"
:disabled="!outputRecordsForTest"
data-test="workflow-step-use-output-as-test"
@click="useAsTestInput(outputRecordsForTest)"
>
<OIcon name="play-circle" size="xs" />
</OButton>
</OTooltip>
<OButton
variant="outline"
size="icon"
@@ -124,29 +158,72 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</div>
</div>
<div
class="border-border-default rounded-default bg-code-bg min-h-0 flex-1 overflow-hidden border"
class="border-border-default rounded-default bg-code-bg flex min-h-0 flex-1 flex-col overflow-hidden border"
>
<!-- The step errored — show the error message(s) as the output. -->
<!-- Nothing produced (no error, no forwarded records) — one centered
empty state, matching the Input pane so both sides stay aligned. -->
<div
v-if="errorMessages.length"
class="flex h-full flex-col gap-1.5 overflow-auto px-3 py-2.5"
>
<div
v-for="(m, i) in errorMessages"
:key="i"
data-test="workflow-step-result-error-line"
class="text-status-error-text text-xs leading-snug whitespace-pre-wrap"
>
{{ m }}
</div>
</div>
<div
v-else
data-test="workflow-step-result-no-output"
v-if="outputIsEmpty"
data-test="workflow-step-result-output-empty"
class="text-text-secondary flex h-full items-center justify-center p-8 text-center text-sm italic"
>
{{ t("workflow.test.stepResult.noOutput") }}
{{ outputEmptyMessage }}
</div>
<template v-else>
<!-- Error message(s). Shown ABOVE the forwarded editor when a node
errored but still passed records on (e.g. a function that failed
on some records, forwarded others). -->
<div
v-if="showError"
data-test="workflow-step-result-error-section"
class="flex shrink-0 flex-col gap-1.5 overflow-auto p-2.5"
:class="{ 'flex-1': !hasForwardedRecords }"
>
<div
v-if="showSectionHeadings"
class="text-status-error-text text-xs font-semibold"
>
{{ t("workflow.test.stepResult.errorHeading") }}
</div>
<div
v-for="(m, i) in errorMessages"
:key="i"
data-test="workflow-step-result-error-line"
class="text-status-error-text text-xs leading-snug whitespace-pre-wrap"
>
{{ m }}
</div>
</div>
<!-- Forwarded records in the SAME read-only editor the Input uses, so
the output gets JSON highlighting / folding. -->
<template v-if="hasForwardedRecords">
<div
v-if="showSectionHeadings"
class="text-text-secondary shrink-0 px-2.5 pt-2 text-xs font-semibold"
>
{{ t("workflow.test.stepResult.forwardedHeading") }}
</div>
<div
v-for="(label, i) in showBranchLabels ? outputBranchLabels : []"
:key="i"
data-test="workflow-step-result-output-branch-label"
class="text-text-secondary flex shrink-0 items-center gap-1 px-2.5 pt-2 text-xs font-medium"
>
<OIcon name="arrow-right" size="xs" />
<span class="min-w-0 truncate">{{ label }}</span>
</div>
<div class="min-h-0 flex-1" data-test="workflow-step-result-output-records">
<CodeQueryEditor
editor-id="workflow-step-output"
language="json"
:query="outputRecordsText || ''"
:read-only="true"
:show-auto-complete="false"
/>
</div>
</template>
</template>
</div>
</div>
</div>
@@ -182,7 +259,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</template>
<script setup lang="ts">
import { computed, onBeforeUnmount, onMounted, ref } from "vue";
import { computed, onBeforeUnmount, onMounted, ref, watch } from "vue";
import { useI18n } from "vue-i18n";
import { useStore } from "vuex";
@@ -201,6 +278,8 @@ import {
nodeMeta,
nodeConfigDetail,
executeTestRun,
nodeTestInput,
nodeTestOutputBranches,
} from "@/plugins/workflows/useWorkflowCanvas";
const { t } = useI18n();
@@ -235,29 +314,74 @@ const errorMessages = computed<string[]>(() =>
errorEntries.value.map((e) => (Array.isArray(e) ? String(e[0]) : String(e))),
);
// Header status — mirrors the node's canvas badge (same source of truth): Errored
// if it's in the errors map, Passed if records reached it, else No Records (it ran
// but was fed nothing — an upstream filter/error). The drawer opens for ANY node
// now, so this can't be hardcoded to "error".
const stepStatus = computed<"ok" | "error" | "skipped">(() => {
const r = result.value;
if (r?.errors?.[nodeId.value]) return "error";
if (r?.inputs) return r.inputs[nodeId.value]?.length ? "ok" : "skipped";
if (r?.blockedNodeIds?.includes(nodeId.value)) return "skipped";
return "ok";
});
const statusVariant = computed<"error-soft" | "success-soft" | "default-soft">(() =>
stepStatus.value === "error"
? "error-soft"
: stepStatus.value === "ok"
? "success-soft"
: "default-soft",
);
// A history run is read-only (viewing a past execution). A live Test run is
// editable + replayable.
const isHistory = computed(() => result.value?.mode === "history");
// ONE central input, shared with the Test dialog (workflowObj.testRun.input).
// Editing it here or in the Test dialog stays in sync; Replay from this node
// feeds this same payload. Bound directly — no per-node copies to desync.
const editableInput = computed<string>({
get: () => workflowObj.testRun.input || "",
set: (val: string) => (workflowObj.testRun.input = val),
});
// A record's `data` field arrives JSON-stringified (a serialization artifact of
// the destination flattening), so a naive stringify renders it as an escaped
// one-liner. Recursively unwrap any string value that is itself JSON (object /
// array) so records read as clean nested JSON. Applied to BOTH panes' display.
const deepUnwrapJson = (value: any): any => {
if (typeof value === "string") {
const t = value.trim();
if (t.startsWith("{") || t.startsWith("[")) {
try {
return deepUnwrapJson(JSON.parse(value));
} catch {
return value;
}
}
return value;
}
if (Array.isArray(value)) return value.map(deepUnwrapJson);
if (value && typeof value === "object") {
const out: Record<string, any> = {};
for (const k of Object.keys(value)) out[k] = deepUnwrapJson(value[k]);
return out;
}
return value;
};
const prettyRecords = (records: any): string => JSON.stringify(deepUnwrapJson(records), null, 2);
// History mode: the per-node input captured for this run (node_map[nodeId]),
// pretty-printed and read-only.
const nodeInputText = computed<string>(() => {
const v = result.value?.nodeInputs?.[nodeId.value];
return v == null ? "" : JSON.stringify(v, null, 2);
});
// Test mode: the Input editor shows the records THIS node actually received on
// the last run (from the backend `inputs` map), editable so Replay can re-run
// from here with a tweaked payload. The stringified `data` field is unwrapped for
// readability — Replay then sends that unwrapped shape (an accepted trade-off).
const editableInput = ref("");
const seedInputFromRun = () => {
const recs = nodeTestInput(nodeId.value);
editableInput.value = recs == null ? "" : prettyRecords(recs);
};
// Seed synchronously so the editor is populated on first paint (not a tick later);
// the watch + onMounted below re-seed if the selected node changes while open.
// Both Test and History runs carry the same per-node `inputs` map, so the Input is
// seeded the same way for both — history just renders it read-only (no Replay).
seedInputFromRun();
// What the Input editor shows: read-only per-node input for a past run, else the
// editable central input. The setter no-ops in history mode.
// What the Input editor shows. In history it's read-only (the editor's :read-only
// binding), so the setter no-ops.
const inputModel = computed<string>({
get: () => (isHistory.value ? nodeInputText.value : editableInput.value),
get: () => editableInput.value,
set: (val: string) => {
if (!isHistory.value) editableInput.value = val;
},
@@ -276,8 +400,92 @@ const parsedReplayInput = computed<any[] | null>(() => {
});
const inputInvalid = computed(() => parsedReplayInput.value === null);
// Output copy = the error text.
const copyableOutput = computed(() => errorMessages.value.join("\n"));
// Did records actually reach this node on the run? Based on what the backend
// captured (not the editable buffer), so it stays stable if the user clears the
// editor. When false, the Input pane shows an empty state instead of a blank
// editor, and Replay is disabled (there's nothing here to re-run).
const receivedRecords = computed(() => nodeTestInput(nodeId.value) != null);
// OUTPUT = per outgoing edge, the records the downstream node received (== what
// this node emitted on that branch). Empty array => terminal node (a sink) with
// no derivable downstream output. A branch with `records: null` passed nothing on
// (filtered out / never reached).
// A node with no outgoing edge is a terminal (destination/sink). Its "output" IS
// what it sends externally = the records it received.
const isTerminal = computed(() => nodeTestOutputBranches(nodeId.value).length === 0);
// The OUTPUT as ONE JSON document for the read-only editor — so it gets the same
// syntax highlighting / folding as the Input pane. A terminal shows the records it
// sent; a single downstream shows that branch's records; a fan-out shows an object
// keyed by "→ target" so each branch stays labelled. Null when nothing was emitted.
const outputRecordsText = computed<string | null>(() => {
if (isTerminal.value) {
const recs = nodeTestInput(nodeId.value);
return recs == null ? null : prettyRecords(recs);
}
const branches = nodeTestOutputBranches(nodeId.value).filter((b) => b.records != null);
if (!branches.length) return null;
if (branches.length === 1) return prettyRecords(branches[0].records);
const byBranch: Record<string, any> = {};
for (const b of branches) {
const typeName = nodeMeta(b.nodeType) ? t(nodeMeta(b.nodeType)!.titleKey) : b.nodeType;
const label = b.detail ? `${typeName} · ${b.detail}` : typeName;
byBranch[`→ ${label}`] = deepUnwrapJson(b.records);
}
return JSON.stringify(byBranch, null, 2);
});
// Caption line(s) shown above the output editor — where the records went and how
// many. One entry per downstream target (terminal → the external send).
const outputBranchLabels = computed<string[]>(() => {
if (isTerminal.value) {
const recs = nodeTestInput(nodeId.value);
return recs && recs.length ? [t("workflow.test.stepResult.sentExternally")] : [];
}
return nodeTestOutputBranches(nodeId.value)
.filter((b) => Array.isArray(b.records) && b.records.length)
.map((b) => {
const typeName = nodeMeta(b.nodeType) ? t(nodeMeta(b.nodeType)!.titleKey) : b.nodeType;
return b.detail ? `${typeName} · ${b.detail}` : typeName;
});
});
// A caption above the output editor is shown ONLY when it adds information: a
// terminal (names the external send) or a fan-out (distinguishes branches). For a
// single downstream the target is obvious from the canvas, so we show no caption at
// all — a "→ to node X" line there just reads as noise.
const showBranchLabels = computed(() => isTerminal.value || outputBranchLabels.value.length > 1);
// When nothing was emitted, explain WHY in the node's own terms: a condition
// matched nothing (the common, actionable case) vs. a generic drop.
const emptyOutputMessage = computed(() =>
node.value?.data?.node_type === "condition"
? t("workflow.test.stepResult.conditionNoMatch")
: t("workflow.test.stepResult.branchFiltered"),
);
// A node can BOTH error AND forward records (a function that errored on some
// records but passed others). Show the error text AND the forwarded editor. The
// exception is an errored TERMINAL (a failed send): no successful output to show.
const showError = computed(() => errorMessages.value.length > 0);
const showForwarded = computed(() => !isTerminal.value || !showError.value);
const hasForwardedRecords = computed(() => showForwarded.value && !!outputRecordsText.value);
// Label the two sections only when both are visible, so a plain output stays clean.
const showSectionHeadings = computed(() => showError.value && hasForwardedRecords.value);
// Nothing produced (no error, no forwarded records) → one centered empty state,
// same container pattern as the Input pane.
const outputIsEmpty = computed(() => !showError.value && !hasForwardedRecords.value);
const outputEmptyMessage = computed(() =>
isTerminal.value ? t("workflow.test.stepResult.destinationNoRecords") : emptyOutputMessage.value,
);
// Output copy = the error text (if any) + the forwarded records.
const copyableOutput = computed(() => {
const parts: string[] = [];
if (showError.value) parts.push(errorMessages.value.join("\n"));
if (hasForwardedRecords.value && outputRecordsText.value) parts.push(outputRecordsText.value);
return parts.join("\n\n");
});
const copy = (text: string, type: "input" | "output") => {
if (!text) return;
@@ -309,7 +517,11 @@ const toggleFullscreen = () => {
const onFullscreenChange = () => {
isFullscreen.value = document.fullscreenElement === ioContainerRef.value;
};
// Seed the Input editor with this node's received records when the drawer opens,
// and re-seed if the selected node changes while open (both Test and History).
watch(nodeId, seedInputFromRun);
onMounted(() => {
seedInputFromRun();
document.addEventListener("fullscreenchange", onFullscreenChange);
document.addEventListener("webkitfullscreenchange", onFullscreenChange);
});
@@ -345,4 +557,37 @@ const close = () => {
const onOpenChange = (open: boolean) => {
if (!open) close();
};
// "Use as Test Input" — promote a pane's records to the central test payload and
// open the Test dialog (set-and-open), so the WHOLE flow can be re-run with them.
// Input uses the (possibly edited) input editor; Output uses what the node emitted
// (its sent records for a terminal, or every downstream branch's records).
const inputRecordsForTest = computed<any[] | null>(() => {
const p = parsedReplayInput.value;
return p && p.length ? p : null;
});
const outputRecordsForTest = computed<any[] | null>(() => {
if (isTerminal.value) return nodeTestInput(nodeId.value);
const recs = nodeTestOutputBranches(nodeId.value).flatMap((b) => b.records || []);
return recs.length ? recs : null;
});
// Disabled-state tooltips explain WHY there's nothing to use.
const useInputTooltip = computed(() =>
inputRecordsForTest.value
? t("workflow.test.stepResult.useAsTestInput")
: t("workflow.test.stepResult.useAsTestInputNoInput"),
);
const useOutputTooltip = computed(() =>
outputRecordsForTest.value
? t("workflow.test.stepResult.useAsTestInput")
: t("workflow.test.stepResult.useAsTestInputNoOutput"),
);
const useAsTestInput = (records: any[] | null) => {
if (!records || !records.length) return;
workflowObj.testRun.input = JSON.stringify(records, null, 2);
close();
workflowObj.testRun.show = true;
};
</script>
web/src/lib/navigation/Stepper/OStepper.spec.ts
+30 -1
View File
@@ -19,6 +19,7 @@ function mountStepper(
animated?: boolean;
expanded?: boolean;
stepCount?: number;
hideHeader?: boolean;
} = {},
) {
const {
@@ -28,6 +29,7 @@ function mountStepper(
animated = false, // disable animations in tests to avoid async waits
expanded = false,
stepCount = 2,
hideHeader = false,
} = options;
const Wrapper = defineComponent({
@@ -41,7 +43,18 @@ function mountStepper(
const done1 = computed(() => active.value > 1);
const done2 = computed(() => active.value > 2);
const done3 = computed(() => active.value > 3);
return { active, orientation, navigable, animated, expanded, stepCount, done1, done2, done3 };
return {
active,
orientation,
navigable,
animated,
expanded,
stepCount,
hideHeader,
done1,
done2,
done3,
};
},
template: `
<OStepper
@@ -50,6 +63,7 @@ function mountStepper(
:navigable="navigable"
:animated="animated"
:expanded="expanded"
:hide-header="hideHeader"
>
<OStep :name="1" title="First" :done="done1">Step 1 content</OStep>
<OStep v-if="stepCount >= 2" :name="2" title="Second" :done="done2">Step 2 content</OStep>
@@ -152,6 +166,21 @@ describe("OStepper", () => {
// --- Orientation ---
it("hideHeader removes the header bar but still renders the active step content", async () => {
const wrapper = mountStepper({ modelValue: 2, hideHeader: true });
await wrapper.vm.$nextTick();
// Header (role="list") is gone...
expect(wrapper.find('[role="list"]').exists()).toBe(false);
// ...but the active step's panel still renders.
expect(wrapper.text()).toContain("Step 2 content");
});
it("renders the header by default (hideHeader off)", async () => {
const wrapper = mountStepper({ modelValue: 1 });
await wrapper.vm.$nextTick();
expect(wrapper.find('[role="list"]').exists()).toBe(true);
});
it("does not render the horizontal header list in vertical orientation", async () => {
const wrapper = mountStepper({ orientation: "vertical", modelValue: 1 });
await wrapper.vm.$nextTick();
web/src/lib/navigation/Stepper/OStepper.types.ts
+8
View File
@@ -31,6 +31,14 @@ export interface OStepperProps {
* orientations. Default: false.
*/
expanded?: boolean;
/**
* Hide the horizontal step header (the indicator/title progress bar), rendering
* only the active step's panel. Use when the flow is effectively single-step so
* the stepper chrome is just noise (e.g. a create form that skips its first
* step). Vertical orientation has no separate header, so this is a no-op there.
* Default: false.
*/
hideHeader?: boolean;
}
export interface OStepperEmits {
web/src/lib/navigation/Stepper/OStepper.vue
+2 -1
View File
@@ -17,6 +17,7 @@ const props = withDefaults(defineProps<OStepperProps>(), {
animated: true,
navigable: false,
expanded: false,
hideHeader: false,
});
const emit = defineEmits<OStepperEmits>();
@@ -125,7 +126,7 @@ function triggerClasses(step: StepRegistration): string {
reactively whenever a step's done/error state changes.
-->
<div
v-if="isHorizontal"
v-if="isHorizontal && !hideHeader"
role="list"
aria-label="Steps"
class="bg-surface-base! sticky top-0 z-10 flex w-full items-start pb-2"
web/src/locales/languages/en-US.json
+17 -7
View File
@@ -2203,7 +2203,7 @@
"inputLabel": "Sample Input",
"invalidJson": "Enter a valid JSON array of records.",
"resultHint": "After running, each step shows a check or an error icon on the canvas.",
"notVerified": "Not verified — an earlier step errored, so records may not have reached this one.",
"notVerified": "No records reached this step — an upstream condition filtered them out, or an earlier step errored.",
"run": "Run Test",
"allPassed": "All steps passed.",
"nErrored": "{count} step(s) errored.",
@@ -2212,15 +2212,24 @@
"errorsN": "{count} error(s)",
"runError": "Unable to run the test.",
"stepResult": {
"viewHint": "Click To View Input / Output",
"viewHint": "Click to view input / output",
"input": "Input",
"output": "Output",
"noInput": "No Input Captured For This Step.",
"noOutput": "No Output Captured For This Step.",
"noInput": "This step received no records on the last run.",
"noOutput": "No output captured for this step.",
"sentExternally": "Sent to the external destination",
"destinationNoRecords": "No records reached this destination — nothing was sent.",
"branchFiltered": "0 records — nothing passed to this branch.",
"conditionNoMatch": "No records matched this condition — nothing continued downstream.",
"errorHeading": "Error",
"forwardedHeading": "Passed Downstream",
"useAsTestInput": "Use these records as the test input",
"useAsTestInputNoInput": "No input records to use as the test input.",
"useAsTestInputNoOutput": "This step produced no output to use as the test input.",
"copyInput": "Copy Input",
"copyOutput": "Copy Output",
"copiedInput": "Input Copied To Clipboard",
"copiedOutput": "Output Copied To Clipboard",
"copiedInput": "Input copied to clipboard",
"copiedOutput": "Output copied to clipboard",
"enterFullscreen": "Enter Fullscreen",
"exitFullscreen": "Exit Fullscreen",
"replay": "Replay",
@@ -2228,7 +2237,7 @@
"status": {
"ok": "Passed",
"error": "Errored",
"skipped": "Not Verified"
"skipped": "No Records"
}
}
},
@@ -2253,6 +2262,7 @@
"linkError": "Unable To Link Alerts."
},
"chooseTrigger": "Choose a Trigger",
"edgeDeleteHint": "Press Backspace / Delete To Remove The Edge",
"node": {
"kindTrigger": "Trigger",
"kindLogic": "Logic",
web/src/plugins/workflows/WorkflowCanvas.vue
+57 -2
View File
@@ -39,6 +39,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
@node-change="onNodeChange"
@nodes-change="onNodesChange"
@edges-change="onEdgesChange"
@edge-click="onEdgeClick"
@connect="onConnect"
@drop="onDrop"
@dragover="onDragOver"
@@ -99,6 +100,18 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</Controls>
</VueFlow>
<!-- Edge-deletion help hint — shown on edge click (same affordance as the
pipeline canvas). Auto-hides after a few seconds; never shown on the
read-only Runs canvas where edges can't be removed. -->
<div
v-if="showEdgeHelpNotification"
data-test="workflow-edge-delete-hint"
class="bg-surface-base text-text-body rounded-default border-border-default absolute top-5 left-1/2 z-1000 flex -translate-x-1/2 items-center border px-4 py-2.5 text-sm shadow-[0_4px_20px_rgba(0,0,0,0.15)] dark:shadow-[0_4px_20px_rgba(0,0,0,0.4)]"
>
<OIcon name="info" class="mr-1" size="sm" />
{{ t("workflow.edgeDeleteHint") }}
</div>
<!-- Empty-canvas start node (replaces the old "add a trigger" hint text). An
OVERLAY, not a Vue Flow node: a real node would land in
`currentSelectedWorkflow.nodes` and show up in save, validation and the
@@ -140,7 +153,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
// workflow canvases cannot drift. Intentionally global: the selectors target
// VueFlow's own markup, which never carries a scoped data-attribute.
import "@/components/flow/flow-canvas.css";
import { ref, computed } from "vue";
import { ref, computed, onMounted, onBeforeUnmount } from "vue";
import { VueFlow, useVueFlow } from "@vue-flow/core";
import { Background } from "@vue-flow/background";
import { Controls, ControlButton } from "@vue-flow/controls";
@@ -148,6 +161,7 @@ import { useI18n } from "vue-i18n";
import WorkflowNode from "./WorkflowNode.vue";
import FlowEdge from "@/components/flow/FlowEdge.vue";
import FlowNodeCard from "@/components/flow/FlowNodeCard.vue";
import OIcon from "@/lib/core/Icon/OIcon.vue";
import useWorkflowCanvas from "./useWorkflowCanvas";
import "@vue-flow/core/dist/style.css";
@@ -166,12 +180,53 @@ const {
openTriggerPicker,
} = useWorkflowCanvas();
const { onNodesInitialized, setViewport, viewport, dimensions, findNode } = useVueFlow();
const {
onNodesInitialized,
setViewport,
viewport,
dimensions,
findNode,
getSelectedEdges,
removeEdges,
} = useVueFlow();
const vueFlowRef = ref<any>(null);
// Read-only inspection canvas (the Runs view) — disables node drag/connect and,
// via WorkflowNode, the hover add/delete + click-to-edit. Run overlays stay.
const readOnly = computed(() => workflowObj.readOnly);
// Edges have no delete button; clicking one selects it, so surface a transient
// hint that Backspace/Delete removes it (same affordance as the pipeline canvas).
// Skipped on the read-only Runs canvas, where edges can't be removed.
const showEdgeHelpNotification = ref(false);
let edgeHintTimeout: ReturnType<typeof setTimeout> | null = null;
const onEdgeClick = () => {
if (readOnly.value) return;
if (edgeHintTimeout) clearTimeout(edgeHintTimeout);
showEdgeHelpNotification.value = true;
edgeHintTimeout = setTimeout(() => {
showEdgeHelpNotification.value = false;
edgeHintTimeout = null;
}, 3500);
};
// Backspace/Delete removes the selected edge (the action the hint advertises).
// Scoped to EDGES only so node deletion keeps flowing through the confirm dialog.
const onKeydown = (event: KeyboardEvent) => {
if (readOnly.value) return;
if (event.key !== "Delete" && event.key !== "Backspace") return;
const selected = getSelectedEdges.value;
if (!selected.length) return;
event.preventDefault();
removeEdges(selected.map((e) => e.id));
showEdgeHelpNotification.value = false;
};
onMounted(() => window.addEventListener("keydown", onKeydown));
onBeforeUnmount(() => {
window.removeEventListener("keydown", onKeydown);
if (edgeHintTimeout) clearTimeout(edgeHintTimeout);
});
// The "Choose a Trigger" start node shows whenever the workflow has NO TRIGGER —
// not only when the canvas is empty. A workflow needs exactly one trigger, and
// the trigger is now deletable (its kind can be swapped); if the user deletes it
web/src/plugins/workflows/WorkflowNode.spec.ts
+40 -4
View File
@@ -530,11 +530,35 @@ describe("WorkflowNode", () => {
});
it("renders the grey not-verified badge for a blocked (downstream-of-error) node", () => {
// No `inputs` map (history-run fallback) → blocked-downstream drives grey.
setResult({ errors: {}, ranNodeIds: ["c1"], blockedNodeIds: ["c1"] });
wrapper = mountNode("c1", CONDITION.data);
const badge = wrapper.find('[data-test="workflow-node-condition-test-skipped"]');
expect(badge.exists()).toBe(true);
expect(wrapper.text()).toContain("Not verified");
expect(wrapper.text()).toContain("No records reached this step");
});
// Live Test run carries a per-node `inputs` map — the badge reflects whether
// records ACTUALLY reached the node, not just graph reachability.
it("green tick when the node received records (in the inputs map)", () => {
setResult({
errors: {},
inputs: { c1: [{ meta: {}, data: [{ x: 1 }] }] },
ranNodeIds: ["c1"],
blockedNodeIds: [],
});
wrapper = mountNode("c1", CONDITION.data);
expect(wrapper.find('[data-test="workflow-node-condition-test-ok"]').exists()).toBe(true);
});
it("grey badge when the node was reached but got 0 records (filtered upstream)", () => {
// c1 is reachable (ranNodeIds) but absent from `inputs` → not a false pass.
setResult({ errors: {}, inputs: {}, ranNodeIds: ["c1"], blockedNodeIds: [] });
wrapper = mountNode("c1", CONDITION.data);
expect(wrapper.find('[data-test="workflow-node-condition-test-skipped"]').exists()).toBe(
true,
);
expect(wrapper.find('[data-test="workflow-node-condition-test-ok"]').exists()).toBe(false);
});
it("renders the red error badge with its messages in the tooltip", () => {
@@ -628,11 +652,23 @@ describe("WorkflowNode", () => {
});
});
it("the ok badge is not clickable (no drawer)", async () => {
setResult({ errors: {}, ranNodeIds: ["c1"], blockedNodeIds: [] });
it("opens the drawer when the ok badge is clicked (every ran node is inspectable)", async () => {
setResult({
errors: {},
inputs: { c1: [{ meta: {}, data: [{ x: 1 }] }] },
ranNodeIds: ["c1"],
blockedNodeIds: [],
});
wrapper = mountNode("c1", CONDITION.data);
await wrapper.find('[data-test="workflow-node-condition-test-ok"]').trigger("click");
expect(workflowObj.testRun.resultDrawer.show).toBe(false);
expect(workflowObj.testRun.resultDrawer).toEqual({ show: true, nodeId: "c1" });
});
it("opens the drawer when the grey (skipped) badge is clicked", async () => {
setResult({ errors: {}, inputs: {}, ranNodeIds: ["c1"], blockedNodeIds: [] });
wrapper = mountNode("c1", CONDITION.data);
await wrapper.find('[data-test="workflow-node-condition-test-skipped"]').trigger("click");
expect(workflowObj.testRun.resultDrawer).toEqual({ show: true, nodeId: "c1" });
});
it("reacts to a test result arriving after mount", async () => {
web/src/plugins/workflows/WorkflowNode.vue
+19 -7
View File
@@ -81,19 +81,26 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
(red, hover for messages, click to open the step drawer). -->
<div
v-if="testStatus === 'ok'"
class="wf-test-badge wf-test-pop nodrag bg-status-positive text-white"
class="wf-test-badge wf-test-pop nodrag bg-status-positive cursor-pointer text-white transition-transform duration-150 hover:scale-110"
:data-test="`workflow-node-${data?.node_type}-test-ok`"
@pointerdown.stop
@click.stop
@click.stop="openResult"
>
<OIcon name="check" size="xs" />
<OTooltip side="top" align="center" :side-offset="8" max-width="320px">
<template #content>
<div class="p-2 text-left text-xs">
{{ t("workflow.test.stepResult.viewHint") }}
</div>
</template>
</OTooltip>
</div>
<div
v-else-if="testStatus === 'skipped'"
class="wf-test-badge wf-test-pop nodrag bg-badge-default-solid-bg text-badge-default-solid-text cursor-help"
class="wf-test-badge wf-test-pop nodrag bg-badge-default-solid-bg text-badge-default-solid-text cursor-pointer transition-transform duration-150 hover:scale-110"
:data-test="`workflow-node-${data?.node_type}-test-skipped`"
@pointerdown.stop
@click.stop
@click.stop="openResult"
>
<OIcon name="remove" size="xs" />
<OTooltip side="top" align="center" :side-offset="8" max-width="320px">
@@ -152,14 +159,19 @@ const { t } = useI18n();
const { editNode, requestDeleteNode, openStepPicker } = useWorkflowCanvas();
// Test result badge state — read from the last Test run. Null (no run, or this
// node wasn't part of a `from_node` run) → no badge. A node passes (✓) only when
// it ran, has no error, AND no upstream node errored — otherwise it's "skipped"
// (not verified), since the backend gives us errors only, not per-node success.
// node wasn't part of a `from_node` run) → no badge. A node is a real ✓ only when
// records ACTUALLY reached it (it's in the per-node `inputs` map) and it didn't
// error. A node the run reached but that received 0 records — e.g. an upstream
// condition filtered everything out — is "skipped" (grey), NOT a false pass.
const testResult = computed<any>(() => workflowObj.testRun?.result);
const testStatus = computed<"ok" | "error" | "skipped" | null>(() => {
const r = testResult.value;
if (!r || !r.ranNodeIds?.includes(props.id)) return null;
if (r.errors?.[props.id]) return "error";
// Live Test run carries the per-node `inputs` map: ✓ only if this node got
// records; otherwise it ran but processed nothing → grey.
if (r.inputs) return r.inputs[props.id]?.length ? "ok" : "skipped";
// History run (no `inputs` map): fall back to the blocked-downstream logic.
if (r.blockedNodeIds?.includes(props.id)) return "skipped";
return "ok";
});
web/src/plugins/workflows/alertFields.spec.ts
+15 -3
View File
@@ -66,9 +66,21 @@ describe("alertFields", () => {
});
});
it("types every meta field as string (the real payload's meta is a string:string map)", () => {
const types = new Set(TRIGGER_META_VARS.map((v) => v.type));
expect([...types]).toEqual(["string"]);
it("types the numeric alert fields as int and the rest as string (matches the real payload)", () => {
const typeByRef = Object.fromEntries(TRIGGER_META_VARS.map((v) => [v.ref, v.type]));
// numeric alert fields + microsecond timestamps -> int
expect(typeByRef["meta.alert_period"]).toBe("int");
expect(typeByRef["meta.alert_threshold"]).toBe("int");
expect(typeByRef["meta.alert_count"]).toBe("int");
expect(typeByRef["meta.alert_start_time"]).toBe("int");
expect(typeByRef["meta.alert_end_time"]).toBe("int");
// everything else is text
expect(typeByRef["meta.org_id"]).toBe("string");
expect(typeByRef["meta.stream_type"]).toBe("string");
expect(typeByRef["meta.stream_name"]).toBe("string");
expect(typeByRef["meta.alert_name"]).toBe("string");
expect(typeByRef["meta.alert_type"]).toBe("string");
expect(typeByRef["meta.alert_operator"]).toBe("string");
});
it("declares enumValues only on meta.alert_type", () => {
web/src/plugins/workflows/alertFields.ts
+11 -11
View File
@@ -46,12 +46,12 @@ export interface TriggerOutputVar {
enumValues?: string[];
}
// Every value in the emitted `meta` block is a STRING — the backend serialises
// the whole block as a string:string map, so numeric-looking fields (count,
// threshold, period) and the microsecond-epoch timestamps arrive quoted (e.g.
// `"alert_count": "8"`, `"alert_start_time": "1784027820000000"`). Types below
// say `string` to match that real payload, so a Condition author isn't misled
// into a numeric comparison. Confirmed against a live alert firing.
// The emitted `meta` block is a MIXED-type map: text fields are strings, but the
// numeric alert fields (period, threshold, count) and the microsecond-epoch
// timestamps (start/end time) are real numbers — the backend now types them
// (e.g. `"alert_count": 3`, `"alert_start_time": 1785493175167770`). Types below
// match that payload so a Condition author reads the right type. Confirmed
// against a live alert firing.
export const TRIGGER_META_VARS: TriggerOutputVar[] = [
{ ref: "meta.org_id", type: "string", descKey: "workflow.triggerMeta.orgId" },
{ ref: "meta.stream_type", type: "string", descKey: "workflow.triggerMeta.streamType" },
@@ -63,12 +63,12 @@ export const TRIGGER_META_VARS: TriggerOutputVar[] = [
descKey: "workflow.triggerMeta.alertType",
enumValues: ["realtime", "scheduled"],
},
{ ref: "meta.alert_period", type: "string", descKey: "workflow.triggerMeta.alertPeriod" },
{ ref: "meta.alert_period", type: "int", descKey: "workflow.triggerMeta.alertPeriod" },
{ ref: "meta.alert_operator", type: "string", descKey: "workflow.triggerMeta.alertOperator" },
{ ref: "meta.alert_threshold", type: "string", descKey: "workflow.triggerMeta.alertThreshold" },
{ ref: "meta.alert_count", type: "string", descKey: "workflow.triggerMeta.alertCount" },
{ ref: "meta.alert_start_time", type: "string", descKey: "workflow.triggerMeta.alertStartTime" },
{ ref: "meta.alert_end_time", type: "string", descKey: "workflow.triggerMeta.alertEndTime" },
{ ref: "meta.alert_threshold", type: "int", descKey: "workflow.triggerMeta.alertThreshold" },
{ ref: "meta.alert_count", type: "int", descKey: "workflow.triggerMeta.alertCount" },
{ ref: "meta.alert_start_time", type: "int", descKey: "workflow.triggerMeta.alertStartTime" },
{ ref: "meta.alert_end_time", type: "int", descKey: "workflow.triggerMeta.alertEndTime" },
];
// All fields are `Utf8` for now — the flattened `meta` block is a string:string
web/src/plugins/workflows/nodes/WorkflowDestination.spec.ts
+6 -1
View File
@@ -36,7 +36,7 @@ const pickerSubmit = vi.fn();
vi.mock("@/components/flow/forms/DestinationPicker.vue", () => ({
default: {
name: "DestinationPicker",
props: ["initialName"],
props: ["initialName", "forcedType"],
emits: ["expand"],
methods: {
submit: (...args: any[]) => pickerSubmit(...args),
@@ -95,6 +95,11 @@ describe("WorkflowDestination", () => {
const wrapper = createWrapper();
expect(picker(wrapper).props("initialName")).toBe("sink-a");
});
it("locks the inline create form to Custom (workflows only support custom)", () => {
const wrapper = createWrapper();
expect(picker(wrapper).props("forcedType")).toBe("custom");
});
});
describe("createNewDestination (exposed to the drawer)", () => {
web/src/plugins/workflows/nodes/WorkflowDestination.vue
+3
View File
@@ -31,9 +31,12 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
-->
<template>
<div data-test="workflow-destination-body" class="w-full">
<!-- Workflows only support Custom HTTP destinations for now, so lock the inline
create form to Custom and skip its type-selection step. -->
<DestinationPicker
ref="picker"
:initial-name="savedData.destination_id || ''"
forced-type="custom"
@expand="(v) => (creating = v)"
/>
</div>
web/src/plugins/workflows/testSample.spec.ts
+30 -24
View File
@@ -47,7 +47,7 @@ describe("buildTestSample", () => {
expect(Object.keys(firing.meta)).toEqual(expectedKeys);
});
it("prefills the documented named defaults (all string values, matching the real payload)", () => {
it("prefills the documented named defaults (numeric fields as numbers, text as strings)", () => {
const [firing] = buildTestSample() as [Envelope];
expect(firing.meta).toEqual({
org_id: "default",
@@ -55,29 +55,35 @@ describe("buildTestSample", () => {
stream_name: "default",
alert_name: "High Error Rate",
alert_type: "scheduled",
alert_period: "10",
alert_period: 10,
alert_operator: ">=",
alert_threshold: "100",
alert_count: "137",
alert_start_time: String(SAMPLE_TS),
alert_end_time: String(SAMPLE_TS + 600000000),
alert_threshold: 100,
alert_count: 137,
alert_start_time: SAMPLE_TS,
alert_end_time: SAMPLE_TS + 600000000,
});
});
it("uses microsecond-epoch timestamps 10 minutes apart", () => {
const [firing] = buildTestSample() as [Envelope];
expect(firing.meta.alert_start_time).toBe(String(SAMPLE_TS));
expect(firing.meta.alert_start_time).toBe(SAMPLE_TS);
expect(Number(firing.meta.alert_end_time) - Number(firing.meta.alert_start_time)).toBe(
600000000,
);
});
it("emits every meta field as a string (the real payload's meta is a string:string map)", () => {
it("types the numeric meta fields as numbers and text fields as strings", () => {
const [firing] = buildTestSample() as [Envelope];
expect(typeof firing.meta.alert_period).toBe("string");
expect(typeof firing.meta.alert_threshold).toBe("string");
expect(typeof firing.meta.alert_count).toBe("string");
expect(typeof firing.meta.alert_start_time).toBe("string");
// numeric alert fields + timestamps -> numbers
expect(typeof firing.meta.alert_period).toBe("number");
expect(typeof firing.meta.alert_threshold).toBe("number");
expect(typeof firing.meta.alert_count).toBe("number");
expect(typeof firing.meta.alert_start_time).toBe("number");
expect(typeof firing.meta.alert_end_time).toBe("number");
// text fields stay strings
expect(typeof firing.meta.alert_name).toBe("string");
expect(typeof firing.meta.stream_type).toBe("string");
expect(typeof firing.meta.alert_operator).toBe("string");
});
it("uses a valid enum member for alert_type", () => {
@@ -144,17 +150,17 @@ describe("buildFlatTestSample", () => {
});
});
it("stringifies every meta value (the flattened meta block is string:string)", () => {
it("preserves each meta value's native type when flattening", () => {
const [row] = buildFlatTestSample() as [Record<string, unknown>];
Object.keys(row)
.filter((k) => k.startsWith("meta_"))
.forEach((k) => {
expect(typeof row[k]).toBe("string");
});
expect(row.meta_alert_period).toBe("10");
expect(row.meta_alert_threshold).toBe("100");
expect(row.meta_alert_count).toBe("137");
expect(row.meta_alert_start_time).toBe(String(SAMPLE_TS));
// numeric alert fields + timestamps stay numbers
expect(row.meta_alert_period).toBe(10);
expect(row.meta_alert_threshold).toBe(100);
expect(row.meta_alert_count).toBe(137);
expect(row.meta_alert_start_time).toBe(SAMPLE_TS);
expect(typeof row.meta_alert_count).toBe("number");
// text fields stay strings
expect(row.meta_alert_name).toBe("High Error Rate");
expect(typeof row.meta_alert_name).toBe("string");
});
it("preserves the original row columns with their original types", () => {
@@ -265,9 +271,9 @@ describe("type-based fallbacks for unmapped meta fields", () => {
it("still prefers the named default over the type default when the key is mapped", async () => {
const { buildTestSample: build } = await loadWith([
{ ref: "meta.alert_count", type: "string", descKey: "x" },
{ ref: "meta.alert_count", type: "int", descKey: "x" },
]);
expect((build() as any)[0].meta).toEqual({ alert_count: "137" });
expect((build() as any)[0].meta).toEqual({ alert_count: 137 });
});
it("produces an empty meta block when the schema is empty", async () => {
web/src/plugins/workflows/testSample.ts
+20 -19
View File
@@ -18,35 +18,36 @@
// Derived from TRIGGER_META_VARS so the sample's `meta` keys stay in sync with
// the Alert-Trigger schema reference. Shape matches a real alert firing (verified
// against live payload):
// [ { meta: { ...fixed alert fields, all STRING values }, data: [ { ...rows } ] } ]
// The `meta` block is a string:string map — even numeric fields (count,
// threshold, period) and the microsecond-epoch timestamps are quoted strings.
// `data[]` rows keep their native column types (the alert query's result rows).
// [ { meta: { ...fixed alert fields }, data: [ { ...rows } ] } ]
// The `meta` block is a MIXED map: text fields are strings, but the numeric alert
// fields (period, threshold, count) and the microsecond-epoch timestamps
// (start/end time) arrive as real numbers — the backend now types them, so the
// sample must too. `data[]` rows keep their native column types.
import { TRIGGER_META_VARS } from "./alertFields";
const SAMPLE_TS = 1700000000000000; // microsecond epoch, matches alert timestamps
// A readable placeholder per meta field, keyed by the field name (ref minus the
// "meta." prefix). All values are STRINGS to match the real payload. Falls back
// to "" for anything unmapped.
const NAMED_DEFAULTS: Record<string, string> = {
// "meta." prefix). Numeric fields are numbers, text fields are strings — matching
// the real payload's types. Falls back to "" for anything unmapped.
const NAMED_DEFAULTS: Record<string, string | number> = {
org_id: "default",
stream_type: "logs",
stream_name: "default",
alert_name: "High Error Rate",
alert_type: "scheduled",
alert_operator: ">=",
alert_period: "10",
alert_threshold: "100",
alert_count: "137",
alert_start_time: String(SAMPLE_TS),
alert_end_time: String(SAMPLE_TS + 600000000),
alert_period: 10,
alert_threshold: 100,
alert_count: 137,
alert_start_time: SAMPLE_TS,
alert_end_time: SAMPLE_TS + 600000000,
};
const typeDefault = (v: { enumValues?: string[] }) => (v.enumValues?.length ? v.enumValues[0] : "");
export const buildTestSample = (): unknown[] => {
const meta: Record<string, string> = {};
const meta: Record<string, string | number> = {};
for (const v of TRIGGER_META_VARS) {
const key = v.ref.replace(/^meta\./, "");
meta[key] = key in NAMED_DEFAULTS ? NAMED_DEFAULTS[key] : typeDefault(v);
@@ -65,15 +66,15 @@ export const buildTestSample = (): unknown[] => {
export const buildTestSampleText = (): string => JSON.stringify(buildTestSample(), null, 2);
// The FLATTENED view a Function node sees when "After Flattening" (RAF, the
// default) is on: the `meta` block becomes `meta_<field>` string columns merged
// onto each `data[]` row — matching the field names Conditions use. Derived from
// buildTestSample() so both stay in sync. Meta values are strings (the flattened
// `meta` block is a string:string map for now).
// default) is on: the `meta` block becomes `meta_<field>` columns merged onto
// each `data[]` row — matching the field names Conditions use. Derived from
// buildTestSample() so both stay in sync, preserving each field's native type
// (numeric alert fields stay numbers, text fields stay strings).
export const buildFlatTestSample = (): unknown[] => {
const [{ meta, data }] = buildTestSample() as [
{ meta: Record<string, unknown>; data: Record<string, unknown>[] },
];
const metaFlat: Record<string, string> = {};
for (const k of Object.keys(meta)) metaFlat[`meta_${k}`] = String(meta[k]);
const metaFlat: Record<string, unknown> = {};
for (const k of Object.keys(meta)) metaFlat[`meta_${k}`] = meta[k];
return data.map((row) => ({ ...metaFlat, ...row }));
};
web/src/plugins/workflows/useWorkflowCanvas.spec.ts
+115 -13
View File
@@ -48,6 +48,8 @@ import useWorkflowCanvas, {
loadWorkflowRun,
executeTestRun,
serializeWorkflow,
nodeTestInput,
nodeTestOutputBranches,
currentTriggerKind,
} from "@/plugins/workflows/useWorkflowCanvas";
@@ -77,12 +79,16 @@ describe("loadWorkflowRun — history run response mapping", () => {
workflowObj.testRun.result = null;
});
it("maps errors.data (array) to a node-keyed map and node_map to nodeInputs", async () => {
it("maps errors.data (array) to a node-keyed map and input_map to the inputs map", async () => {
const envelope = [{ meta: { alert_name: "t" }, data: [{ a: 1 }] }];
mockRun.mockResolvedValue({
data: {
errors: { run_id: "r1", data: [{ node_id: "n2", error: ["boom"] }] },
data: { complete: envelope, node_map: { n2: envelope } },
// input_map = per-node input for ALL nodes (same shape as a Test run)
data: {
input_map: { n1: envelope, n2: envelope, n3: envelope },
error_node_map: { n2: envelope },
},
},
});
@@ -94,20 +100,28 @@ describe("loadWorkflowRun — history run response mapping", () => {
expect(res.runId).toBe("r1");
// array -> keyed map, in the { error_count, errors: [[msg]] } badge shape
expect(res.errors.n2).toEqual({ error_count: 1, errors: [["boom"]] });
// per-node input carried through verbatim (the {meta,data} envelope)
expect(res.nodeInputs.n2).toEqual(envelope);
expect(res.fullInput).toEqual(envelope);
// per-node input stored under `inputs` (drives Input + derived Output + badges)
expect(res.inputs).toEqual({ n1: envelope, n2: envelope, n3: envelope });
// every node counts as "ran"; n3 is downstream of the errored n2 -> blocked
expect(res.ranNodeIds).toEqual(["n1", "n2", "n3"]);
expect(res.blockedNodeIds).toContain("n3");
expect(res.blockedNodeIds).not.toContain("n2");
});
it("falls back to error_node_map for older runs that lack input_map", async () => {
const envelope = [{ meta: {}, data: [{ a: 1 }] }];
mockRun.mockResolvedValue({
data: { errors: { data: [] }, data: { error_node_map: { n2: envelope } } },
});
await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r1b" });
expect((workflowObj.testRun.result as any).inputs).toEqual({ n2: envelope });
});
it("normalizes a non-array error field into a single-message list", async () => {
mockRun.mockResolvedValue({
data: {
errors: { data: [{ node_id: "n2", error: "single" }] },
data: { node_map: {} },
data: { error_node_map: {} },
},
});
await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r2" });
@@ -119,7 +133,7 @@ describe("loadWorkflowRun — history run response mapping", () => {
it("handles a clean run (no errors) with empty maps", async () => {
mockRun.mockResolvedValue({
data: { errors: { data: [] }, data: { node_map: {} } },
data: { errors: { data: [] }, data: { error_node_map: {} } },
});
const r = await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r3" });
expect(r.ok).toBe(true);
@@ -142,18 +156,18 @@ describe("loadWorkflowRun — history run response mapping", () => {
{ node_id: "deleted-node", error: ["gone"] },
],
},
data: { node_map: {} },
data: { error_node_map: {} },
},
});
await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r1" });
expect((workflowObj.testRun.result as any).ghostNodeIds).toEqual(["deleted-node"]);
});
it("also flags a ghost referenced only by node_map (no error)", async () => {
it("also flags a ghost referenced only by error_node_map (no error)", async () => {
mockRun.mockResolvedValue({
data: {
errors: { data: [] },
data: { node_map: { n1: [], "old-node": [] } },
data: { error_node_map: { n1: [], "old-node": [] } },
},
});
await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r2" });
@@ -164,18 +178,18 @@ describe("loadWorkflowRun — history run response mapping", () => {
mockRun.mockResolvedValue({
data: {
errors: { data: [{ node_id: "n2", error: ["boom"] }] },
data: { node_map: { n1: [], n3: [] } },
data: { error_node_map: { n1: [], n3: [] } },
},
});
await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r3" });
expect((workflowObj.testRun.result as any).ghostNodeIds).toEqual([]);
});
it("does not double-report a ghost referenced by BOTH errors and node_map", async () => {
it("does not double-report a ghost referenced by BOTH errors and error_node_map", async () => {
mockRun.mockResolvedValue({
data: {
errors: { data: [{ node_id: "zombie", error: ["x"] }] },
data: { node_map: { zombie: [] } },
data: { error_node_map: { zombie: [] } },
},
});
await loadWorkflowRun({ orgId: "o", workflowId: "wf1", runId: "r4" });
@@ -317,6 +331,94 @@ describe("executeTestRun — ran-node scope + badge state", () => {
});
});
// nodeTestInput / nodeTestOutputBranches derive per-node Input and Output from the
// backend `inputs` map. Output on an edge == the child's input (single-incoming
// tree), so these two helpers power the whole step-drawer.
describe("nodeTestInput + nodeTestOutputBranches — per-node I/O derivation", () => {
beforeEach(() => {
// trigger(t) -> function(f) -> destination(d); t also -> condition(c) (fan-out)
workflowObj.currentSelectedWorkflow = {
id: "wf1",
name: "wf",
nodes: [
{ id: "t", data: { node_type: "workflow_trigger" } },
{ id: "f", data: { node_type: "function", name: "fn" } },
{ id: "d", data: { node_type: "destination", destination_id: "sink" } },
{ id: "c", data: { node_type: "condition" } },
],
edges: [
{ source: "t", target: "f" },
{ source: "t", target: "c" },
{ source: "f", target: "d" },
],
} as any;
workflowObj.testRun.result = {
errors: {},
inputs: { t: [{ x: 0 }], f: [{ x: 1 }], d: [{ x: 2 }] }, // c got nothing
ranNodeIds: ["t", "f", "c", "d"],
blockedNodeIds: [],
} as any;
});
it("nodeTestInput returns the records a node received, null when absent", () => {
expect(nodeTestInput("f")).toEqual([{ x: 1 }]);
expect(nodeTestInput("c")).toBeNull(); // filtered out — not in inputs
expect(nodeTestInput("missing")).toBeNull();
});
it("nodeTestInput is null when there is no run", () => {
workflowObj.testRun.result = null as any;
expect(nodeTestInput("f")).toBeNull();
});
it("nodeTestOutputBranches: a node's output == each child's input", () => {
const branches = nodeTestOutputBranches("f");
expect(branches).toHaveLength(1);
expect(branches[0]).toMatchObject({ targetId: "d", nodeType: "destination" });
expect(branches[0].records).toEqual([{ x: 2 }]); // == inputs[d]
});
it("fan-out yields one branch per outgoing edge; a filtered branch has null records", () => {
const branches = nodeTestOutputBranches("t");
expect(branches.map((b) => b.targetId).sort()).toEqual(["c", "f"]);
const toC = branches.find((b) => b.targetId === "c")!;
const toF = branches.find((b) => b.targetId === "f")!;
expect(toC.records).toBeNull(); // c received nothing
expect(toF.records).toEqual([{ x: 1 }]);
});
it("a terminal (destination) has no outgoing edges → no branches", () => {
expect(nodeTestOutputBranches("d")).toEqual([]);
});
});
// executeTestRun stores the per-node inputs map so the drawer/badges can read it.
describe("executeTestRun — stores the per-node inputs map", () => {
it("keeps res.data.inputs on testRun.result", async () => {
workflowObj.currentSelectedWorkflow = {
id: "wf1",
name: "wf",
nodes: [{ id: "t", data: { node_type: "workflow_trigger" } }],
edges: [],
} as any;
mockTest.mockResolvedValue({ data: { errors: {}, inputs: { t: [{ x: 1 }] } } });
await executeTestRun({ orgId: "o", inputs: [{ a: 1 }] });
expect((workflowObj.testRun.result as any).inputs).toEqual({ t: [{ x: 1 }] });
});
it("defaults inputs to {} when the response omits it", async () => {
workflowObj.currentSelectedWorkflow = {
id: "wf1",
name: "wf",
nodes: [{ id: "t", data: { node_type: "workflow_trigger" } }],
edges: [],
} as any;
mockTest.mockResolvedValue({ data: { errors: {} } });
await executeTestRun({ orgId: "o", inputs: [{ a: 1 }] });
expect((workflowObj.testRun.result as any).inputs).toEqual({});
});
});
// serializeWorkflow builds the backend `Workflow` object shared by both the
// create/update payload and the Test run — so testing without saving sends the
// exact same graph the editor would persist.
web/src/plugins/workflows/useWorkflowCanvas.ts
+63 -17
View File
@@ -401,6 +401,10 @@ export const executeTestRun = async (opts: {
from_node: opts.fromNode || undefined,
});
const errors = res.data?.errors || {};
// Per-node INPUT map: node_id -> the records that node received. A node's
// OUTPUT is derived from this (see nodeTestOutputBranches): since the graph is
// a single-incoming tree, a child's input IS its parent's output on that edge.
const inputs = res.data?.inputs || {};
// Which nodes ran: from a replay, `fromNode` + everything downstream;
// otherwise everything reachable from the trigger. Nodes NOT reachable
// (unwired / disconnected) never executed, so they must not paint a ✓.
@@ -411,6 +415,7 @@ export const executeTestRun = async (opts: {
const ranNodeIds = startId ? [...reachableFrom(wf.edges || [], [startId])] : [];
workflowObj.testRun.result = {
errors,
inputs,
ranNodeIds,
blockedNodeIds: downstreamOfErrorNodes(Object.keys(errors)),
};
@@ -422,14 +427,55 @@ export const executeTestRun = async (opts: {
}
};
// A single downstream branch of a node's OUTPUT: the target it feeds and the
// records that target received (== what this node emitted on that edge). `records`
// is null when the target got nothing (filtered out / never reached).
export interface NodeOutputBranch {
targetId: string;
nodeType: string;
detail: string;
records: any[] | null;
}
// The INPUT a node received on the last Test run — the raw records from the
// backend `inputs` map (shape varies by node type; rendered as-is). Null when the
// node isn't in the map (0 records reached it) or there's no run.
export const nodeTestInput = (nodeId: string): any[] | null => {
const inputs = workflowObj.testRun.result?.inputs;
const v = inputs?.[nodeId];
return Array.isArray(v) ? v : null;
};
// A node's OUTPUT, per outgoing edge. The graph is a single-incoming tree, so
// each child's input came ONLY from this node — child input == this node's output
// on that branch. One entry per outgoing edge (so fan-out reads per-target); an
// empty array means a terminal node (a destination/sink) with no derivable output.
export const nodeTestOutputBranches = (nodeId: string): NodeOutputBranch[] => {
const wf = workflowObj.currentSelectedWorkflow;
const byId = new Map<string, any>((wf.nodes || []).map((n: any) => [n.id, n]));
return (wf.edges || [])
.filter((e: any) => e.source === nodeId)
.map((e: any) => {
const target = byId.get(e.target);
return {
targetId: e.target,
nodeType: target?.data?.node_type || "",
detail: nodeConfigDetail(target?.data, 40),
records: nodeTestInput(e.target),
};
});
};
// Load a PAST run (from the Executions history) into the same testRun.result the
// canvas already reads — so error nodes paint ✗ and open the step drawer, but
// read-only (no editable input / Replay). The run detail carries:
// errors.data: [{ node_id, error: string[] }] — errored nodes + messages
// data.node_map: { node_id: [{ meta, data }] } — per-node input processed
// data.complete: [{ meta, data }] — full workflow input
// The array-vs-map difference in errors.data is bridged here (each entry has its
// node_id), so it lines up with node_map by key.
// canvas already reads — read-only (no editable input / Replay). The run detail
// now mirrors the Test response shape, so history shows per-node Input/Output for
// EVERY node (not just error nodes):
// errors.data: [{ node_id, error: string[] }] — errored nodes + messages
// data.input_map: { node_id: [records] } — per-node INPUT (all nodes)
// data.error_node_map: { node_id: [records] } — legacy: errored node's input
// input_map is the same per-node `inputs` map a Test run produces, so we store it
// under the same key and the whole drawer (Input + derived Output + badges) works
// identically to Test — just read-only. Falls back to error_node_map for older runs.
export const loadWorkflowRun = async (opts: {
orgId: string;
workflowId: string;
@@ -458,26 +504,27 @@ export const loadWorkflowRun = async (opts: {
};
}
const nodeInputs = payload.data?.node_map || {};
// Per-node INPUT for the whole run (all nodes) — the same shape/semantics as a
// Test run's `inputs`, so the drawer derives Output the same way. Older runs
// only carried error_node_map (errored node's input) — fall back to that.
const inputs = payload.data?.input_map || payload.data?.error_node_map || {};
// GHOST NODES — the run references a node the workflow no longer has (it was
// edited/deleted after the run). Its badge has nowhere to render, so an error
// would silently vanish and the run would look cleaner than it was. Surface
// them so the Runs view can say the graph no longer matches this run.
const currentNodeIds = new Set((wf.nodes || []).map((n: any) => n.id));
const ghostNodeIds = [...new Set([...Object.keys(errors), ...Object.keys(nodeInputs)])].filter(
const ghostNodeIds = [...new Set([...Object.keys(errors), ...Object.keys(inputs)])].filter(
(id) => !currentNodeIds.has(id),
);
workflowObj.testRun.result = {
errors,
// Every node "ran": upstream shows ✓, errored ✗, downstream ⊘ — via the
// existing testStatus logic. Only ✗ nodes are clickable.
// Same per-node inputs map as a Test run — drives the ✓/grey/✗ badges and the
// drawer's Input + derived Output for every node.
inputs,
ranNodeIds: (wf.nodes || []).map((n: any) => n.id),
blockedNodeIds: downstreamOfErrorNodes(Object.keys(errors)),
// Per-node input the drawer shows (read-only) for an errored node.
nodeInputs,
fullInput: payload.data?.complete ?? null,
mode: "history",
runId: opts.runId,
ghostNodeIds,
@@ -808,9 +855,8 @@ export default function useWorkflowCanvas() {
},
},
];
// Open the (read-only) trigger panel on the now-real node, so closing it
// dismisses a panel rather than discarding the node.
editNode(id);
// Don't auto-open the trigger's (read-only) detail panel — placing the trigger
// shouldn't interrupt the build flow. The user can click the node to open it.
}
const NODE_W = 240;
web/src/services/workflows.ts
+1 -1
View File
@@ -105,7 +105,7 @@ const workflows = {
},
// Detail of a single run — errors (per errored node) plus the run's input
// data (`complete` = full workflow input, `node_map` = per-node input the
// data (`complete` = full workflow input, `error_node_map` = per-node input the
// node processed/errored on). Powers the read-only run inspection in the
// editor (click a run in history -> error nodes show Input/Output).
getWorkflowRun: ({