starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(flows): make the flow-value round-trip preserve display-only fields in one place (#10382)

Browse Source 
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-07-28 13:09:08 +02:00
committed by GitHub
parent ecde94567c
commit a0798a3d82
6 changed files with 103 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/cache.rs
+87 -14
View File
@@ -297,20 +297,54 @@ impl std::fmt::Debug for FlowData {
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
/// Top-level fields of a stored flow value that [`FlowValue`] does not model, so parsing a
/// flow into a `FlowValue` drops them. Every write-back that round-trips a stored flow
/// through `FlowValue` must capture them first and re-attach them with
/// [`FlowExtras::reattach`], or they are destroyed on save. Adding a display-only flow
/// field means adding it here — this is the only list, and `FlowValue` must never model a
/// field named here: `reattach` flattens the two together, so a name in both would be
/// emitted twice and the value would no longer deserialize.
#[derive(Debug, Clone, Deserialize, Serialize, Default)]
pub struct FlowExtras {
pub notes: Option<Box<RawValue>>,
pub groups: Option<Box<RawValue>>,
}
impl FlowExtras {
/// Serialize `flow` with these extras folded back in. Fallible on purpose: the result is
/// written straight over a deployed flow value, so a serialization failure must abort the
/// write rather than persist a truncated value.
pub fn reattach(&self, flow: &FlowValue) -> error::Result<Box<RawValue>> {
// `flatten` + `RawValue` is fine for serialization; only deserialization breaks.
#[derive(Serialize)]
struct FlowValueWithExtras<'a> {
#[serde(flatten)]
flow: &'a FlowValue,
#[serde(skip_serializing_if = "Option::is_none")]
notes: Option<&'a Box<RawValue>>,
#[serde(skip_serializing_if = "Option::is_none")]
groups: Option<&'a Box<RawValue>>,
}
serde_json::value::to_raw_value(&FlowValueWithExtras {
flow,
notes: self.notes.as_ref(),
groups: self.groups.as_ref(),
})
.map_err(|e| error::Error::internal_err(format!("Failed to serialize flow value: {e}")))
}
/// Capture the extras carried by a raw stored flow value.
pub fn capture(raw_flow: &RawValue) -> Self {
serde_json::from_str::<FlowExtras>(raw_flow.get())
.map_err(|e| tracing::warn!("Failed to parse flow extras: {e}"))
.unwrap_or_default()
}
}
impl FlowData {
pub fn extras(&self) -> Option<FlowExtras> {
serde_json::from_str::<FlowExtras>(self.raw_flow.get())
.map_err(|e| {
tracing::error!("Failed to parse flow extras: {}", e);
error::Error::internal_err(format!("Failed to parse flow extras: {}", e))
})
.ok()
pub fn extras(&self) -> FlowExtras {
FlowExtras::capture(&self.raw_flow)
}
}
impl FlowData {
@@ -1294,7 +1328,7 @@ mod tests {
assert!(data.value().modules.is_empty());
// But extras() recovers them from the raw JSON
let extras = data.extras().expect("extras should parse");
let extras = data.extras();
let notes: serde_json::Value =
serde_json::from_str(extras.notes.expect("notes present").get()).unwrap();
assert_eq!(notes.as_array().unwrap().len(), 1);
@@ -1312,9 +1346,7 @@ mod tests {
let raw = serde_json::value::to_raw_value(&json!({"modules": []})).unwrap();
let data = FlowData::from_raw(raw).unwrap();
let extras = data
.extras()
.expect("extras should parse even without notes/groups");
let extras = data.extras();
assert!(extras.notes.is_none());
assert!(extras.groups.is_none());
}
@@ -1337,10 +1369,51 @@ mod tests {
let data2 = FlowData::from_raw(stripped_raw).unwrap();
// Notes are gone after the FlowValue round-trip
let extras = data2.extras().expect("extras should parse");
assert!(
extras.notes.is_none(),
data2.extras().notes.is_none(),
"notes lost after FlowValue round-trip"
);
}
#[test]
fn flow_extras_reattach_restores_what_the_roundtrip_drops() {
// Every write-back that re-serializes a stored flow through FlowValue must go
// through reattach, or notes/groups are destroyed.
let raw = serde_json::value::to_raw_value(&json!({
"modules": [{
"id": "a",
"value": {"type": "rawscript", "content": "x", "language": "bun",
"input_transforms": {}}
}],
"same_worker": true,
"notes": [{"id": "n1", "text": "t", "color": "blue", "type": "free"}],
"groups": [{"start_id": "a", "end_id": "b", "summary": "grp"}]
}))
.unwrap();
let data = FlowData::from_raw(raw.clone()).unwrap();
let reattached: serde_json::Value =
serde_json::from_str(data.extras().reattach(data.value()).unwrap().get()).unwrap();
let original: serde_json::Value = serde_json::from_str(raw.get()).unwrap();
for key in original.as_object().unwrap().keys() {
assert_eq!(
reattached.get(key),
original.get(key),
"{key} did not survive the FlowValue round-trip"
);
}
// Absent extras must stay absent rather than serialize as null, which would show
// up as a spurious change in flow diffs.
let without =
FlowData::from_raw(serde_json::value::to_raw_value(&json!({ "modules": [] })).unwrap())
.unwrap();
let output = without
.extras()
.reattach(without.value())
.unwrap()
.to_string();
assert!(!output.contains("notes") && !output.contains("groups"));
}
}
backend/windmill-common/src/flows.rs
+2 -61
View File
@@ -10,7 +10,6 @@ pub use windmill_types::flows::*;
use anyhow::Context;
use serde::Deserialize;
use serde::Serialize;
use sqlx::types::Json;
use sqlx::types::JsonRawValue;
@@ -99,18 +98,6 @@ pub async fn get_full_hub_flow_by_path(
.flow)
}
/// Serialize-only wrapper that combines resolved FlowValue with display-only extras.
/// flatten + RawValue is fine for serialization (only deserialization breaks).
#[derive(Serialize)]
struct FlowValueWithExtras<'a> {
#[serde(flatten)]
flow: &'a FlowValue,
#[serde(skip_serializing_if = "Option::is_none")]
notes: Option<&'a Box<JsonRawValue>>,
#[serde(skip_serializing_if = "Option::is_none")]
groups: Option<&'a Box<JsonRawValue>>,
}
/// Resolve the value of a flow if any.
pub async fn resolve_maybe_value<T>(
e: &sqlx::PgPool,
@@ -130,17 +117,13 @@ pub async fn resolve_maybe_value<T>(
}
/// Resolve modules recursively.
/// Stashes display-only fields (notes, groups) before the FlowValue round-trip
/// and re-injects them after, since FlowValue doesn't carry them.
async fn resolve_value_for_api(
e: &sqlx::PgPool,
workspace_id: &str,
value: &mut Box<JsonRawValue>,
with_code: bool,
) -> Result<(), Error> {
let extras = serde_json::from_str::<FlowExtras>(value.get())
.map_err(|e| tracing::warn!("Failed to parse flow extras: {e}"))
.ok();
let extras = FlowExtras::capture(value);
let mut val = serde_json::from_str::<FlowValue>(value.get()).map_err(|err| {
Error::internal_err(format!("resolve: Failed to parse flow value: {}", err))
@@ -149,12 +132,7 @@ async fn resolve_value_for_api(
resolve_module(e, workspace_id, &mut module.value, with_code).await?;
}
let extras = extras.unwrap_or(FlowExtras { notes: None, groups: None });
*value = to_raw_value(&FlowValueWithExtras {
flow: &val,
notes: extras.notes.as_ref(),
groups: extras.groups.as_ref(),
});
*value = extras.reattach(&val)?;
Ok(())
}
@@ -271,43 +249,6 @@ pub async fn resolve_modules(
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn flow_value_with_extras_serializes_notes_and_groups() {
let input = json!({
"modules": [],
"notes": [{"id": "n1", "text": "hello", "color": "yellow", "type": "free"}],
"groups": [{"start_id": "a", "end_id": "b", "summary": "grp"}]
});
let input_str = serde_json::to_string(&input).unwrap();
// Parse FlowValue (drops notes/groups) and FlowExtras (captures them)
let val: FlowValue = serde_json::from_str(&input_str).unwrap();
let extras: FlowExtras = serde_json::from_str(&input_str).unwrap();
// Serialize via FlowValueWithExtras — should include both
let combined = FlowValueWithExtras {
flow: &val,
notes: extras.notes.as_ref(),
groups: extras.groups.as_ref(),
};
let output: serde_json::Value =
serde_json::from_str(&serde_json::to_string(&combined).unwrap()).unwrap();
assert_eq!(output["notes"], input["notes"]);
assert_eq!(output["groups"], input["groups"]);
assert!(output["modules"].is_array());
}
#[test]
fn flow_value_with_extras_omits_none_extras() {
let val: FlowValue = serde_json::from_str(r#"{"modules":[]}"#).unwrap();
let combined = FlowValueWithExtras { flow: &val, notes: None, groups: None };
let output = serde_json::to_string(&combined).unwrap();
assert!(!output.contains("notes"));
assert!(!output.contains("groups"));
}
#[test]
fn extract_hub_flow_id_accepts_id_only_paths() {
backend/windmill-worker/src/worker_lockfiles.rs
+3 -30
View File
@@ -6,7 +6,6 @@ use std::fs::{create_dir_all, remove_dir_all};
use crate::ansible_executor::{get_git_repos_lock, AnsibleDependencyLocks};
use async_recursion::async_recursion;
use itertools::Itertools;
use serde::Serialize;
use serde_json::value::RawValue;
use serde_json::{from_value, json, Value};
use sha2::Digest;
@@ -33,7 +32,7 @@ use windmill_parser_yaml::AnsibleRequirements;
use windmill_common::{
apps::AppScriptId,
cache::{self, RawData},
error::{self, to_anyhow},
error,
flows::{add_virtual_items_if_necessary, FlowValue},
scripts::ScriptLang,
DB,
@@ -647,26 +646,7 @@ pub async fn handle_flow_dependency_job(
.await?;
}
#[derive(Debug, Clone, Serialize)]
struct FlowValueWithExtras<'a> {
#[serde(flatten)]
value: &'a FlowValue,
#[serde(skip_serializing_if = "Option::is_none")]
notes: Option<Box<RawValue>>,
#[serde(skip_serializing_if = "Option::is_none")]
groups: Option<Box<RawValue>>,
}
let new_flow_value = Json(
serde_json::value::to_raw_value(&FlowValueWithExtras {
value: &flow,
notes: extras.as_ref().and_then(|e| e.notes.clone()),
groups: extras.as_ref().and_then(|e| e.groups.clone()),
})
.map_err(to_anyhow)?,
);
let new_flow_value = Json(extras.reattach(&flow)?);
// Re-check cancellation to ensure we don't accidentally override a flow.
if sqlx::query_scalar!(
@@ -758,14 +738,7 @@ pub async fn handle_flow_dependency_job(
)
.await?;
let value_lite_with_extras = Json(
serde_json::value::to_raw_value(&FlowValueWithExtras {
value: &value_lite,
notes: extras.as_ref().and_then(|e| e.notes.clone()),
groups: extras.as_ref().and_then(|e| e.groups.clone()),
})
.map_err(to_anyhow)?,
);
let value_lite_with_extras = Json(extras.reattach(&value_lite)?);
sqlx::query!(
"INSERT INTO flow_version_lite (id, value) VALUES ($1, $2)
ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value",
frontend/src/lib/components/copilot/chat/flow/core.ts
+3 -1
View File
@@ -1164,6 +1164,7 @@ AI agents can use tools to accomplish tasks. When creating an AI agent module:
{
id: "search_docs",
summary: "Search_documentation",
description: "Search the product documentation. Use it whenever the user asks how a feature works.",
value: {
tool_type: "flowmodule",
type: "rawscript",
@@ -1178,7 +1179,8 @@ AI agents can use tools to accomplish tasks. When creating an AI agent module:
\`\`\`
- **Tool IDs**: Cannot contain spaces - use underscores
- **Tool summaries**: Cannot contain spaces - use underscores
- **Tool summaries**: Cannot contain spaces - use underscores. This is the tool *name* the agent sees
- **Tool descriptions**: Optional free text telling the agent when and how to call the tool. Set it whenever the name alone does not make that obvious - it overrides the description derived from the underlying script
- **Tool types**: \`flowmodule\` for scripts/flows, \`mcp\` for MCP server tools
### Contexts
frontend/src/lib/components/flows/agentToolUtils.ts
+6 -3
View File
@@ -122,10 +122,13 @@ export function agentToolToFlowModule(tool: FlowModuleTool): FlowModule {
}
/**
* Convert a FlowModule back to an AgentTool
* Used when saving changes back to the AI Agent tools array
* Wrap a newly created FlowModule as an AgentTool.
*
* Only valid for a module that is not already a tool: FlowModule carries none of the
* AgentTool-level metadata (`description`), so folding an edited module back into an
* existing tool through here would drop it — spread over the existing tool instead.
*/
export function flowModuleToAgentTool(flowModule: FlowModule): AgentTool {
export function newFlowModuleAgentTool(flowModule: FlowModule): AgentTool {
return {
id: flowModule.id,
summary: flowModule.summary,
frontend/src/lib/components/flows/map/FlowModuleSchemaMap.svelte
+2 -3
View File
@@ -45,7 +45,7 @@
import {
type AgentTool,
type SpecialToolKind,
flowModuleToAgentTool,
newFlowModuleAgentTool,
createMcpTool,
createWebsearchTool,
createAiAgentTool,
@@ -250,8 +250,7 @@
;(modules as AgentTool[]).splice(index, 0, aiAgentTool)
return modules as AgentTool[]
} else if (toolKind === 'flowmoduleTool') {
// Create AgentTool from FlowModule
const agentTool = flowModuleToAgentTool(module)
const agentTool = newFlowModuleAgentTool(module)
;(modules as AgentTool[]).splice(index, 0, agentTool)
return modules as AgentTool[]
} else {