From 6f0b2dcb07cc4e8912bc5ab2732e5f02a3cec1ab Mon Sep 17 00:00:00 2001 From: Yashodhan Joshi Date: Tue, 4 Aug 2026 13:52:01 +0530 Subject: [PATCH] fix: workflow history (#13588) 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 Co-authored-by: Yashodhan Joshi Co-authored-by: sai nikhil kethe --- .../management/src/request/workflows/mod.rs | 12 +- src/core/src/pipeline/batch_execution.rs | 62 +- src/core/src/workflows/mod.rs | 89 ++- .../flow/forms/DestinationPicker.spec.ts | 10 + .../flow/forms/DestinationPicker.vue | 6 +- .../NodeForm/CreateDestinationForm.spec.ts | 50 ++ .../NodeForm/CreateDestinationForm.vue | 68 +- .../workflows/WorkflowEditor.spec.ts | 11 +- .../components/workflows/WorkflowRuns.spec.ts | 23 +- web/src/components/workflows/WorkflowRuns.vue | 43 +- .../workflows/WorkflowRunsPanel.spec.ts | 26 +- .../workflows/WorkflowRunsPanel.vue | 9 +- .../WorkflowStepResultDrawer.spec.ts | 626 ++++++++++-------- .../workflows/WorkflowStepResultDrawer.vue | 339 ++++++++-- .../lib/navigation/Stepper/OStepper.spec.ts | 31 +- .../lib/navigation/Stepper/OStepper.types.ts | 8 + web/src/lib/navigation/Stepper/OStepper.vue | 3 +- web/src/locales/languages/en-US.json | 24 +- web/src/plugins/workflows/WorkflowCanvas.vue | 59 +- .../plugins/workflows/WorkflowNode.spec.ts | 44 +- web/src/plugins/workflows/WorkflowNode.vue | 26 +- web/src/plugins/workflows/alertFields.spec.ts | 18 +- web/src/plugins/workflows/alertFields.ts | 22 +- .../nodes/WorkflowDestination.spec.ts | 7 +- .../workflows/nodes/WorkflowDestination.vue | 3 + web/src/plugins/workflows/testSample.spec.ts | 54 +- web/src/plugins/workflows/testSample.ts | 39 +- .../workflows/useWorkflowCanvas.spec.ts | 128 +++- .../plugins/workflows/useWorkflowCanvas.ts | 80 ++- web/src/services/workflows.ts | 2 +- 30 files changed, 1417 insertions(+), 505 deletions(-) diff --git a/src/api/management/src/request/workflows/mod.rs b/src/api/management/src/request/workflows/mod.rs index 476843052d..3ab12d39bf 100644 --- a/src/api/management/src/request/workflows/mod.rs +++ b/src/api/management/src/request/workflows/mod.rs @@ -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, + inputs: HashMap>, } #[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, ) -> 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), } } diff --git a/src/core/src/pipeline/batch_execution.rs b/src/core/src/pipeline/batch_execution.rs index f2279ebfd3..9d48e52cf2 100644 --- a/src/core/src/pipeline/batch_execution.rs +++ b/src/core/src/pipeline/batch_execution.rs @@ -70,6 +70,7 @@ struct BatchBuffer { pub struct WorkflowResult { pub stream_details: HashMap>, pub errors: HashMap, + pub inputs: HashMap>, } #[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, Option)>(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 = 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>, result_sender: Option>, error_sender: Sender<(String, String, String, Option, Option)>, + inputs_sender: Option>, +} + +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; diff --git a/src/core/src/workflows/mod.rs b/src/core/src/workflows/mod.rs index 66d3a7d6d9..a2eaaff2b8 100644 --- a/src/core/src/workflows/mod.rs +++ b/src/core/src/workflows/mod.rs @@ -41,8 +41,10 @@ use crate::{ pub mod runtime; #[derive(Serialize, Deserialize)] pub struct InputMap { - complete: Vec, - node_map: HashMap>, + // even though a bad naming convention, node_map is + error_node_map: HashMap>, + #[serde(default)] + input_map: HashMap>, } #[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) diff --git a/web/src/components/flow/forms/DestinationPicker.spec.ts b/web/src/components/flow/forms/DestinationPicker.spec.ts index 383a237a01..f0978f48d9 100644 --- a/web/src/components/flow/forms/DestinationPicker.spec.ts +++ b/web/src/components/flow/forms/DestinationPicker.spec.ts @@ -50,6 +50,7 @@ function createWrapper(props: Record = {}) { OSwitch: OSwitchStub, CreateDestinationForm: { name: "CreateDestinationForm", + props: ["forcedType"], emits: ["created", "cancel"], template: '
', }, @@ -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(); diff --git a/web/src/components/flow/forms/DestinationPicker.vue b/web/src/components/flow/forms/DestinationPicker.vue index 3f75e2be9f..8f41e2f4c0 100644 --- a/web/src/components/flow/forms/DestinationPicker.vue +++ b/web/src/components/flow/forms/DestinationPicker.vue @@ -41,6 +41,7 @@ along with this program. If not, see .
@@ -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 }>(); diff --git a/web/src/components/pipeline/NodeForm/CreateDestinationForm.spec.ts b/web/src/components/pipeline/NodeForm/CreateDestinationForm.spec.ts index f5c3ffa39d..b044d0a465 100644 --- a/web/src/components/pipeline/NodeForm/CreateDestinationForm.spec.ts +++ b/web/src/components/pipeline/NodeForm/CreateDestinationForm.spec.ts @@ -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 () => { diff --git a/web/src/components/pipeline/NodeForm/CreateDestinationForm.vue b/web/src/components/pipeline/NodeForm/CreateDestinationForm.vue index 13fe1ab76a..ae1a41fc7f 100644 --- a/web/src/components/pipeline/NodeForm/CreateDestinationForm.vue +++ b/web/src/components/pipeline/NodeForm/CreateDestinationForm.vue @@ -28,9 +28,18 @@ along with this program. If not, see . footer below stays full-width so its top border lines up with the full-width page header (matching the other destination forms). -->
- - - + + + +
Select Destination Type *
@@ -382,6 +391,7 @@ along with this program. If not, see .
(); 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({ 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 diff --git a/web/src/components/workflows/WorkflowEditor.spec.ts b/web/src/components/workflows/WorkflowEditor.spec.ts index cf20f63d1a..e11abcc8a2 100644 --- a/web/src/components/workflows/WorkflowEditor.spec.ts +++ b/web/src/components/workflows/WorkflowEditor.spec.ts @@ -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); }); diff --git a/web/src/components/workflows/WorkflowRuns.spec.ts b/web/src/components/workflows/WorkflowRuns.spec.ts index d18d7e0ed2..f35176e2e0 100644 --- a/web/src/components/workflows/WorkflowRuns.spec.ts +++ b/web/src/components/workflows/WorkflowRuns.spec.ts @@ -75,7 +75,7 @@ const OButtonStub = { name: "OButton", props: ["variant"], emits: ["click"], - template: ``, + template: ``, }; 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"); diff --git a/web/src/components/workflows/WorkflowRuns.vue b/web/src/components/workflows/WorkflowRuns.vue index 077bf44edb..1f7b253579 100644 --- a/web/src/components/workflows/WorkflowRuns.vue +++ b/web/src/components/workflows/WorkflowRuns.vue @@ -27,8 +27,10 @@ along with this program. If not, see . └──────────────────────────────┴──────────────────────┘ 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. --> diff --git a/web/src/lib/navigation/Stepper/OStepper.spec.ts b/web/src/lib/navigation/Stepper/OStepper.spec.ts index c454283a0d..04613e6736 100644 --- a/web/src/lib/navigation/Stepper/OStepper.spec.ts +++ b/web/src/lib/navigation/Stepper/OStepper.spec.ts @@ -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: ` Step 1 content Step 2 content @@ -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(); diff --git a/web/src/lib/navigation/Stepper/OStepper.types.ts b/web/src/lib/navigation/Stepper/OStepper.types.ts index 46527b383d..277ffa02f2 100644 --- a/web/src/lib/navigation/Stepper/OStepper.types.ts +++ b/web/src/lib/navigation/Stepper/OStepper.types.ts @@ -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 { diff --git a/web/src/lib/navigation/Stepper/OStepper.vue b/web/src/lib/navigation/Stepper/OStepper.vue index 841a157178..892c76abf3 100644 --- a/web/src/lib/navigation/Stepper/OStepper.vue +++ b/web/src/lib/navigation/Stepper/OStepper.vue @@ -17,6 +17,7 @@ const props = withDefaults(defineProps(), { animated: true, navigable: false, expanded: false, + hideHeader: false, }); const emit = defineEmits(); @@ -125,7 +126,7 @@ function triggerClasses(step: StepRegistration): string { reactively whenever a step's done/error state changes. -->
. @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 . + +
+ + {{ t("workflow.edgeDeleteHint") }} +
+
+ + +
@@ -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(() => 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"; }); diff --git a/web/src/plugins/workflows/alertFields.spec.ts b/web/src/plugins/workflows/alertFields.spec.ts index 0a9d4d4ce5..d4d0187123 100644 --- a/web/src/plugins/workflows/alertFields.spec.ts +++ b/web/src/plugins/workflows/alertFields.spec.ts @@ -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", () => { diff --git a/web/src/plugins/workflows/alertFields.ts b/web/src/plugins/workflows/alertFields.ts index d9d642e933..f88ba0580d 100644 --- a/web/src/plugins/workflows/alertFields.ts +++ b/web/src/plugins/workflows/alertFields.ts @@ -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 diff --git a/web/src/plugins/workflows/nodes/WorkflowDestination.spec.ts b/web/src/plugins/workflows/nodes/WorkflowDestination.spec.ts index b273a59451..50d71a204d 100644 --- a/web/src/plugins/workflows/nodes/WorkflowDestination.spec.ts +++ b/web/src/plugins/workflows/nodes/WorkflowDestination.spec.ts @@ -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)", () => { diff --git a/web/src/plugins/workflows/nodes/WorkflowDestination.vue b/web/src/plugins/workflows/nodes/WorkflowDestination.vue index f6ef45ea01..44a8688a37 100644 --- a/web/src/plugins/workflows/nodes/WorkflowDestination.vue +++ b/web/src/plugins/workflows/nodes/WorkflowDestination.vue @@ -31,9 +31,12 @@ along with this program. If not, see . -->